update_room() - Code Metrics - Inspection of "Refactored tornado.py to a separate module" - Deathangel908/pychat - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7365f5...41cb1b )

by Andrew

created 2017-07-02 18:48 UTC

update_room() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	5
rs	9.4285

import base64
import json
import logging
import re
import sys
from io import BytesIO

import requests
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.core.files.uploadedfile import InMemoryUploadedFile
from django.core.mail import send_mail
from django.core.validators import validate_email
from django.db import connection, OperationalError, InterfaceError
from django.template import RequestContext
from django.template.loader import render_to_string
from django.utils.safestring import mark_safe

from chat import local
from chat import settings
from chat.models import Room
from chat.models import User
from chat.models import UserProfile, Verification, RoomUsers, IpAddress
from chat.py2_3 import urlopen
from chat.settings import ISSUES_REPORT_LINK, SITE_PROTOCOL, ALL_ROOM_ID
from chat.tornado.constants import RedisPrefix
from chat.tornado.constants import VarNames

USERNAME_REGEX = str(settings.MAX_USERNAME_LENGTH).join(['^[a-zA-Z-_0-9]{1,', '}$'])
API_URL = getattr(settings, "IP_API_URL", None)
RECAPTCHA_SECRET_KEY = getattr(settings, "RECAPTCHA_SECRET_KEY", None)
GOOGLE_OAUTH_2_CLIENT_ID = getattr(settings, "GOOGLE_OAUTH_2_CLIENT_ID", None)
GOOGLE_OAUTH_2_HOST = getattr(settings, "GOOGLE_OAUTH_2_HOST", None)
FACEBOOK_ACCESS_TOKEN = getattr(settings, "FACEBOOK_ACCESS_TOKEN", None)

logger = logging.getLogger(__name__)


def is_blank(check_str):
	if check_str and check_str.strip():
		return False
	else:
		return True


def get_users_in_current_user_rooms(user_id):
	user_rooms = Room.objects.filter(users__id=user_id, disabled=False).values('id', 'name')
	res = {room['id']: {
			VarNames.ROOM_NAME: room['name'],
			VarNames.ROOM_USERS: {}
		} for room in user_rooms}
	room_ids = (room_id for room_id in res)
	rooms_users = User.objects.filter(rooms__in=room_ids).values('id', 'username', 'sex', 'rooms__id')
	for user in rooms_users:
		RedisPrefix.set_js_user_structure(
			res[user['rooms__id']][VarNames.ROOM_USERS],
			user['id'],
			user['username'],
			user['sex']
		)
	return res


def update_room(room_id, disabled):
	if not disabled:
		raise ValidationError('This room already exist')
	else:
		Room.objects.filter(id=room_id).update(disabled=False)


def do_db(callback, *args, **kwargs):
	try:
		return callback(*args, **kwargs)
	except (OperationalError, InterfaceError) as e:
		if 'MySQL server has gone away' in str(e):
			logger.warning('%s, reconnecting' % e)
			connection.close()
			return callback(*args, **kwargs)
		else:
			raise e


def execute_query(query, *args, **kwargs):
	cursor = connection.cursor()
	cursor.execute(query, *args, **kwargs)
	desc = cursor.description
	return [
		dict(zip([col[0] for col in desc], row))
		for row in cursor.fetchall()
	]


def hide_fields(post, fields, huge=False, fill_with='****'):
	"""
	:param post: Object that will be copied
	:type post: QueryDict
	:param fields: fields that will be removed
	:param fill_with: replace characters for hidden fields
	:param huge: if true object will be cloned and then fields will be removed
	:return: a shallow copy of dictionary without specified fields
	"""
	if not huge:
		# hide *fields in shallow copy
		res = post.copy()
		for field in fields:
			if field in post:  # check if field was absent
				res[field] = fill_with
	else:
		# copy everything but *fields
		res = {}
		for field in post:
			# _______________________if this is field to remove
			res[field] = post[field] if field not in fields else fill_with
	return res


