<commit_before>from datetime import date, timedelta

from flask import Blueprint, request, render_template, redirect, url_for, abort, g, flash
from flask.ext.babel import lazy_gettext as l_, _, ngettext
from flask.ext.login import login_required

from formencode import All
from formencode.validators import Email, NotEmpty
from sprox.formbase import EditableForm, Field
from sprox.widgets import PropertySingleSelectField
from tw.forms import TextField, CheckBox

from sqlalchemy.sql.expression import and_, or_
from sqlalchemy import func

from skylines import db
from skylines.forms import (
    BootstrapForm, units, ChangePasswordForm, ChangeClubForm, CreateClubForm
)
from skylines.lib.validators import UniqueValueUnless
from skylines.lib.dbutil import get_requested_record
from skylines.lib.decorators import validate
from skylines.model import (
    User, Club, Flight, Follower, Location, IGCFile, Notification, Event
)
from skylines.model.event import (
    create_follower_notification, create_club_join_event
)
from skylines.views.users import recover_user_password

user_blueprint = Blueprint('user', 'skylines')


@user_blueprint.url_value_preprocessor
def _pull_user_id(endpoint, values):
    g.user_id = values.pop('user_id')
    g.user = get_requested_record(User, g.user_id)


@user_blueprint.url_defaults
def _add_user_id(endpoint, values):
    if hasattr(g, 'user_id'):
        values.setdefault('user_id', g.user_id)


def _get_distance_flight(distance):
    return Flight.query() \
        .filter(Flight.pilot == g.user) \
        .filter(Flight.olc_classic_distance >= distance) \
        .order_by(Flight.landing_time) \
        .first()


def _get_distance_flights():
    distance_flights = []

    largest_flight = g.user.get_largest_flights().first()
    if largest_flight:
        distance_flights.append([largest_flight.olc_classic_distance,
                                 largest_flight])

    for distance in [50000, 100000, 300000, 500000, 700000, 1000000]:
        distance_flight = _get_distance_flight(distance)
        if distance_flight is not None:
            distance_flights.append([distance, distance_flight])

    distance_flights.sort()
    return distance_flights


def _get_last_year_statistics():
    query = db.session.query(func.count('*').label('flights'),
                             func.sum(Flight.olc_classic_distance).label('distance'),
                             func.sum(Flight.duration).label('duration')) \
                      .filter(Flight.pilot == g.user) \
                      .filter(Flight.date_local > (date.today() - timedelta(days=365))) \
                      .first()

    last_year_statistics = dict(flights=0,
                                distance=0,
                                duration=timedelta(0),
                                speed=0)

    if query and query.flights > 0:
        duration_seconds = query.duration.days * 24 * 3600 + query.duration.seconds

        if duration_seconds > 0:
            last_year_statistics['speed'] = float(query.distance) / duration_seconds

        last_year_statistics['flights'] = query.flights
        last_year_statistics['distance'] = query.distance
        last_year_statistics['duration'] = query.duration

        last_year_statistics['average_distance'] = query.distance / query.flights
        last_year_statistics['average_duration'] = query.duration / query.flights

    return last_year_statistics


def _get_takeoff_locations():
    return Location.get_clustered_locations(Flight.takeoff_location_wkt,
                                            filter=(Flight.pilot == g.user))


def mark_user_notifications_read(user):
    if not g.current_user:
        return

    def add_user_filter(query):
        return query.filter(Event.actor_id == user.id)

    Notification.mark_all_read(g.current_user, filter_func=add_user_filter)
    db.session.commit()


@user_blueprint.route('/')
def index():
    mark_user_notifications_read(g.user)

    return render_template(
        'users/view.jinja',
        active_page='settings', user=g.user,
        distance_flights=_get_distance_flights(),
        takeoff_locations=_get_takeoff_locations(),
        last_year_statistics=_get_last_year_statistics())


@user_blueprint.route('/change_password', methods=['GET', 'POST'])
def change_password():
    if not g.user.is_writable(g.current_user):
        abort(403)

    form = ChangePasswordForm()
    if form.validate_on_submit():
        response = change_password_post(form)
        if response:
            return response

    return render_template('users/change_password.jinja', form=form)


