514 lines
15 KiB
Python
514 lines
15 KiB
Python
# w.c.s. - web application for online forms
|
||
# Copyright (C) 2005-2017 Entr'ouvert
|
||
#
|
||
# This program is free software; you can redistribute it and/or modify
|
||
# it under the terms of the GNU General Public License as published by
|
||
# the Free Software Foundation; either version 2 of the License, or
|
||
# (at your option) any later version.
|
||
#
|
||
# This program is distributed in the hope that it will be useful,
|
||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||
# GNU General Public License for more details.
|
||
#
|
||
# You should have received a copy of the GNU General Public License
|
||
# along with this program; if not, see <http://www.gnu.org/licenses/>.
|
||
|
||
import datetime
|
||
from decimal import Decimal
|
||
from decimal import InvalidOperation as DecimalInvalidOperation
|
||
from decimal import DivisionByZero as DecimalDivisionByZero
|
||
import hashlib
|
||
import math
|
||
import string
|
||
import random
|
||
|
||
import pyproj
|
||
from pyproj import Geod
|
||
|
||
try:
|
||
import langdetect
|
||
from langdetect.lang_detect_exception import LangDetectException
|
||
except ImportError:
|
||
langdetect = None
|
||
|
||
from django import template
|
||
from django.template import defaultfilters
|
||
from django.utils import dateparse
|
||
from django.utils import six
|
||
from django.utils.encoding import force_bytes, force_text
|
||
from django.utils.safestring import mark_safe
|
||
from wcs.qommon import evalutils
|
||
from wcs.qommon import tokens
|
||
from wcs.qommon.admin.texts import TextsDirectory
|
||
from wcs.roles import Role
|
||
|
||
register = template.Library()
|
||
|
||
|
||
@register.filter
|
||
def get(mapping, key):
|
||
if hasattr(key, 'get_value'):
|
||
key = key.get_value() # unlazy
|
||
if hasattr(mapping, 'get'):
|
||
return mapping.get(key)
|
||
if isinstance(mapping, (tuple, list)):
|
||
try:
|
||
key = int(key)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return mapping[key]
|
||
|
||
|
||
@register.filter
|
||
def startswith(string, substring):
|
||
return string and force_text(string).startswith(force_text(substring))
|
||
|
||
|
||
@register.filter
|
||
def endswith(string, substring):
|
||
return string and force_text(string).endswith(force_text(substring))
|
||
|
||
|
||
@register.filter
|
||
def split(string, separator=' '):
|
||
if not string:
|
||
return []
|
||
return force_text(string).split(force_text(separator))
|
||
|
||
|
||
@register.filter
|
||
def strip(string, chars=None):
|
||
if not string:
|
||
return ''
|
||
if chars:
|
||
return force_text(string).strip(force_text(chars))
|
||
else:
|
||
return force_text(string).strip()
|
||
|
||
|
||
@register.filter
|
||
def parse_date(date_string):
|
||
try:
|
||
return evalutils.make_date(date_string)
|
||
except ValueError:
|
||
pass
|
||
# fallback to Django function
|
||
try:
|
||
return dateparse.parse_date(date_string)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def date(value, arg=None):
|
||
if arg is None:
|
||
value = parse_date(value)
|
||
if not value:
|
||
return ''
|
||
from wcs.variables import lazy_date
|
||
return lazy_date(parse_date(value))
|
||
if not isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
|
||
value = parse_datetime(value) or parse_date(value)
|
||
return defaultfilters.date(value, arg=arg)
|
||
|
||
|
||
@register.filter
|
||
def parse_datetime(datetime_string):
|
||
try:
|
||
return evalutils.make_datetime(datetime_string)
|
||
except ValueError:
|
||
pass
|
||
# fallback to Django function
|
||
try:
|
||
return dateparse.parse_datetime(datetime_string)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
@register.filter(name='datetime', expects_localtime=True, is_safe=False)
|
||
def datetime_(value, arg=None):
|
||
if arg is None:
|
||
value = parse_datetime(value)
|
||
if not value:
|
||
return ''
|
||
from wcs.variables import lazy_date
|
||
return lazy_date(parse_datetime(value))
|
||
if not isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
|
||
value = parse_datetime(value)
|
||
return defaultfilters.date(value, arg=arg)
|
||
|
||
|
||
@register.filter
|
||
def parse_time(time_string):
|
||
# if input is a datetime, extract its time
|
||
try:
|
||
dt = parse_datetime(time_string)
|
||
if dt:
|
||
return dt.time()
|
||
except (ValueError, TypeError):
|
||
pass
|
||
# fallback to Django function
|
||
try:
|
||
return dateparse.parse_time(time_string)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def time(value, arg=None):
|
||
if arg is None:
|
||
parsed = parse_time(value)
|
||
return parsed if parsed is not None else '' # because bool(midnight) == False
|
||
if not isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
|
||
value = parse_time(value)
|
||
return defaultfilters.date(value, arg=arg)
|
||
|
||
|
||
def parse_decimal(value, do_raise=False):
|
||
if hasattr(value, 'get_value'):
|
||
value = value.get_value() # unlazy
|
||
if isinstance(value, six.string_types):
|
||
# replace , by . for French users comfort
|
||
value = value.replace(',', '.')
|
||
try:
|
||
return Decimal(value).quantize(Decimal('1.0000')).normalize()
|
||
except (ArithmeticError, TypeError):
|
||
if do_raise:
|
||
raise
|
||
return Decimal(0)
|
||
|
||
|
||
@register.filter(is_safe=False)
|
||
def decimal(value, arg=None):
|
||
if not isinstance(value, Decimal):
|
||
value = parse_decimal(value)
|
||
if arg is None:
|
||
return value
|
||
if hasattr(arg, 'get_value'):
|
||
arg = arg.get_value() # unlazy
|
||
return defaultfilters.floatformat(value, arg=arg)
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def add_days(value, arg):
|
||
if hasattr(value, 'timetuple'):
|
||
# extract real value in case of lazy object
|
||
value = value.timetuple()
|
||
value = parse_date(value) # consider only date, not hours
|
||
if not value:
|
||
return ''
|
||
from wcs.variables import lazy_date
|
||
arg = parse_decimal(arg)
|
||
if not arg:
|
||
return lazy_date(value)
|
||
result = value + datetime.timedelta(days=float(arg))
|
||
if hasattr(result, 'date'):
|
||
result = result.date()
|
||
return lazy_date(result)
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def add_hours(value, arg):
|
||
if hasattr(value, 'timetuple'):
|
||
# extract real value in case of lazy object
|
||
value = value.timetuple()
|
||
value = parse_datetime(value)
|
||
if not value:
|
||
return ''
|
||
from wcs.variables import lazy_date
|
||
arg = parse_decimal(arg)
|
||
if not arg:
|
||
return lazy_date(value)
|
||
return lazy_date(value + datetime.timedelta(hours=float(arg)))
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def age_in_days(value, now=None):
|
||
try:
|
||
return evalutils.age_in_days(value, now)
|
||
except ValueError:
|
||
return ''
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def age_in_hours(value, now=None):
|
||
# consider value and now as datetimes (and not dates)
|
||
if hasattr(value, 'timetuple'):
|
||
# extract real value in case of lazy object
|
||
value = value.timetuple()
|
||
value = parse_datetime(value)
|
||
if not value:
|
||
return ''
|
||
if now is not None:
|
||
if hasattr(now, 'timetuple'):
|
||
now = now.timetuple()
|
||
now = parse_datetime(now)
|
||
if not now:
|
||
return ''
|
||
else:
|
||
now = datetime.datetime.now()
|
||
return int((now - value).total_seconds() / 3600)
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def age_in_years(value, today=None):
|
||
try:
|
||
return evalutils.age_in_years_and_months(value, today)[0]
|
||
except ValueError:
|
||
return ''
|
||
|
||
|
||
@register.filter(expects_localtime=True, is_safe=False)
|
||
def age_in_months(value, today=None):
|
||
try:
|
||
years, months = evalutils.age_in_years_and_months(value, today)
|
||
except ValueError:
|
||
return ''
|
||
return years*12 + months
|
||
|
||
|
||
@register.simple_tag
|
||
def standard_text(text_id):
|
||
return mark_safe(TextsDirectory.get_html_text(str(text_id)))
|
||
|
||
|
||
@register.simple_tag(takes_context=True)
|
||
def action_button(context, action_id, label, delay=3):
|
||
from wcs.formdef import FormDef
|
||
formdata_id = context.get('form_number_raw')
|
||
formdef_urlname = context.get('form_slug')
|
||
if not (formdef_urlname and formdata_id):
|
||
return ''
|
||
formdef = FormDef.get_by_urlname(formdef_urlname)
|
||
formdata = formdef.data_class().get(formdata_id, ignore_errors=True)
|
||
token = tokens.Token(expiration_delay=delay*86400, size=64)
|
||
token.type = 'action'
|
||
token.context = {
|
||
'form_slug': formdef_urlname,
|
||
'form_number_raw': formdata_id,
|
||
'action_id': action_id,
|
||
'label': label,
|
||
}
|
||
token.store()
|
||
return '---===BUTTON:%s:%s===---' % (token.id, label)
|
||
|
||
|
||
@register.filter
|
||
def add(term1, term2):
|
||
'''replace the "add" native django filter'''
|
||
|
||
# consider None content as the empty string
|
||
if term1 is None:
|
||
term1 = ''
|
||
if term2 is None:
|
||
term2 = ''
|
||
|
||
# return available number if the other term is the empty string
|
||
if term1 == '':
|
||
try:
|
||
return parse_decimal(term2, do_raise=True)
|
||
except (ArithmeticError, TypeError):
|
||
pass
|
||
if term2 == '':
|
||
try:
|
||
return parse_decimal(term1, do_raise=True)
|
||
except (ArithmeticError, TypeError):
|
||
pass
|
||
|
||
# compute addition if both terms are numbers
|
||
try:
|
||
return parse_decimal(term1, do_raise=True) + parse_decimal(term2, do_raise=True)
|
||
except (ArithmeticError, TypeError):
|
||
pass
|
||
|
||
# fallback to django add filter
|
||
if hasattr(term1, 'get_value'):
|
||
term1 = term1.get_value() # unlazy
|
||
if hasattr(term2, 'get_value'):
|
||
term2 = term2.get_value() # unlazy
|
||
return defaultfilters.add(term1, term2)
|
||
|
||
|
||
@register.filter
|
||
def subtract(term1, term2):
|
||
return parse_decimal(term1) - parse_decimal(term2)
|
||
|
||
|
||
@register.filter
|
||
def multiply(term1, term2):
|
||
return parse_decimal(term1) * parse_decimal(term2)
|
||
|
||
|
||
@register.filter
|
||
def divide(term1, term2):
|
||
try:
|
||
return parse_decimal(term1) / parse_decimal(term2)
|
||
except DecimalInvalidOperation:
|
||
return ''
|
||
except DecimalDivisionByZero:
|
||
return ''
|
||
|
||
|
||
@register.filter
|
||
def ceil(value):
|
||
'''the smallest integer value greater than or equal to value'''
|
||
return decimal(math.ceil(parse_decimal(value)))
|
||
|
||
|
||
@register.filter
|
||
def floor(value):
|
||
return decimal(math.floor(parse_decimal(value)))
|
||
|
||
|
||
@register.filter(name='abs')
|
||
def abs_(value):
|
||
return decimal(abs(parse_decimal(value)))
|
||
|
||
|
||
@register.simple_tag
|
||
def version_hash():
|
||
from wcs.qommon.admin.menu import get_vc_version
|
||
return hashlib.md5(force_bytes(get_vc_version())).hexdigest()
|
||
|
||
|
||
def generate_token(alphabet, length):
|
||
r = random.SystemRandom()
|
||
return ''.join([r.choice(alphabet) for i in range(length)])
|
||
|
||
|
||
@register.simple_tag
|
||
def token_decimal(length=6):
|
||
# entropy by default is log(10^6)/log(2) = 19.93 bits
|
||
# decimal always need more length than alphanum for the same security level
|
||
# for 128bits security level, length must be more than log(2^128)/log(10) = 38.53 digits
|
||
return generate_token(string.digits, length)
|
||
|
||
|
||
@register.simple_tag
|
||
def token_alphanum(length=4):
|
||
# use of a 28 characters alphabet using uppercase letters and digits but
|
||
# removing confusing characters and digits 0, O, 1 and I.
|
||
# entropy by default is log(28^4)/log(2) = 19.22 bits
|
||
# for 128 bits security level length must be more than log(2^128)/log(28) = 26.62 characters
|
||
return generate_token('23456789ABCDEFGHJKLMNPQRSTUVWXYZ', length)
|
||
|
||
|
||
@register.filter
|
||
def token_check(token1, token2):
|
||
return force_text(token1).strip().upper() == force_text(token2).strip().upper()
|
||
|
||
|
||
def get_latlon(obj):
|
||
if getattr(obj, 'geoloc', None):
|
||
if 'base' in obj.geoloc:
|
||
return obj.geoloc['base']['lat'], obj.geoloc['base']['lon']
|
||
return None, None
|
||
if hasattr(obj, 'get_value'):
|
||
obj = obj.get_value() # unlazy
|
||
if isinstance(obj, dict) and 'lat' in obj and 'lon' in obj:
|
||
try:
|
||
return float(obj['lat']), float(obj['lon'])
|
||
except (TypeError, ValueError):
|
||
pass
|
||
if isinstance(obj, dict) and 'lat' in obj and 'lng' in obj:
|
||
try:
|
||
return float(obj['lat']), float(obj['lng'])
|
||
except (TypeError, ValueError):
|
||
pass
|
||
if isinstance(obj, six.string_types) and ';' in obj:
|
||
try:
|
||
return float(obj.split(';')[0]), float(obj.split(';')[1])
|
||
except ValueError:
|
||
pass
|
||
return None, None
|
||
|
||
|
||
@register.filter
|
||
def distance(obj1, obj2):
|
||
lat1, lon1 = get_latlon(obj1)
|
||
if lat1 is None or lon1 is None:
|
||
return None
|
||
lat2, lon2 = get_latlon(obj2)
|
||
if lat2 is None or lon2 is None:
|
||
return None
|
||
geod = Geod(ellps='WGS84')
|
||
distance = geod.inv(lon1, lat1, lon2, lat2)[2]
|
||
return distance
|
||
|
||
|
||
@register.filter
|
||
def distance_filter(queryset, distance=1000):
|
||
return queryset.distance_filter(distance=int(distance))
|
||
|
||
|
||
@register.filter
|
||
def order_by(queryset, attribute):
|
||
return queryset.order_by(attribute)
|
||
|
||
|
||
@register.filter
|
||
def reproj(coords, projection_name):
|
||
proj = pyproj.Proj(init='EPSG:4326')
|
||
target_proj = pyproj.Proj(init=projection_name)
|
||
return pyproj.transform(proj, target_proj, coords['lon'], coords['lat'])
|
||
|
||
|
||
@register.filter
|
||
def has_role(user, role_name):
|
||
if not callable(getattr(user, 'get_roles', None)):
|
||
# do not fail on non-user objects, just return False
|
||
return False
|
||
for role_id in user.get_roles():
|
||
try:
|
||
if role_name == Role.get(role_id).name:
|
||
return True
|
||
except KeyError: # role has been deleted
|
||
pass
|
||
return False
|
||
|
||
|
||
@register.filter
|
||
def language_detect(value):
|
||
if langdetect is None:
|
||
return ''
|
||
try:
|
||
return langdetect.detect(str(value))
|
||
except LangDetectException:
|
||
return ''
|
||
|
||
|
||
@register.filter(is_safe=False)
|
||
def phonenumber_fr(value, separator=' '):
|
||
DROMS = ('262', '508', '590', '594', '596')
|
||
|
||
if not value or not isinstance(value, six.string_types):
|
||
return value
|
||
number = value.strip()
|
||
if not number:
|
||
return value
|
||
if number[0] == '+':
|
||
international = '+'
|
||
number = '00' + number[1:]
|
||
else:
|
||
international = '00' + separator
|
||
number = ''.join(c for c in number if c in '0123456789')
|
||
|
||
def in_pairs(num):
|
||
return separator.join(num[i*2:i*2+2] for i in range(len(num)//2))
|
||
|
||
# local number
|
||
if len(number) == 10 and number[0] == '0' and number[1] in '123456789':
|
||
return in_pairs(number)
|
||
# international
|
||
if len(number) == 14 and number[0:5] == '00330':
|
||
# +/00 33 (0)x xx xx xx xx : remove (0)
|
||
number = number[0:4] + number[5:]
|
||
if len(number) == 13 and number[0:4] == '0033':
|
||
return international + '33' + separator + number[4] + separator + in_pairs(number[5:])
|
||
if len(number) == 11 and number[0:2] == '00' and number[2:5] in DROMS:
|
||
return international + number[2:5] + separator + in_pairs(number[5:])
|
||
|
||
# unknown
|
||
return value
|