def check_password(password):
	"""
	Checks if password is secure
	:raises ValidationError exception if password is not valid
	"""
	if is_blank(password):
		raise ValidationError("password can't be empty")
	if not re.match(u'^\S.+\S$', password):
		raise ValidationError("password should be at least 3 symbols")


def check_email(email):
	"""
	:raises ValidationError if specified email is registered or not valid
	"""
	if not email:
		return
	try:
		validate_email(email)
		# theoretically can throw returning 'more than 1' error
		UserProfile.objects.get(email=email)
		raise ValidationError('Email {} is already used'.format(email))
	except User.DoesNotExist:
		pass


def check_user(username):
	"""
	Checks if specified username is free to register
	:type username str
	:raises ValidationError exception if username is not valid
	"""
	# Skip javascript validation, only summary message
	if is_blank(username):
		raise ValidationError("Username can't be empty")
	if not re.match(USERNAME_REGEX, username):
		raise ValidationError("Username {} doesn't match regex {}".format(username, USERNAME_REGEX))
	try:
		# theoretically can throw returning 'more than 1' error
		User.objects.get(username=username)
		raise ValidationError("Username {} is already used. Please select another one".format(username))
	except User.DoesNotExist:
		pass


def get_client_ip(request):
	x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
	return x_forwarded_for.split(',')[-1].strip() if x_forwarded_for else request.META.get('REMOTE_ADDR')


def check_captcha(request):
	"""
	:type request: WSGIRequest
	:raises ValidationError: if captcha is not valid or not set
	If RECAPTCHA_SECRET_KEY is enabled in settings validates request with it
	"""
	if not RECAPTCHA_SECRET_KEY:
		logger.debug('Skipping captcha validation')
		return
	try:
		captcha_rs = request.POST.get('g-recaptcha-response')
		url = "https://www.google.com/recaptcha/api/siteverify"
		params = {
			'secret': RECAPTCHA_SECRET_KEY,
			'response': captcha_rs,
			'remoteip': local.client_ip
		}
		raw_response = requests.post(url, params=params, verify=True)
		response = raw_response.json()
		if not response.get('success', False):
			logger.debug('Captcha is NOT valid, response: %s', raw_response)
			raise ValidationError(
				response['error-codes'] if response.get('error-codes', None) else 'This captcha already used')
		logger.debug('Captcha is valid, response: %s', raw_response)
	except Exception as e:
		raise ValidationError('Unable to check captcha because {}'.format(e))


def send_sign_up_email(user, site_address, request):
	if user.email is not None:
		verification = Verification(user=user, type_enum=Verification.TypeChoices.register)
		verification.save()
		user.email_verification = verification
		user.save(update_fields=['email_verification'])
		link = "{}://{}/confirm_email?token={}".format(SITE_PROTOCOL, site_address, verification.token)
		text = ('Hi {}, you have registered pychat'
				  '\nTo complete your registration please click on the url bellow: {}'
				  '\n\nIf you find any bugs or propositions you can post them {}').format(
			user.username, link, ISSUES_REPORT_LINK)
		start_message = mark_safe((
			"You have registered in <b>Pychat</b>. If you find any bugs or propositions you can post them"
			" <a href='{}'>here</a>. To complete your registration please click on the link below.").format(
				ISSUES_REPORT_LINK))
		context = {
			'username': user.username,
			'magicLink': link,
			'btnText': "CONFIRM SIGN UP",
			'greetings': start_message
		}
		html_message = render_to_string('sign_up_email.html', context, context_instance=RequestContext(request))
		logger.info('Sending verification email to userId %s (email %s)', user.id, user.email)
		send_mail("Confirm chat registration", text, site_address, [user.email, ], html_message=html_message)
		logger.info('Email %s has been sent', user.email)