def change_password_post(form):
    g.user.password = <PASSWORD>.password.data
    g.user.recover_key = None

    db.session.commit()

    return redirect(url_for('.index'))


class DelaySelectField(PropertySingleSelectField):
    def _my_update_params(self, d, nullable=False):
        options = [(0, _('None'))]
        for x in range(1, 10) + range(10, 30, 5) + range(30, 61, 15):
            options.append((x, ngettext(u'%(num)u minute', u'%(num)u minutes', x)))
        d['options'] = options
        return d


def filter_user_id(model):
    return model.id == g.user.id


class EditUserForm(EditableForm):
    __base_widget_type__ = BootstrapForm
    __model__ = User
    __hide_fields__ = ['id']
    __limit_fields__ = ['email_address', 'name',
                        'tracking_delay', 'unit_preset',
                        'distance_unit', 'speed_unit',
                        'lift_unit', 'altitude_unit',
                        'eye_candy']
    __field_widget_args__ = {
        'email_address': dict(label_text=l_('eMail Address')),
        'name': dict(label_text=l_('Name')),
        'tracking_delay': dict(label_text=l_('Tracking Delay')),
        'unit_preset': dict(label_text=l_('Unit Preset')),
        'distance_unit': dict(label_text=l_('Distance Unit')),
        'speed_unit': dict(label_text=l_('Speed Unit')),
        'lift_unit': dict(label_text=l_('Lift Unit')),
        'altitude_unit': dict(label_text=l_('Altitude Unit')),
        'eye_candy': dict(label_text=l_('Eye Candy')),
    }

    email_address = Field(TextField, All(
        Email(not_empty=True),
        UniqueValueUnless(filter_user_id, db.session, __model__, 'email_address')))
    name = Field(TextField, NotEmpty)
    tracking_delay = DelaySelectField
    unit_preset = units.PresetSelectField("unit_preset")
    distance_unit = units.DistanceSelectField
    speed_unit = units.SpeedSelectField
    lift_unit = units.LiftSelectField
    altitude_unit = units.AltitudeSelectField
    eye_candy = Field(CheckBox)

edit_user_form = EditUserForm(db.session)


@user_blueprint.route('/edit')
def edit():
    if not g.user.is_writable(g.current_user):
        abort(403)

    return render_template(
        'generic/form.jinja',
        active_page='settings', title=_('Edit User'),
        form=edit_user_form, values=g.user,
        include_script='users/after-edit-form.jinja')


@user_blueprint.route('/edit', methods=['POST'])
@validate(edit_user_form, edit)
def edit_post():
    if not g.user.is_writable(g.current_user):
        abort(403)

    g.user.email_address = request.form['email_address']
    g.user.name = request.form['name']
    g.user.tracking_delay = request.form.get('tracking_delay', 0)

    unit_preset = request.form.get('unit_preset', 1, type=int)
    if unit_preset == 0:
        g.user.distance_unit = request.form.get('distance_unit', 1, type=int)
        g.user.speed_unit = request.form.get('speed_unit', 1, type=int)
        g.user.lift_unit = request.form.get('lift_unit', 0, type=int)
        g.user.altitude_unit = request.form.get('altitude_unit', 0, type=int)
    else:
        g.user.unit_preset = unit_preset

    g.user.eye_candy = request.form.get('eye_candy', False, type=bool)

    db.session.commit()

    return redirect(url_for('.index'))


@user_blueprint.route('/change_club', methods=['GET', 'POST'])
def change_club():
    if not g.user.is_writable(g.current_user):
        abort(403)

    change_form = ChangeClubForm(club=g.user.club_id)
    create_form = CreateClubForm()

    if request.endpoint.endswith('.change_club'):
        if change_form.validate_on_submit():
            response = change_club_post(change_form)
            if response:
                return response

    if request.endpoint.endswith('.create_club'):
        if create_form.validate_on_submit():
            response = create_club_post(create_form)
            if response:
                return response

    return render_template(
        'users/change_club.jinja',
        change_form=change_form, create_form=create_form)


