utils.py
148 lines
| 4.0 KiB
| text/x-python
|
PythonLexer
/ boards / utils.py
|
r78 | """ | ||
This module contains helper functions and helper classes. | ||||
""" | ||||
|
r1305 | import hashlib | ||
|
r1368 | from random import random | ||
|
r716 | import time | ||
|
r853 | import hmac | ||
|
r1106 | |||
|
r957 | from django.core.cache import cache | ||
from django.db.models import Model | ||||
|
r1328 | from django import forms | ||
|
r542 | from django.utils import timezone | ||
|
r1328 | from django.utils.translation import ugettext_lazy as _ | ||
|
r1371 | import magic | ||
|
r1368 | from portage import os | ||
|
r1362 | |||
|
r1328 | import boards | ||
|
r1362 | from boards.settings import get_bool | ||
|
r81 | from neboard import settings | ||
|
r78 | |||
|
r1106 | CACHE_KEY_DELIMITER = '_' | ||
|
r1109 | PERMISSION_MODERATE = 'moderation' | ||
|
r95 | |||
|
r1362 | HTTP_FORWARDED = 'HTTP_X_FORWARDED_FOR' | ||
META_REMOTE_ADDR = 'REMOTE_ADDR' | ||||
SETTING_MESSAGES = 'Messages' | ||||
SETTING_ANON_MODE = 'AnonymousMode' | ||||
ANON_IP = '127.0.0.1' | ||||
|
r1368 | UPLOAD_DIRS ={ | ||
'PostImage': 'images/', | ||||
'Attachment': 'files/', | ||||
} | ||||
FILE_EXTENSION_DELIMITER = '.' | ||||
|
r1362 | |||
def is_anonymous_mode(): | ||||
return get_bool(SETTING_MESSAGES, SETTING_ANON_MODE) | ||||
|
r210 | def get_client_ip(request): | ||
|
r1362 | if is_anonymous_mode(): | ||
ip = ANON_IP | ||||
|
r210 | else: | ||
|
r1362 | x_forwarded_for = request.META.get(HTTP_FORWARDED) | ||
if x_forwarded_for: | ||||
ip = x_forwarded_for.split(',')[-1].strip() | ||||
else: | ||||
ip = request.META.get(META_REMOTE_ADDR) | ||||
|
r542 | return ip | ||
|
r990 | # TODO The output format is not epoch because it includes microseconds | ||
|
r542 | def datetime_to_epoch(datetime): | ||
return int(time.mktime(timezone.localtime( | ||||
datetime,timezone.get_current_timezone()).timetuple()) | ||||
|
r765 | * 1000000 + datetime.microsecond) | ||
|
r853 | |||
def get_websocket_token(user_id='', timestamp=''): | ||||
""" | ||||
Create token to validate information provided by new connection. | ||||
""" | ||||
sign = hmac.new(settings.CENTRIFUGE_PROJECT_SECRET.encode()) | ||||
sign.update(settings.CENTRIFUGE_PROJECT_ID.encode()) | ||||
sign.update(user_id.encode()) | ||||
sign.update(timestamp.encode()) | ||||
token = sign.hexdigest() | ||||
|
r957 | return token | ||
|
r1106 | def cached_result(key_method=None): | ||
|
r957 | """ | ||
Caches method result in the Django's cache system, persisted by object name, | ||||
object name and model id if object is a Django model. | ||||
""" | ||||
|
r1106 | def _cached_result(function): | ||
def inner_func(obj, *args, **kwargs): | ||||
# TODO Include method arguments to the cache key | ||||
cache_key_params = [obj.__class__.__name__, function.__name__] | ||||
if isinstance(obj, Model): | ||||
cache_key_params.append(str(obj.id)) | ||||
if key_method is not None: | ||||
cache_key_params += [str(arg) for arg in key_method(obj)] | ||||
cache_key = CACHE_KEY_DELIMITER.join(cache_key_params) | ||||
|
r957 | |||
|
r1106 | persisted_result = cache.get(cache_key) | ||
if persisted_result is not None: | ||||
result = persisted_result | ||||
else: | ||||
result = function(obj, *args, **kwargs) | ||||
cache.set(cache_key, result) | ||||
|
r957 | |||
|
r1106 | return result | ||
|
r957 | |||
|
r1106 | return inner_func | ||
return _cached_result | ||||
|
r1109 | |||
def is_moderator(request): | ||||
try: | ||||
moderate = request.user.has_perm(PERMISSION_MODERATE) | ||||
except AttributeError: | ||||
moderate = False | ||||
|
r1305 | return moderate | ||
def get_file_hash(file) -> str: | ||||
md5 = hashlib.md5() | ||||
for chunk in file.chunks(): | ||||
md5.update(chunk) | ||||
return md5.hexdigest() | ||||
|
r1328 | |||
def validate_file_size(size: int): | ||||
max_size = boards.settings.get_int('Forms', 'MaxFileSize') | ||||
if size > max_size: | ||||
raise forms.ValidationError( | ||||
_('File must be less than %s bytes') | ||||
% str(max_size)) | ||||
|
r1368 | |||
def get_upload_filename(model_instance, old_filename): | ||||
# TODO Use something other than random number in file name | ||||
|
r1370 | if hasattr(model_instance, 'mimetype'): | ||
extension = model_instance.mimetype | ||||
|
r1369 | else: | ||
extension = old_filename.split(FILE_EXTENSION_DELIMITER)[-1:][0] | ||||
|
r1368 | new_name = '{}{}.{}'.format( | ||
str(int(time.mktime(time.gmtime()))), | ||||
str(int(random() * 1000)), | ||||
|
r1369 | extension) | ||
|
r1368 | |||
directory = UPLOAD_DIRS[type(model_instance).__name__] | ||||
return os.path.join(directory, new_name) | ||||
|
r1371 | |||
def get_file_mimetype(file) -> str: | ||||
return magic.from_buffer(file.chunks().__next__(), mime=True) \ | ||||
.decode().split('/')[-1] | ||||