def send_reset_password_email(request, user_profile, verification):
	link = "{}://{}/restore_password?token={}".format(SITE_PROTOCOL, request.get_host(), verification.token)
	message = "{},\n" \
				 "You requested to change a password on site {}.\n" \
				 "To proceed click on the link {}\n" \
				 "If you didn't request the password change just ignore this mail" \
		.format(user_profile.username, request.get_host(), link)
	ip_info = get_or_create_ip(get_client_ip(request), logger)
	start_message = mark_safe(
		"You have requested to send you a magic link to quickly restore password to <b>Pychat</b>. "
		"If it wasn't you, you can safely ignore this email")
	context = {
		'username': user_profile.username,
		'magicLink': link,
		'ipInfo': ip_info.info,
		'ip': ip_info.ip,
		'btnText': "CHANGE PASSWORD",
		'timeCreated': verification.time,
		'greetings': start_message
	}
	html_message = render_to_string('reset_pass_email.html', context, context_instance=RequestContext(request))
	send_mail("Pychat: restore password", message, request.get_host(), (user_profile.email,), fail_silently=False,
				 html_message=html_message)


def extract_photo(image_base64, filename=None):
	base64_type_data = re.search(r'data:(\w+/(\w+));base64,(.*)$', image_base64)
	logger.debug('Parsing base64 image')
	image_data = base64_type_data.group(3)
	f = BytesIO(base64.b64decode(image_data))
	content_type = base64_type_data.group(1)
	name = filename or ".{}".format(base64_type_data.group(2))
	logger.debug('Base64 filename extension %s, content_type %s', name, content_type)
	image = InMemoryUploadedFile(
		f,
		field_name='photo',
		name=name,
		content_type=content_type,
		size=sys.getsizeof(f),
		charset=None)
	return image


def get_max_key(dictionary):
	return max(dictionary.keys()) if dictionary else None


def create_user_model(user):
	user.save()
	RoomUsers(user_id=user.id, room_id=ALL_ROOM_ID).save()
	logger.info('Signed up new user %s, subscribed for channels with id %d', user, ALL_ROOM_ID)
	return user


def get_or_create_ip(ip, logger):
	"""

	@param ip: ip to fetch info from
	@param logger initialized logger:
	@type IpAddress
	"""
	try:
		ip_address = IpAddress.objects.get(ip=ip)
	except IpAddress.DoesNotExist:
		try:
			if not API_URL:
				raise Exception('api url is absent')
			logger.debug("Creating ip record %s", ip)
			f = urlopen(API_URL % ip)
			raw_response = f.read().decode("utf-8")
			response = json.loads(raw_response)
			if response['status'] != "success":
				raise Exception("Creating iprecord failed, server responded: %s" % raw_response)
			ip_address = IpAddress.objects.create(
				ip=ip,
				isp=response['isp'],
				country=response['country'],
				region=response['regionName'],
				city=response['city'],
				country_code=response['countryCode']
			)
		except Exception as e:
			logger.error("Error while creating ip with country info, because %s", e)
			ip_address = IpAddress.objects.create(ip=ip)
	return ip_address


class EmailOrUsernameModelBackend(object):
	"""
	This is a ModelBacked that allows authentication with either a username or an email address.
	"""

	def authenticate(self, username=None, password=None):
		try:
			if '@' in username:
				user = UserProfile.objects.get(email=username)
			else:
				user = UserProfile.objects.get(username=username)
			if user.check_password(password):
				return user
		except User.DoesNotExist:
			return None

	def get_user(self, username):
		try:
			return get_user_model().objects.get(pk=username)
		except get_user_model().DoesNotExist:
			return None