@user_blueprint.route('/create_club', methods=['GET', 'POST'])
def create_club():
    return change_club()


def change_club_post(form):
    old_club_id = g.user.club_id
    new_club_id = form.club.data if form.club.data != 0 else None

    if old_club_id == new_club_id:
        return

    g.user.club_id = new_club_id

    create_club_join_event(new_club_id, g.user)

    # assign the user's new club to all of his flights that have
    # no club yet
    flights = Flight.query().join(IGCFile)
    flights = flights.filter(and_(Flight.club_id == None,
                                  or_(Flight.pilot_id == g.user.id,
                                      IGCFile.owner_id == g.user.id)))
    for flight in flights:
        flight.club_id = g.user.club_id

    db.session.commit()

    return redirect(url_for('.index'))


def create_club_post(form):
    club = Club(name=form.name.data)
    club.owner_id = g.current_user.id
    db.session.add(club)

    g.user.club = club

    db.session.commit()

    return redirect(url_for('.index'))


@user_blueprint.route('/follow')
@login_required
def follow():
    Follower.follow(g.current_user, g.user)
    create_follower_notification(g.user, g.current_user)
    db.session.commit()
    return redirect(url_for('.index'))


@user_blueprint.route('/unfollow')
@login_required
def unfollow():
    Follower.unfollow(g.current_user, g.user)
    db.session.commit()
    return redirect(url_for('.index'))


@user_blueprint.route('/tracking_register')
def tracking_register():
    if not g.user.is_writable(g.current_user):
        abort(403)

    g.user.generate_tracking_key()
    db.session.commit()

    return redirect(request.values.get('came_from', '/tracking/info'))


@user_blueprint.route('/recover_password')
def recover_password():
    if not g.current_user or not g.current_user.is_manager():
        abort(403)

    recover_user_password(g.user)
    flash('A password recovery email was sent to that user.')

    db.session.commit()

    return redirect(url_for('.index'))
<commit_msg>form: Simplified user form handling<commit_after>from datetime import date, timedelta

from flask import Blueprint, request, render_template, redirect, url_for, abort, g, flash
from flask.ext.babel import lazy_gettext as l_, _, ngettext
from flask.ext.login import login_required

from formencode import All
from formencode.validators import Email, NotEmpty
from sprox.formbase import EditableForm, Field
from sprox.widgets import PropertySingleSelectField
from tw.forms import TextField, CheckBox

from sqlalchemy.sql.expression import and_, or_
from sqlalchemy import func

from skylines import db
from skylines.forms import (
    BootstrapForm, units, ChangePasswordForm, ChangeClubForm, CreateClubForm
)
from skylines.lib.validators import UniqueValueUnless
from skylines.lib.dbutil import get_requested_record
from skylines.lib.decorators import validate
from skylines.model import (
    User, Club, Flight, Follower, Location, IGCFile, Notification, Event
)
from skylines.model.event import (
    create_follower_notification, create_club_join_event
)
from skylines.views.users import recover_user_password

user_blueprint = Blueprint('user', 'skylines')


@user_blueprint.url_value_preprocessor
def _pull_user_id(endpoint, values):
    g.user_id = values.pop('user_id')
    g.user = get_requested_record(User, g.user_id)


@user_blueprint.url_defaults
def _add_user_id(endpoint, values):
    if hasattr(g, 'user_id'):
        values.setdefault('user_id', g.user_id)


def _get_distance_flight(distance):
    return Flight.query() \
        .filter(Flight.pilot == g.user) \
        .filter(Flight.olc_classic_distance >= distance) \
        .order_by(Flight.landing_time) \
        .first()


def _get_distance_flights():
    distance_flights = []

    largest_flight = g.user.get_largest_flights().first()
    if largest_flight:
        distance_flights.append([largest_flight.olc_classic_distance,
                                 largest_flight])

    for distance in [50000, 100000, 300000, 500000, 700000, 1000000]:
        distance_flight = _get_distance_flight(distance)
        if distance_flight is not None:
            distance_flights.append([distance, distance_flight])

    distance_flights.sort()
    return distance_flights


