<commit_before># -*- coding: utf-8 -*-

"""
General utilities
"""

import http.client
import json
import socket
import time
import urllib
from functools import wraps
from itertools import chain
from typing import Iterable

from flask import flash, redirect, request, url_for, session
from flask_login import current_user

from datetime import datetime, timedelta

from sipa.config.default import PBX_URI


def timetag_today():
    """Return the timetag for today"""
    return int(time.time() // 86400)


def get_bustimes(stopname, count=10):
    """Parses the VVO-Online API return string.
    API returns in format [["line", "to", "minutes"],[__],[__]], where "__" are
    up to nine more Elements.

    :param stopname: Requested stop.
    :param count: Limit the entries for the stop.
    """
    conn = http.client.HTTPConnection('widgets.vvo-online.de', timeout=1)

    stopname = stopname.replace(' ', '%20')
    try:
        conn.request(
            'GET',
            '/abfahrtsmonitor/Abfahrten.do?ort=Dresden&hst={}'.format(stopname)
        )
        response = conn.getresponse()
    except socket.error:
        return None

    response_data = json.loads(response.read().decode())

    return ({
        'line': i[0],
        'dest': i[1],
        'minutes_left': int(i[2]) if i[2] else 0,
    } for i in response_data)
# TODO: check whether this is the correct format


def support_hotline_available():
    """Asks the PBX if there are agents logged in to anwser calls to our
    support hotline.

    :return: True if the hotline is available
    """
    [avail, time] = session.get('PBX_available', [False, datetime.fromtimestamp(0)])

    if datetime.now() - time > timedelta(minutes=2):
        # refresh availability from pbx
        try:
            avail = urllib.request.urlopen(PBX_URI, timeout=0.5).read()
            session['PBX_available'] = [avail, datetime.now()]
        except urllib.error.URLError:
            avail = False

    if avail == b'AVAILABLE':
        return True
    else:
        return False


def password_changeable(user):
    """A decorator used to disable functions (routes) if a certain feature
    is not provided by the User class.

    given_features has to be a callable to ensure runtime distinction
    between datasources.

    :param needed_feature: The feature needed
    :param given_features: A callable returning the set of supported features
    :return:
    """
    def feature_decorator(func):
        @wraps(func)
        def decorated_view(*args, **kwargs):
            if user.is_authenticated and user.can_change_password:
                return func(*args, **kwargs)
            else:
                def not_supported():
                    flash("Diese Funktion ist nicht verfügbar.", 'error')
                    return redirect(redirect_url())
                return not_supported()

        return decorated_view<commit_msg>Handle timeout errors when fetching hotline availability<commit_after># -*- coding: utf-8 -*-

"""
General utilities
"""

import http.client
import json
import requests
import socket
import time
from functools import wraps
from itertools import chain
from typing import Iterable

from flask import flash, redirect, request, url_for, session
from flask_login import current_user

from datetime import datetime, timedelta

from sipa.config.default import PBX_URI


def timetag_today():
    """Return the timetag for today"""
    return int(time.time() // 86400)


def get_bustimes(stopname, count=10):
    """Parses the VVO-Online API return string.
    API returns in format [["line", "to", "minutes"],[__],[__]], where "__" are
    up to nine more Elements.

    :param stopname: Requested stop.
    :param count: Limit the entries for the stop.
    """
    conn = http.client.HTTPConnection('widgets.vvo-online.de', timeout=1)

    stopname = stopname.replace(' ', '%20')
    try:
        conn.request(
            'GET',
            '/abfahrtsmonitor/Abfahrten.do?ort=Dresden&hst={}'.format(stopname)
        )
        response = conn.getresponse()
    except socket.error:
        return None

    response_data = json.loads(response.read().decode())

    return ({
        'line': i[0],
        'dest': i[1],
        'minutes_left': int(i[2]) if i[2] else 0,
    } for i in response_data)
# TODO: check whether this is the correct format


def support_hotline_available():
    """Asks the PBX if there are agents logged in to anwser calls to our
    support hotline.

    :return: True if the hotline is available
    """
    [avail, time] = session.get('PBX_available', [False, datetime.fromtimestamp(0)])

    if datetime.now() - time > timedelta(minutes=2):
        # refresh availability from pbx
        try:
            r = requests.get(PBX_URI, timeout=0.5)
            r.raise_for_status()
            avail = r.text
            session['PBX_available'] = [avail, datetime.now()]
        except requests.exceptions.RequestException:
            avail = False

    if avail == b'AVAILABLE':
        return True
    else:
        return False


def password_changeable(user):
    """A decorator used to disable functions (routes) if a certain feature
    is not provided by the User class.

    given_features has to be a callable to ensure runtime distinction
    between datasources.

    :param needed_feature: The feature needed
    :param given_features: A callable returning the set of supported features
    :return:
    """
    def feature_decorator(func):
        @wraps(func)
        def decorated_view(*args, **kwargs):
            if user.is_authenticated and user.can_change_password:
                return func(*args, **kwargs)
            else:
                def not_supported():
                    flash("Diese Funktion ist nicht verfügbar.", 'error')
                    return redirect(redirect_url())
                return not_supported()

        return decorated_view
{'fraction_non_alphanumeric': 0.10174369392246488, 'fraction_numerical': 0.006771626883358727, 'mean_word_length': 2.685589519650655, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1321411', 'n_tokens_mistral': 1758, 'n_tokens_neox': 1678, 'n_words': 542}