1			import base64
2			import json
3			import logging
4			import re
5			import sys
6			from io import BytesIO
7
8			import requests
9			from django.contrib.auth import get_user_model
10			from django.core.exceptions import ValidationError
11			from django.core.files.uploadedfile import InMemoryUploadedFile
12			from django.core.mail import send_mail
13			from django.core.validators import validate_email
14			from django.db import connection, OperationalError, InterfaceError
15			from django.template import RequestContext
16			from django.template.loader import render_to_string
17			from django.utils.safestring import mark_safe
18
19			from chat import local
20			from chat import settings
21			from chat.models import Room
22			from chat.models import User
23			from chat.models import UserProfile, Verification, RoomUsers, IpAddress
24			from chat.py2_3 import urlopen
25			from chat.settings import ISSUES_REPORT_LINK, SITE_PROTOCOL, ALL_ROOM_ID
26			from chat.tornado.constants import RedisPrefix
27			from chat.tornado.constants import VarNames
28
29			USERNAME_REGEX = str(settings.MAX_USERNAME_LENGTH).join(['^[a-zA-Z-_0-9]{1,', '}$'])
30			API_URL = getattr(settings, "IP_API_URL", None)
31			RECAPTCHA_SECRET_KEY = getattr(settings, "RECAPTCHA_SECRET_KEY", None)
32			GOOGLE_OAUTH_2_CLIENT_ID = getattr(settings, "GOOGLE_OAUTH_2_CLIENT_ID", None)
33			GOOGLE_OAUTH_2_HOST = getattr(settings, "GOOGLE_OAUTH_2_HOST", None)
34			FACEBOOK_ACCESS_TOKEN = getattr(settings, "FACEBOOK_ACCESS_TOKEN", None)
35
36			logger = logging.getLogger(__name__)
37
38
39			def is_blank(check_str):
40			if check_str and check_str.strip():
41			return False
42			else:
43			return True
44
45
46			def get_users_in_current_user_rooms(user_id):
47			user_rooms = Room.objects.filter(users__id=user_id, disabled=False).values('id', 'name')
48			res = {room['id']: {
49			VarNames.ROOM_NAME: room['name'],
50			VarNames.ROOM_USERS: {}
51			} for room in user_rooms}
52			room_ids = (room_id for room_id in res)
53			rooms_users = User.objects.filter(rooms__in=room_ids).values('id', 'username', 'sex', 'rooms__id')
54			for user in rooms_users:
55			RedisPrefix.set_js_user_structure(
56			res[user['rooms__id']][VarNames.ROOM_USERS],
57			user['id'],
58			user['username'],
59			user['sex']
60			)
61			return res
62
63
64			def update_room(room_id, disabled):
65			if not disabled:
66			raise ValidationError('This room already exist')
67			else:
68			Room.objects.filter(id=room_id).update(disabled=False)
69
70
71			def do_db(callback, args, *kwargs):
72			try:
73			return callback(args, *kwargs)
74			except (OperationalError, InterfaceError) as e:
75			if 'MySQL server has gone away' in str(e):
76			logger.warning('%s, reconnecting' % e)
77			connection.close()
78			return callback(args, *kwargs)
79			else:
80			raise e
81
82
83			def execute_query(query, args, *kwargs):
84			cursor = connection.cursor()
85			cursor.execute(query, args, *kwargs)
86			desc = cursor.description
87			return [
88			dict(zip([col[0] for col in desc], row))
89			for row in cursor.fetchall()
90			]
91
92
93			def hide_fields(post, fields, huge=False, fill_with='****'):
94			"""
95			:param post: Object that will be copied
96			:type post: QueryDict
97			:param fields: fields that will be removed
98			:param fill_with: replace characters for hidden fields
99			:param huge: if true object will be cloned and then fields will be removed
100			:return: a shallow copy of dictionary without specified fields
101			"""
102			if not huge:
103			# hide *fields in shallow copy
104			res = post.copy()
105			for field in fields:
106			if field in post: # check if field was absent
107			res[field] = fill_with
108			else:
109			# copy everything but *fields
110			res = {}
111			for field in post:
112			# _______________________if this is field to remove
113			res[field] = post[field] if field not in fields else fill_with
114			return res
115
116
117			def check_password(password):
118			"""
119			Checks if password is secure
120			:raises ValidationError exception if password is not valid
121			"""
122			if is_blank(password):
123			raise ValidationError("password can't be empty")
124			if not re.match(u'^\S.+\S$', password):
125			raise ValidationError("password should be at least 3 symbols")
126
127
128			def check_email(email):
129			"""
130			:raises ValidationError if specified email is registered or not valid
131			"""
132			if not email:
133			return
134			try:
135			validate_email(email)
136			# theoretically can throw returning 'more than 1' error
137			UserProfile.objects.get(email=email)
138			raise ValidationError('Email {} is already used'.format(email))
139			except User.DoesNotExist:
140			pass
141
142
143			def check_user(username):
144			"""
145			Checks if specified username is free to register
146			:type username str
147			:raises ValidationError exception if username is not valid
148			"""
149			# Skip javascript validation, only summary message
150			if is_blank(username):
151			raise ValidationError("Username can't be empty")
152			if not re.match(USERNAME_REGEX, username):
153			raise ValidationError("Username {} doesn't match regex {}".format(username, USERNAME_REGEX))
154			try:
155			# theoretically can throw returning 'more than 1' error
156			User.objects.get(username=username)
157			raise ValidationError("Username {} is already used. Please select another one".format(username))
158			except User.DoesNotExist:
159			pass
160
161
162			def get_client_ip(request):
163			x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
164			return x_forwarded_for.split(',')[-1].strip() if x_forwarded_for else request.META.get('REMOTE_ADDR')
165
166
167			def check_captcha(request):
168			"""
169			:type request: WSGIRequest
170			:raises ValidationError: if captcha is not valid or not set
171			If RECAPTCHA_SECRET_KEY is enabled in settings validates request with it
172			"""
173			if not RECAPTCHA_SECRET_KEY:
174			logger.debug('Skipping captcha validation')
175			return
176			try:
177			captcha_rs = request.POST.get('g-recaptcha-response')
178			url = "https://www.google.com/recaptcha/api/siteverify"
179			params = {
180			'secret': RECAPTCHA_SECRET_KEY,
181			'response': captcha_rs,
182			'remoteip': local.client_ip
183			}
184			raw_response = requests.post(url, params=params, verify=True)
185			response = raw_response.json()
186			if not response.get('success', False):
187			logger.debug('Captcha is NOT valid, response: %s', raw_response)
188			raise ValidationError(
189			response['error-codes'] if response.get('error-codes', None) else 'This captcha already used')
190			logger.debug('Captcha is valid, response: %s', raw_response)
191			except Exception as e:
192			raise ValidationError('Unable to check captcha because {}'.format(e))
193
194
195			def send_sign_up_email(user, site_address, request):
196			if user.email is not None:
197			verification = Verification(user=user, type_enum=Verification.TypeChoices.register)
198			verification.save()
199			user.email_verification = verification
200			user.save(update_fields=['email_verification'])
201			link = "{}://{}/confirm_email?token={}".format(SITE_PROTOCOL, site_address, verification.token)
202			text = ('Hi {}, you have registered pychat'
203			'\nTo complete your registration please click on the url bellow: {}'
204			'\n\nIf you find any bugs or propositions you can post them {}').format(
205			user.username, link, ISSUES_REPORT_LINK)
206			start_message = mark_safe((
207			"You have registered in <b>Pychat</b>. If you find any bugs or propositions you can post them"
208			" <a href='{}'>here</a>. To complete your registration please click on the link below.").format(
209			ISSUES_REPORT_LINK))
210			context = {
211			'username': user.username,
212			'magicLink': link,
213			'btnText': "CONFIRM SIGN UP",
214			'greetings': start_message
215			}
216			html_message = render_to_string('sign_up_email.html', context, context_instance=RequestContext(request))
217			logger.info('Sending verification email to userId %s (email %s)', user.id, user.email)
218			send_mail("Confirm chat registration", text, site_address, [user.email, ], html_message=html_message)
219			logger.info('Email %s has been sent', user.email)
220
221
222			def send_reset_password_email(request, user_profile, verification):
223			link = "{}://{}/restore_password?token={}".format(SITE_PROTOCOL, request.get_host(), verification.token)
224			message = "{},\n" \
225			"You requested to change a password on site {}.\n" \
226			"To proceed click on the link {}\n" \
227			"If you didn't request the password change just ignore this mail" \
228			.format(user_profile.username, request.get_host(), link)
229			ip_info = get_or_create_ip(get_client_ip(request), logger)
230			start_message = mark_safe(
231			"You have requested to send you a magic link to quickly restore password to <b>Pychat</b>. "
232			"If it wasn't you, you can safely ignore this email")
233			context = {
234			'username': user_profile.username,
235			'magicLink': link,
236			'ipInfo': ip_info.info,
237			'ip': ip_info.ip,
238			'btnText': "CHANGE PASSWORD",
239			'timeCreated': verification.time,
240			'greetings': start_message
241			}
242			html_message = render_to_string('reset_pass_email.html', context, context_instance=RequestContext(request))
243			send_mail("Pychat: restore password", message, request.get_host(), (user_profile.email,), fail_silently=False,
244			html_message=html_message)
245
246
247			def extract_photo(image_base64, filename=None):
248			base64_type_data = re.search(r'data:(\w+/(\w+));base64,(.*)$', image_base64)
249			logger.debug('Parsing base64 image')
250			image_data = base64_type_data.group(3)
251			f = BytesIO(base64.b64decode(image_data))
252			content_type = base64_type_data.group(1)
253			name = filename or ".{}".format(base64_type_data.group(2))
254			logger.debug('Base64 filename extension %s, content_type %s', name, content_type)
255			image = InMemoryUploadedFile(
256			f,
257			field_name='photo',
258			name=name,
259			content_type=content_type,
260			size=sys.getsizeof(f),
261			charset=None)
262			return image
263
264
265			def get_max_key(dictionary):
266			return max(dictionary.keys()) if dictionary else None
267
268
269			def create_user_model(user):
270			user.save()
271			RoomUsers(user_id=user.id, room_id=ALL_ROOM_ID).save()
272			logger.info('Signed up new user %s, subscribed for channels with id %d', user, ALL_ROOM_ID)
273			return user
274
275
276			def get_or_create_ip(ip, logger):
277			"""
278
279			@param ip: ip to fetch info from
280			@param logger initialized logger:
281			@type IpAddress
282			"""
283			try:
284			ip_address = IpAddress.objects.get(ip=ip)
285			except IpAddress.DoesNotExist:
286			try:
287			if not API_URL:
288			raise Exception('api url is absent')
289			logger.debug("Creating ip record %s", ip)
290			f = urlopen(API_URL % ip)
291			raw_response = f.read().decode("utf-8")
292			response = json.loads(raw_response)
293			if response['status'] != "success":
294			raise Exception("Creating iprecord failed, server responded: %s" % raw_response)
295			ip_address = IpAddress.objects.create(
296			ip=ip,
297			isp=response['isp'],
298			country=response['country'],
299			region=response['regionName'],
300			city=response['city'],
301			country_code=response['countryCode']
302			)
303			except Exception as e:
304			logger.error("Error while creating ip with country info, because %s", e)
305			ip_address = IpAddress.objects.create(ip=ip)
306			return ip_address
307
308
309			class EmailOrUsernameModelBackend(object):
310			"""
311			This is a ModelBacked that allows authentication with either a username or an email address.
312			"""
313
314			def authenticate(self, username=None, password=None):
315			try:
316			if '@' in username:
317			user = UserProfile.objects.get(email=username)
318			else:
319			user = UserProfile.objects.get(username=username)
320			if user.check_password(password):
321			return user
322			except User.DoesNotExist:
323			return None
324
325			def get_user(self, username):
326			try:
327			return get_user_model().objects.get(pk=username)
328			except get_user_model().DoesNotExist:
329			return None
330

Deathangel908 / pychat

Push — master ( 7365f5...41cb1b )

update_room() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like