def _get_last_year_statistics():
    query = db.session.query(func.count('*').label('flights'),
                             func.sum(Flight.olc_classic_distance).label('distance'),
                             func.sum(Flight.duration).label('duration')) \
                      .filter(Flight.pilot == g.user) \
                      .filter(Flight.date_local > (date.today() - timedelta(days=365))) \
                      .first()

    last_year_statistics = dict(flights=0,
                                distance=0,
                                duration=timedelta(0),
                                speed=0)

    if query and query.flights > 0:
        duration_seconds = query.duration.days * 24 * 3600 + query.duration.seconds

        if duration_seconds > 0:
            last_year_statistics['speed'] = float(query.distance) / duration_seconds

        last_year_statistics['flights'] = query.flights
        last_year_statistics['distance'] = query.distance
        last_year_statistics['duration'] = query.duration

        last_year_statistics['average_distance'] = query.distance / query.flights
        last_year_statistics['average_duration'] = query.duration / query.flights

    return last_year_statistics


def _get_takeoff_locations():
    return Location.get_clustered_locations(Flight.takeoff_location_wkt,
                                            filter=(Flight.pilot == g.user))


def mark_user_notifications_read(user):
    if not g.current_user:
        return

    def add_user_filter(query):
        return query.filter(Event.actor_id == user.id)

    Notification.mark_all_read(g.current_user, filter_func=add_user_filter)
    db.session.commit()


@user_blueprint.route('/')
def index():
    mark_user_notifications_read(g.user)

    return render_template(
        'users/view.jinja',
        active_page='settings', user=g.user,
        distance_flights=_get_distance_flights(),
        takeoff_locations=_get_takeoff_locations(),
        last_year_statistics=_get_last_year_statistics())


@user_blueprint.route('/change_password', methods=['GET', 'POST'])
def change_password():
    if not g.user.is_writable(g.current_user):
        abort(403)

    form = ChangePasswordForm()
    if form.validate_on_submit():
        return change_password_post(form)

    return render_template('users/change_password.jinja', form=form)


def change_password_post(form):
    g.user.password = <PASSWORD>.data
    g.user.recover_key = None

    db.session.commit()

    return redirect(url_for('.index'))


class DelaySelectField(PropertySingleSelectField):
    def _my_update_params(self, d, nullable=False):
        options = [(0, _('None'))]
        for x in range(1, 10) + range(10, 30, 5) + range(30, 61, 15):
            options.append((x, ngettext(u'%(num)u minute', u'%(num)u minutes', x)))
        d['options'] = options
        return d


def filter_user_id(model):
    return model.id == g.user.id


class EditUserForm(EditableForm):
    __base_widget_type__ = BootstrapForm
    __model__ = User
    __hide_fields__ = ['id']
    __limit_fields__ = ['email_address', 'name',
                        'tracking_delay', 'unit_preset',
                        'distance_unit', 'speed_unit',
                        'lift_unit', 'altitude_unit',
                        'eye_candy']
    __field_widget_args__ = {
        'email_address': dict(label_text=l_('eMail Address')),
        'name': dict(label_text=l_('Name')),
        'tracking_delay': dict(label_text=l_('Tracking Delay')),
        'unit_preset': dict(label_text=l_('Unit Preset')),
        'distance_unit': dict(label_text=l_('Distance Unit')),
        'speed_unit': dict(label_text=l_('Speed Unit')),
        'lift_unit': dict(label_text=l_('Lift Unit')),
        'altitude_unit': dict(label_text=l_('Altitude Unit')),
        'eye_candy': dict(label_text=l_('Eye Candy')),
    }

    email_address = Field(TextField, All(
        Email(not_empty=True),
        UniqueValueUnless(filter_user_id, db.session, __model__, 'email_address')))
    name = Field(TextField, NotEmpty)
    tracking_delay = DelaySelectField
    unit_preset = units.PresetSelectField("unit_preset")
    distance_unit = units.DistanceSelectField
    speed_unit = units.SpeedSelectField
    lift_unit = units.LiftSelectField
    altitude_unit = units.AltitudeSelectField
    eye_candy = Field(CheckBox)

