Assemblr-Backend/routes/sipro.py

import datetime
from enum import StrEnum
from flask import Blueprint, request, jsonify
from sqlalchemy import func, cast, String, Numeric, Float

import auxiliary
import database
import database.enums

sipro_blueprint = Blueprint('sipro', __name__)


class ExpandParam(StrEnum):
    USER = 'user'


@sipro_blueprint.before_request
def auth():
    API_KEY = '5D809ED08080B5F204443B31374BD6A5'
    if request.headers.get('Authorization') != API_KEY:
        return {'error': 'Unauthorized'}, 401


@sipro_blueprint.post('/assemblyInfo')
def assembly_info():
    params: dict = request.json
    expand_param = params.get('expand')
    order_ids = params.get('orderIds')
    expand_list = expand_param if expand_param else []
    datetime_format = 'YYYY.MM.DD, HH24:MI'
    entity_list = [
        database.Assembly.id.label('id'),
        database.Assembly.order_id.label('order_id'),
        func.to_char(database.Assembly.created_at, datetime_format).label('created_at'),
        func.to_char(database.Assembly.ended_at, datetime_format).label('ended_at'),
        database.Assembly.is_active.label('is_active'),
        database.Assembly.state.label('state')
    ]
    query = database.Assembly.query
    for expand in expand_list:
        match expand:
            case ExpandParam.USER:
                query = query.join(database.User)
                entity_list.extend([
                    database.User.id.label('user.id'),
                    database.User.login.label('user.login'),
                ])

    if order_ids:
        query = query.filter(database.Assembly.order_id.in_(order_ids))
    query = query.with_entities(*entity_list)
    result = query.all()
    json_result = [auxiliary.to_nested_dict(row) for row in result]
    return json_result


@sipro_blueprint.post('/tableAssemblyStats')
def table_assembly_stats():
    filters = request.json
    date_from = filters.get('from')
    date_to = filters.get('to')

    dated_query = (
        database.Assembly.query
        .join(database.User)
        .filter(
            database.User.is_admin == False
        )
        .group_by(database.User.login, database.Assembly.user_id)
        .with_entities(
            database.User.login.label('user_login'),
            func.count(database.Assembly.user_id).label('assembled'),
            func.string_agg(func.cast(database.Assembly.order_id, database.db.String), ',').label('order_ids')
        )
    )

    if len(date_from) > 0 and len(date_to) > 0:
        dated_query = dated_query.filter(
            database.Assembly.ended_at >= date_from,
            database.Assembly.ended_at <= date_to
        )

    json_result = []
    for row in dated_query.all():
        json_result.append({
            'user_login': row.user_login,
            'assembled': row.assembled,
            'order_ids': row.order_ids
        })

    return jsonify(json_result)


@sipro_blueprint.get('/dailyAssemblyData')
def daily_assembly_stats():
    query = (
        database.Assembly.query
        .join(database.User)
        .filter(
            database.Assembly.ended_at >= func.current_date()
        )
        .group_by(database.User.login, database.Assembly.user_id)
        .with_entities(
            database.User.login.label('user_login'),
            func.count(database.Assembly.user_id).label('assembled'),
            func.string_agg(func.cast(database.Assembly.order_id, database.db.String), ',').label('order_ids')
        )
    )

    json_result = []
    for row in query.all():
        json_result.append({
            'user_login': row.user_login,
            'assembled': row.assembled,
            'order_ids': row.order_ids
        })

    return json_result


@sipro_blueprint.get('/users')
def get_users():
    query = (database.User.query.
             with_entities(database.User.id,
                           database.User.login)
             .all())
    response = [{'id': user.id, 'login': user.login} for user in query]
    return jsonify(response)


@sipro_blueprint.get('/statistics')
def get_users_statistics():
    data: dict = request.args
    user_id = int(data['userId'])
    date_from = datetime.datetime.fromisoformat(data['dateFrom'])
    date_to = datetime.datetime.fromisoformat(data['dateTo']) + datetime.timedelta(hours=24, minutes=59, seconds=59)
    query = (database.BalanceTransaction.query
             .filter(database.BalanceTransaction.user_id == user_id,
                     database.BalanceTransaction.created_at.between(date_from, date_to))
             .order_by(func.date_trunc('day', database.BalanceTransaction.created_at))
             .group_by(func.date_trunc('day', database.BalanceTransaction.created_at))
             .with_entities(func.date_trunc('day', database.BalanceTransaction.created_at).label('date'),
                            func.cast(func.round(func.cast(func.sum(database.BalanceTransaction.amount), Numeric), 2),
                                      Float).label(
                                'value'),
                            func.string_agg(cast(database.BalanceTransaction.json_data['order_id'], String), ',').label(
                                'order_ids'))
             .all())
    result = [{'date': row.date.isoformat(), 'value': row.value, 'order_ids': row.order_ids} for row in query]
    return jsonify(result)


@sipro_blueprint.get('/get-orders-by-date')
def get_orders_by_date():
    data: dict = request.args
    date = datetime.date.fromisoformat(data['date'])
    query = (
        database.Assembly.query
        .filter(
            func.date_trunc('day', database.Assembly.ended_at) == func.date_trunc('day', date),
        )
        .with_entities(
            func.string_agg(func.cast(database.Assembly.order_id, database.db.String), ',').label('order_ids')
        )
        .scalar()
    )
    if not query:
        return jsonify({'order_ids': []})
    return jsonify({'order_ids': list(map(int, query.split(',')))})