send_email_changed() - Code Metrics - Inspection of "Added missing templates to prev commit" - Deathangel908/pychat - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 5003b8...b307ec )

by Andrew

created 2018-04-08 22:31 UTC

send_email_changed() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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

import requests
from datetime import datetime
from django.conf import settings
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 IntegrityError
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.models import Image
from chat.models import Room, get_milliseconds
from chat.models import User
from chat.models import UserProfile, Verification, RoomUsers, IpAddress
from chat.py2_3 import urlopen
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,', '}$'])

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', 'roomusers__notifications', 'roomusers__volume')
	res = {room['id']: {
			VarNames.ROOM_NAME: room['name'],
			VarNames.NOTIFICATIONS: room['roomusers__notifications'],
			VarNames.VOLUME: room['roomusers__volume'],
			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:
		dict = res[user['rooms__id']][VarNames.ROOM_USERS]
		dict[user['id']] = RedisPrefix.set_js_user_structure(user['username'], user['sex'])
	return res


def get_or_create_room(channels, room_id, user_id):
	if room_id not in channels:
		raise ValidationError("Access denied, only allowed for channels {}".format(channels))
	room = do_db(Room.objects.get, id=room_id)
	if room.is_private:
		raise ValidationError("You can't add users to direct room, create a new room instead")
	try:
		Room.users.through.objects.create(room_id=room_id, user_id=user_id)
	except IntegrityError:
		raise ValidationError("User is already in channel")
	return room


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 create_room(self_user_id, user_id):
	# get all self private rooms ids
	user_rooms = evaluate(Room.users.through.objects.filter(user_id=self_user_id, room__name__isnull=True).values('room_id'))
	# get private room that contains another user from rooms above
	if user_rooms and self_user_id == user_id:
		room_id = create_self_room(self_user_id,user_rooms)
	elif user_rooms:
		room_id = create_other_room(self_user_id, user_id, user_rooms)
	else:
		room_id = create_other_room_wo_check(self_user_id, user_id)
	return room_id


def create_other_room_wo_check(self_user_id, user_id):
	room = Room()
	room.save()
	room_id = room.id
	if self_user_id == user_id:
		RoomUsers(user_id=user_id, room_id=room_id).save()
	else:
		RoomUsers.objects.bulk_create([
			RoomUsers(user_id=self_user_id, room_id=room_id),
			RoomUsers(user_id=user_id, room_id=room_id),
		])
	return room_id


def create_other_room(self_user_id, user_id, user_rooms):
	rooms_query = RoomUsers.objects.filter(user_id=user_id, room__in=user_rooms)
	query = rooms_query.values('room__id', 'room__disabled')
	try:
		room = do_db(query.get)
		room_id = room['room__id']
		update_room(room_id, room['room__disabled'])
	except RoomUsers.DoesNotExist:
		room = Room()
		room.save()
		room_id = room.id
		RoomUsers.objects.bulk_create([
			RoomUsers(user_id=self_user_id, room_id=room_id),
			RoomUsers(user_id=user_id, room_id=room_id),
		])
	return room_id


def evaluate(query_set):
	do_db(len, query_set)
	return query_set


def create_self_room(self_user_id, user_rooms):
	room_ids = list([room['room_id'] for room in evaluate(user_rooms)])
	query = execute_query(settings.SELECT_SELF_ROOM, [room_ids, ])
	if query:
		room_id = query[0]['room__id']
		update_room(room_id, query[0]['room__disabled'])
	else:
		room = Room()
		room.save()
		room_id = room.id
		RoomUsers(user_id=self_user_id, room_id=room_id, notifications=False).save()
	return room_id

def validate_edit_message(self_id, message):
	if message.sender_id != self_id:
		raise ValidationError("You can only edit your messages")
	if message.time + 600000 < get_milliseconds():
		raise ValidationError("You can only edit messages that were send not more than 10 min ago")
	if message.deleted:
		raise ValidationError("Already deleted")


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
	validate_user(username)
	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 validate_user(username):
	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))


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 hasattr(settings, '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': settings.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(settings.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, settings.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(
				settings.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)
		try:
			send_mail("Confirm chat registration", text, site_address, [user.email, ], html_message=html_message, fail_silently=True)
		except Exception as e:
			logging.error("Failed to send registration email because {}".format(e))
		else:
			logger.info('Email %s has been sent', user.email)


def send_reset_password_email(request, user_profile, verification):
	link = "{}://{}/restore_password?token={}".format(settings.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 send_password_changed(request, email):
	message = "Password has been changed for user {}".format(request.user.username)
	ip_info = get_or_create_ip(get_client_ip(request), logger)
	context = {
		'username': request.user.username,
		'ipInfo': ip_info.info,
		'ip': ip_info.ip,
		'timeCreated': datetime.now(),
	}
	html_message = render_to_string('change_password.html', context, context_instance=RequestContext(request))
	send_mail("Pychat: password change", message, request.get_host(), (email,), fail_silently=False,
				 html_message=html_message)


def send_email_changed(request, old_email, new_email):
	message = "Dmail been changed for user {}".format(request.user.username)
	ip_info = get_or_create_ip(get_client_ip(request), logger)
	context = {
		'username': request.user.username,
		'ipInfo': ip_info.info,
		'ip': ip_info.ip,
		'timeCreated': datetime.now(),
		'email': new_email,
	}
	html_message = render_to_string('change_email.html', context, context_instance=RequestContext(request))
	send_mail("Pychat: email change", message, request.get_host(), (old_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 create_user_model(user):
	user.save()
	RoomUsers(user_id=user.id, room_id=settings.ALL_ROOM_ID, notifications=False).save()
	logger.info('Signed up new user %s, subscribed for channels with id %d', user, settings.ALL_ROOM_ID)
	return user


def create_ip_structure(ip, raw_response):
	response = json.loads(raw_response)
	if response['status'] != "success":
		raise Exception("Creating iprecord failed, server responded: %s" % raw_response)
	return IpAddress.objects.create(
		ip=ip,
		isp=response['isp'],
		country=response['country'],
		region=response['regionName'],
		city=response['city'],
		country_code=response['countryCode']
	)


def get_or_create_ip(ip, logger):
	def create_ip():
		f = urlopen(settings.IP_API_URL % ip)
		decode = f.read().decode("utf-8")
		return create_ip_structure(ip, decode)
	return get_or_create_ip_wrapper(ip, logger, create_ip)


def get_or_create_ip_wrapper(ip, logger, fetcher_ip_function):
	"""
	@param ip: ip to fetch info from
	@param logger initialized logger:
	@type IpAddress
	"""
	try:
		return IpAddress.objects.get(ip=ip)
	except IpAddress.DoesNotExist:
		try:
			if not hasattr(settings, 'IP_API_URL'):
				raise Exception("IP_API_URL aint set")
			return fetcher_ip_function()
		except Exception as e:
			logger.error("Error while creating ip with country info, because %s", e)
			return IpAddress.objects.create(ip=ip)


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


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


def process_images(images, message):
	if images:
		if message.symbol:
			replace_symbols_if_needed(images, message)
		new_symbol = get_max_key(images)
		if message.symbol is None or new_symbol > message.symbol:
			message.symbol = new_symbol
	db_images = save_images(images, message.id)
	if message.symbol:  # fetch all, including that we just store
		db_images = Image.objects.filter(message_id=message.id)
	return prepare_img(db_images, message.id)


def save_images(images, message_id):
	db_images = []
	if images:
		db_images = [Image(
			message_id=message_id,
			img=extract_photo(
				images[k][VarNames.IMG_B64],
				images[k][VarNames.IMG_FILE_NAME]
			),
			symbol=k) for k in images]
		Image.objects.bulk_create(db_images)
	return db_images


def replace_symbols_if_needed(images, message):
	# if message was edited user wasn't notified about that and he edits message again
	# his symbol can go out of sync
	order = ord(message.symbol)
	new_dict = []
	for img in images:
		if img <= message.symbol:
			order += 1
			new_symb = chr(order)
			new_dict.append({
				'new': new_symb,
				'old': img,
				'value': images[img]
			})
			message.content = message.content.replace(img, new_symb)
	for d in new_dict:  # dictionary changed size during iteration
		del images[d['old']]
		images[d['new']] = d['value']


def prepare_img(images, message_id):
	"""
	:type message_id: int
	:type images: list[chat.models.Image]
	"""
	if images:
		return {x.symbol: x.img.url for x in images if x.message_id == message_id}


def get_message_images(messages):
	ids = [message.id for message in messages if message.symbol]
	if ids:
		images = Image.objects.filter(message_id__in=ids)
	else:
		images = []
	return images

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 datetime import datetime
10			from django.conf import settings
11			from django.contrib.auth import get_user_model
12			from django.core.exceptions import ValidationError
13			from django.core.files.uploadedfile import InMemoryUploadedFile
14			from django.core.mail import send_mail
15			from django.core.validators import validate_email
16			from django.db import IntegrityError
17			from django.db import connection, OperationalError, InterfaceError
18			from django.template import RequestContext
19			from django.template.loader import render_to_string
20			from django.utils.safestring import mark_safe
21
22			from chat import local
23			from chat.models import Image
24			from chat.models import Room, get_milliseconds
25			from chat.models import User
26			from chat.models import UserProfile, Verification, RoomUsers, IpAddress
27			from chat.py2_3 import urlopen
28			from chat.tornado.constants import RedisPrefix
29			from chat.tornado.constants import VarNames
30
31			USERNAME_REGEX = str(settings.MAX_USERNAME_LENGTH).join(['^[a-zA-Z-_0-9]{1,', '}$'])
32
33			logger = logging.getLogger(__name__)
34
35
36			def is_blank(check_str):
37			if check_str and check_str.strip():
38			return False
39			else:
40			return True
41
42
43			def get_users_in_current_user_rooms(user_id):
44			user_rooms = Room.objects.filter(users__id=user_id, disabled=False).values('id', 'name', 'roomusers__notifications', 'roomusers__volume')
45			res = {room['id']: {
46			VarNames.ROOM_NAME: room['name'],
47			VarNames.NOTIFICATIONS: room['roomusers__notifications'],
48			VarNames.VOLUME: room['roomusers__volume'],
49			VarNames.ROOM_USERS: {}
50			} for room in user_rooms}
51			room_ids = (room_id for room_id in res)
52			rooms_users = User.objects.filter(rooms__in=room_ids).values('id', 'username', 'sex', 'rooms__id')
53			for user in rooms_users:
54			dict = res[user['rooms__id']][VarNames.ROOM_USERS]
55			dict[user['id']] = RedisPrefix.set_js_user_structure(user['username'], user['sex'])
56			return res
57
58
59			def get_or_create_room(channels, room_id, user_id):
60			if room_id not in channels:
61			raise ValidationError("Access denied, only allowed for channels {}".format(channels))
62			room = do_db(Room.objects.get, id=room_id)
63			if room.is_private:
64			raise ValidationError("You can't add users to direct room, create a new room instead")
65			try:
66			Room.users.through.objects.create(room_id=room_id, user_id=user_id)
67			except IntegrityError:
68			raise ValidationError("User is already in channel")
69			return room
70
71
72			def update_room(room_id, disabled):
73			if not disabled:
74			raise ValidationError('This room already exist')
75			else:
76			Room.objects.filter(id=room_id).update(disabled=False)
77
78
79			def create_room(self_user_id, user_id):
80			# get all self private rooms ids
81			user_rooms = evaluate(Room.users.through.objects.filter(user_id=self_user_id, room__name__isnull=True).values('room_id'))
82			# get private room that contains another user from rooms above
83			if user_rooms and self_user_id == user_id:
84			room_id = create_self_room(self_user_id,user_rooms)
85			elif user_rooms:
86			room_id = create_other_room(self_user_id, user_id, user_rooms)
87			else:
88			room_id = create_other_room_wo_check(self_user_id, user_id)
89			return room_id
90
91
92			def create_other_room_wo_check(self_user_id, user_id):
93			room = Room()
94			room.save()
95			room_id = room.id
96			if self_user_id == user_id:
97			RoomUsers(user_id=user_id, room_id=room_id).save()
98			else:
99			RoomUsers.objects.bulk_create([
100			RoomUsers(user_id=self_user_id, room_id=room_id),
101			RoomUsers(user_id=user_id, room_id=room_id),
102			])
103			return room_id
104
105
106			def create_other_room(self_user_id, user_id, user_rooms):
107			rooms_query = RoomUsers.objects.filter(user_id=user_id, room__in=user_rooms)
108			query = rooms_query.values('room__id', 'room__disabled')
109			try:
110			room = do_db(query.get)
111			room_id = room['room__id']
112			update_room(room_id, room['room__disabled'])
113			except RoomUsers.DoesNotExist:
114			room = Room()
115			room.save()
116			room_id = room.id
117			RoomUsers.objects.bulk_create([
118			RoomUsers(user_id=self_user_id, room_id=room_id),
119			RoomUsers(user_id=user_id, room_id=room_id),
120			])
121			return room_id
122
123
124			def evaluate(query_set):
125			do_db(len, query_set)
126			return query_set
127
128
129			def create_self_room(self_user_id, user_rooms):
130			room_ids = list([room['room_id'] for room in evaluate(user_rooms)])
131			query = execute_query(settings.SELECT_SELF_ROOM, [room_ids, ])
132			if query:
133			room_id = query[0]['room__id']
134			update_room(room_id, query[0]['room__disabled'])
135			else:
136			room = Room()
137			room.save()
138			room_id = room.id
139			RoomUsers(user_id=self_user_id, room_id=room_id, notifications=False).save()
140			return room_id
141
142			def validate_edit_message(self_id, message):
143			if message.sender_id != self_id:
144			raise ValidationError("You can only edit your messages")
145			if message.time + 600000 < get_milliseconds():
146			raise ValidationError("You can only edit messages that were send not more than 10 min ago")
147			if message.deleted:
148			raise ValidationError("Already deleted")
149
150
151			def do_db(callback, args, *kwargs):
152			try:
153			return callback(args, *kwargs)
154			except (OperationalError, InterfaceError) as e:
155			if 'MySQL server has gone away' in str(e):
156			logger.warning('%s, reconnecting' % e)
157			connection.close()
158			return callback(args, *kwargs)
159			else:
160			raise e
161
162
163			def execute_query(query, args, *kwargs):
164			cursor = connection.cursor()
165			cursor.execute(query, args, *kwargs)
166			desc = cursor.description
167			return [
168			dict(zip([col[0] for col in desc], row))
169			for row in cursor.fetchall()
170			]
171
172
173			def hide_fields(post, fields, huge=False, fill_with='****'):
174			"""
175			:param post: Object that will be copied
176			:type post: QueryDict
177			:param fields: fields that will be removed
178			:param fill_with: replace characters for hidden fields
179			:param huge: if true object will be cloned and then fields will be removed
180			:return: a shallow copy of dictionary without specified fields
181			"""
182			if not huge:
183			# hide *fields in shallow copy
184			res = post.copy()
185			for field in fields:
186			if field in post: # check if field was absent
187			res[field] = fill_with
188			else:
189			# copy everything but *fields
190			res = {}
191			for field in post:
192			# _______________________if this is field to remove
193			res[field] = post[field] if field not in fields else fill_with
194			return res
195
196
197			def check_password(password):
198			"""
199			Checks if password is secure
200			:raises ValidationError exception if password is not valid
201			"""
202			if is_blank(password):
203			raise ValidationError("password can't be empty")
204			if not re.match(u'^\S.+\S$', password):
205			raise ValidationError("password should be at least 3 symbols")
206
207
208			def check_email(email):
209			"""
210			:raises ValidationError if specified email is registered or not valid
211			"""
212			if not email:
213			return
214			try:
215			validate_email(email)
216			# theoretically can throw returning 'more than 1' error
217			UserProfile.objects.get(email=email)
218			raise ValidationError('Email {} is already used'.format(email))
219			except User.DoesNotExist:
220			pass
221
222
223			def check_user(username):
224			"""
225			Checks if specified username is free to register
226			:type username str
227			:raises ValidationError exception if username is not valid
228			"""
229			# Skip javascript validation, only summary message
230			validate_user(username)
231			try:
232			# theoretically can throw returning 'more than 1' error
233			User.objects.get(username=username)
234			raise ValidationError("Username {} is already used. Please select another one".format(username))
235			except User.DoesNotExist:
236			pass
237
238
239			def validate_user(username):
240			if is_blank(username):
241			raise ValidationError("Username can't be empty")
242			if not re.match(USERNAME_REGEX, username):
243			raise ValidationError("Username {} doesn't match regex {}".format(username, USERNAME_REGEX))
244
245
246			def get_client_ip(request):
247			x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
248			return x_forwarded_for.split(',')[-1].strip() if x_forwarded_for else request.META.get('REMOTE_ADDR')
249
250
251			def check_captcha(request):
252			"""
253			:type request: WSGIRequest
254			:raises ValidationError: if captcha is not valid or not set
255			If RECAPTCHA_SECRET_KEY is enabled in settings validates request with it
256			"""
257			if not hasattr(settings, 'RECAPTCHA_SECRET_KEY'):
258			logger.debug('Skipping captcha validation')
259			return
260			try:
261			captcha_rs = request.POST.get('g-recaptcha-response')
262			url = "https://www.google.com/recaptcha/api/siteverify"
263			params = {
264			'secret': settings.RECAPTCHA_SECRET_KEY,
265			'response': captcha_rs,
266			'remoteip': local.client_ip
267			}
268			raw_response = requests.post(url, params=params, verify=True)
269			response = raw_response.json()
270			if not response.get('success', False):
271			logger.debug('Captcha is NOT valid, response: %s', raw_response)
272			raise ValidationError(
273			response['error-codes'] if response.get('error-codes', None) else 'This captcha already used')
274			logger.debug('Captcha is valid, response: %s', raw_response)
275			except Exception as e:
276			raise ValidationError('Unable to check captcha because {}'.format(e))
277
278
279			def send_sign_up_email(user, site_address, request):
280			if user.email is not None:
281			verification = Verification(user=user, type_enum=Verification.TypeChoices.register)
282			verification.save()
283			user.email_verification = verification
284			user.save(update_fields=['email_verification'])
285			link = "{}://{}/confirm_email?token={}".format(settings.SITE_PROTOCOL, site_address, verification.token)
286			text = ('Hi {}, you have registered pychat'
287			'\nTo complete your registration please click on the url bellow: {}'
288			'\n\nIf you find any bugs or propositions you can post them {}').format(
289			user.username, link, settings.ISSUES_REPORT_LINK)
290			start_message = mark_safe((
291			"You have registered in <b>Pychat</b>. If you find any bugs or propositions you can post them"
292			" <a href='{}'>here</a>. To complete your registration please click on the link below.").format(
293			settings.ISSUES_REPORT_LINK))
294			context = {
295			'username': user.username,
296			'magicLink': link,
297			'btnText': "CONFIRM SIGN UP",
298			'greetings': start_message
299			}
300			html_message = render_to_string('sign_up_email.html', context, context_instance=RequestContext(request))
301			logger.info('Sending verification email to userId %s (email %s)', user.id, user.email)
302			try:
303			send_mail("Confirm chat registration", text, site_address, [user.email, ], html_message=html_message, fail_silently=True)
304			except Exception as e:
305			logging.error("Failed to send registration email because {}".format(e))
306			else:
307			logger.info('Email %s has been sent', user.email)
308
309
310			def send_reset_password_email(request, user_profile, verification):
311			link = "{}://{}/restore_password?token={}".format(settings.SITE_PROTOCOL, request.get_host(), verification.token)
312			message = "{},\n" \
313			"You requested to change a password on site {}.\n" \
314			"To proceed click on the link {}\n" \
315			"If you didn't request the password change just ignore this mail" \
316			.format(user_profile.username, request.get_host(), link)
317			ip_info = get_or_create_ip(get_client_ip(request), logger)
318			start_message = mark_safe(
319			"You have requested to send you a magic link to quickly restore password to <b>Pychat</b>. "
320			"If it wasn't you, you can safely ignore this email")
321			context = {
322			'username': user_profile.username,
323			'magicLink': link,
324			'ipInfo': ip_info.info,
325			'ip': ip_info.ip,
326			'btnText': "CHANGE PASSWORD",
327			'timeCreated': verification.time,
328			'greetings': start_message
329			}
330			html_message = render_to_string('reset_pass_email.html', context, context_instance=RequestContext(request))
331			send_mail("Pychat: restore password", message, request.get_host(), (user_profile.email,), fail_silently=False,
332			html_message=html_message)
333
334
335			def send_password_changed(request, email):
336			message = "Password has been changed for user {}".format(request.user.username)
337			ip_info = get_or_create_ip(get_client_ip(request), logger)
338			context = {
339			'username': request.user.username,
340			'ipInfo': ip_info.info,
341			'ip': ip_info.ip,
342			'timeCreated': datetime.now(),
343			}
344			html_message = render_to_string('change_password.html', context, context_instance=RequestContext(request))
345			send_mail("Pychat: password change", message, request.get_host(), (email,), fail_silently=False,
346			html_message=html_message)
347
348
349			def send_email_changed(request, old_email, new_email):
350			message = "Dmail been changed for user {}".format(request.user.username)
351			ip_info = get_or_create_ip(get_client_ip(request), logger)
352			context = {
353			'username': request.user.username,
354			'ipInfo': ip_info.info,
355			'ip': ip_info.ip,
356			'timeCreated': datetime.now(),
357			'email': new_email,
358			}
359			html_message = render_to_string('change_email.html', context, context_instance=RequestContext(request))
360			send_mail("Pychat: email change", message, request.get_host(), (old_email,), fail_silently=False,
361			html_message=html_message)
362
363
364			def extract_photo(image_base64, filename=None):
365			base64_type_data = re.search(r'data:(\w+/(\w+));base64,(.*)$', image_base64)
366			logger.debug('Parsing base64 image')
367			image_data = base64_type_data.group(3)
368			f = BytesIO(base64.b64decode(image_data))
369			content_type = base64_type_data.group(1)
370			name = filename or ".{}".format(base64_type_data.group(2))
371			logger.debug('Base64 filename extension %s, content_type %s', name, content_type)
372			image = InMemoryUploadedFile(
373			f,
374			field_name='photo',
375			name=name,
376			content_type=content_type,
377			size=sys.getsizeof(f),
378			charset=None)
379			return image
380
381
382			def create_user_model(user):
383			user.save()
384			RoomUsers(user_id=user.id, room_id=settings.ALL_ROOM_ID, notifications=False).save()
385			logger.info('Signed up new user %s, subscribed for channels with id %d', user, settings.ALL_ROOM_ID)
386			return user
387
388
389			def create_ip_structure(ip, raw_response):
390			response = json.loads(raw_response)
391			if response['status'] != "success":
392			raise Exception("Creating iprecord failed, server responded: %s" % raw_response)
393			return IpAddress.objects.create(
394			ip=ip,
395			isp=response['isp'],
396			country=response['country'],
397			region=response['regionName'],
398			city=response['city'],
399			country_code=response['countryCode']
400			)
401
402
403			def get_or_create_ip(ip, logger):
404			def create_ip():
405			f = urlopen(settings.IP_API_URL % ip)
406			decode = f.read().decode("utf-8")
407			return create_ip_structure(ip, decode)
408			return get_or_create_ip_wrapper(ip, logger, create_ip)
409
410
411			def get_or_create_ip_wrapper(ip, logger, fetcher_ip_function):
412			"""
413			@param ip: ip to fetch info from
414			@param logger initialized logger:
415			@type IpAddress
416			"""
417			try:
418			return IpAddress.objects.get(ip=ip)
419			except IpAddress.DoesNotExist:
420			try:
421			if not hasattr(settings, 'IP_API_URL'):
422			raise Exception("IP_API_URL aint set")
423			return fetcher_ip_function()
424			except Exception as e:
425			logger.error("Error while creating ip with country info, because %s", e)
426			return IpAddress.objects.create(ip=ip)
427
428
429			class EmailOrUsernameModelBackend(object):
430			"""
431			This is a ModelBacked that allows authentication with either a username or an email address.
432			"""
433
434			def authenticate(self, username=None, password=None):
435			try:
436			if '@' in username:
437			user = UserProfile.objects.get(email=username)
438			else:
439			user = UserProfile.objects.get(username=username)
440			if user.check_password(password):
441			return user
442			except User.DoesNotExist:
443			return None
444
445			def get_user(self, username):
446			try:
447			return get_user_model().objects.get(pk=username)
448			except get_user_model().DoesNotExist:
449			return None
450
451
452			def get_max_key(dictionary):
453			return max(dictionary.keys()) if dictionary else None
454
455
456			def process_images(images, message):
457			if images:
458			if message.symbol:
459			replace_symbols_if_needed(images, message)
460			new_symbol = get_max_key(images)
461			if message.symbol is None or new_symbol > message.symbol:
462			message.symbol = new_symbol
463			db_images = save_images(images, message.id)
464			if message.symbol: # fetch all, including that we just store
465			db_images = Image.objects.filter(message_id=message.id)
466			return prepare_img(db_images, message.id)
467
468
469			def save_images(images, message_id):
470			db_images = []
471			if images:
472			db_images = [Image(
473			message_id=message_id,
474			img=extract_photo(
475			images[k][VarNames.IMG_B64],
476			images[k][VarNames.IMG_FILE_NAME]
477			),
478			symbol=k) for k in images]
479			Image.objects.bulk_create(db_images)
480			return db_images
481
482
483			def replace_symbols_if_needed(images, message):
484			# if message was edited user wasn't notified about that and he edits message again
485			# his symbol can go out of sync
486			order = ord(message.symbol)
487			new_dict = []
488			for img in images:
489			if img <= message.symbol:
490			order += 1
491			new_symb = chr(order)
492			new_dict.append({
493			'new': new_symb,
494			'old': img,
495			'value': images[img]
496			})
497			message.content = message.content.replace(img, new_symb)
498			for d in new_dict: # dictionary changed size during iteration
499			del images[d['old']]
500			images[d['new']] = d['value']
501
502
503			def prepare_img(images, message_id):
504			"""
505			:type message_id: int
506			:type images: list[chat.models.Image]
507			"""
508			if images:
509			return {x.symbol: x.img.url for x in images if x.message_id == message_id}
510
511
512			def get_message_images(messages):
513			ids = [message.id for message in messages if message.symbol]
514			if ids:
515			images = Image.objects.filter(message_id__in=ids)
516			else:
517			images = []
518			return images

Deathangel908 / pychat

Push — master ( 5003b8...b307ec )

send_email_changed() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like