edit_user_form = EditUserForm(db.session)


@user_blueprint.route('/edit')
def edit():
    if not g.user.is_writable(g.current_user):
        abort(403)

    return render_template(
        'generic/form.jinja',
        active_page='settings', title=_('Edit User'),
        form=edit_user_form, values=g.user,
        include_script='users/after-edit-form.jinja')


@user_blueprint.route('/edit', methods=['POST'])
@validate(edit_user_form, edit)
def edit_post():
    if not g.user.is_writable(g.current_user):
        abort(403)

    g.user.email_address = request.form['email_address']
    g.user.name = request.form['name']
    g.user.tracking_delay = request.form.get('tracking_delay', 0)

    unit_preset = request.form.get('unit_preset', 1, type=int)
    if unit_preset == 0:
        g.user.distance_unit = request.form.get('distance_unit', 1, type=int)
        g.user.speed_unit = request.form.get('speed_unit', 1, type=int)
        g.user.lift_unit = request.form.get('lift_unit', 0, type=int)
        g.user.altitude_unit = request.form.get('altitude_unit', 0, type=int)
    else:
        g.user.unit_preset = unit_preset

    g.user.eye_candy = request.form.get('eye_candy', False, type=bool)

    db.session.commit()

    return redirect(url_for('.index'))


@user_blueprint.route('/change_club', methods=['GET', 'POST'])
def change_club():
    if not g.user.is_writable(g.current_user):
        abort(403)

    change_form = ChangeClubForm(club=g.user.club_id)
    create_form = CreateClubForm()

    if request.endpoint.endswith('.change_club'):
        if change_form.validate_on_submit():
            return change_club_post(change_form)

    if request.endpoint.endswith('.create_club'):
        if create_form.validate_on_submit():
            return create_club_post(create_form)

    return render_template(
        'users/change_club.jinja',
        change_form=change_form, create_form=create_form)


@user_blueprint.route('/create_club', methods=['GET', 'POST'])
def create_club():
    return change_club()


def change_club_post(form):
    old_club_id = g.user.club_id
    new_club_id = form.club.data if form.club.data != 0 else None

    if old_club_id == new_club_id:
        return redirect(url_for('.index'))

    g.user.club_id = new_club_id

    create_club_join_event(new_club_id, g.user)

    # assign the user's new club to all of his flights that have
    # no club yet
    flights = Flight.query().join(IGCFile)
    flights = flights.filter(and_(Flight.club_id == None,
                                  or_(Flight.pilot_id == g.user.id,
                                      IGCFile.owner_id == g.user.id)))
    for flight in flights:
        flight.club_id = g.user.club_id

    db.session.commit()

    return redirect(url_for('.index'))


def create_club_post(form):
    club = Club(name=form.name.data)
    club.owner_id = g.current_user.id
    db.session.add(club)

    g.user.club = club

    db.session.commit()

    return redirect(url_for('.index'))


@user_blueprint.route('/follow')
@login_required
def follow():
    Follower.follow(g.current_user, g.user)
    create_follower_notification(g.user, g.current_user)
    db.session.commit()
    return redirect(url_for('.index'))


@user_blueprint.route('/unfollow')
@login_required
def unfollow():
    Follower.unfollow(g.current_user, g.user)
    db.session.commit()
    return redirect(url_for('.index'))


@user_blueprint.route('/tracking_register')
def tracking_register():
    if not g.user.is_writable(g.current_user):
        abort(403)

    g.user.generate_tracking_key()
    db.session.commit()

    return redirect(request.values.get('came_from', '/tracking/info'))


@user_blueprint.route('/recover_password')
def recover_password():
    if not g.current_user or not g.current_user.is_manager():
        abort(403)

    recover_user_password(g.user)
    flash('A password recovery email was sent to that user.')

    db.session.commit()

    return redirect(url_for('.index'))

{'fraction_non_alphanumeric': 0.11596851279167839, 'fraction_numerical': 0.008621497516633868, 'mean_word_length': 3.2204864544196163, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3912296', 'n_tokens_mistral': 7066, 'n_tokens_neox': 6649, 'n_words': 1215}