TestingWebApp/flask_app/app/blueprints/main.py

"""Главный blueprint — посадочная страница, статистика и health-чек."""
from __future__ import annotations

from flask import Blueprint, jsonify, render_template

from .. import db as app_db
from ..auth.decorators import login_required
from ..auth.decorators import current_user as get_current_user

main_bp = Blueprint('main', __name__)


def _get_stats() -> dict:
    """Собирает статистику прохождений для дашборда."""
    from sqlalchemy import func, case
    from ..db import get_session
    from ..models import Department, User, Test, TestAttempt

    session = get_session()
    try:
        total_tests = session.query(func.count(Test.id)).scalar() or 0
        active_tests = session.query(func.count(Test.id)).filter(Test.is_active.is_(True)).scalar() or 0

        total_completed = (
            session.query(func.count(TestAttempt.id))
            .filter(TestAttempt.status == 'completed')
            .scalar() or 0
        )
        total_passed = (
            session.query(func.count(TestAttempt.id))
            .filter(TestAttempt.status == 'completed', TestAttempt.passed.is_(True))
            .scalar() or 0
        )
        total_users = session.query(func.count(User.id)).filter(User.is_active.is_(True)).scalar() or 0

        # Статистика по отделам
        dept_rows = (
            session.query(
                Department.name,
                func.count(TestAttempt.id).label('total'),
                func.sum(
                    case((TestAttempt.passed.is_(True), 1), else_=0)
                ).label('passed_count'),
            )
            .join(User, User.department_id == Department.id)
            .join(TestAttempt, TestAttempt.user_id == User.id)
            .filter(TestAttempt.status == 'completed')
            .group_by(Department.id, Department.name)
            .order_by(Department.name)
            .all()
        )

        from ..models import TestVersion
        recent_rows = (
            session.query(
                TestAttempt.started_at,
                TestAttempt.completed_at,
                TestAttempt.passed,
                TestAttempt.correct_count,
                TestAttempt.total_questions,
                User.full_name,
                User.login,
                Test.title.label('test_title'),
            )
            .join(User, User.id == TestAttempt.user_id)
            .join(TestVersion, TestVersion.id == TestAttempt.test_version_id)
            .join(Test, Test.id == TestVersion.test_id)
            .filter(TestAttempt.status == 'completed')
            .order_by(TestAttempt.completed_at.desc())
            .limit(10)
            .all()
        )

        dept_stats = [
            {
                'name': r.name,
                'total': r.total,
                'passed': int(r.passed_count or 0),
                'rate': round(100 * int(r.passed_count or 0) / r.total) if r.total else 0,
            }
            for r in dept_rows
        ]

        recent_list = [
            {
                'user': r.full_name or r.login,
                'test': r.test_title,
                'passed': r.passed,
                'score': f'{r.correct_count}/{r.total_questions}' if r.correct_count is not None else '—',
                'at': r.completed_at.strftime('%d.%m %H:%M') if r.completed_at else '—',
            }
            for r in recent_rows
        ]

        pass_rate = round(100 * total_passed / total_completed) if total_completed else 0

        return {
            'total_tests': total_tests,
            'active_tests': active_tests,
            'total_completed': total_completed,
            'total_passed': total_passed,
            'total_users': total_users,
            'pass_rate': pass_rate,
            'dept_stats': dept_stats,
            'recent': recent_list,
        }
    except Exception:
        return {
            'total_tests': 0, 'active_tests': 0, 'total_completed': 0,
            'total_passed': 0, 'total_users': 0, 'pass_rate': 0,
            'dept_stats': [], 'recent': [],
        }


@main_bp.route('/health')
def health():
    """Smoke-проверка приложения и подключений к БД (без авторизации)."""
    db_status = app_db.ping()
    overall = 'ok' if db_status.get('main') == 'ok' else 'degraded'
    return jsonify(
        status=overall,
        service='testing-flask-app',
        db=db_status,
    )


@main_bp.route('/')
@login_required
def index():
    return render_template('index.html')


@main_bp.route('/stats')
@login_required
def stats_page():
    return render_template('stats.html', stats=_get_stats())


@main_bp.route('/assignments')
@login_required
def assignments_page():
    from ..db import get_session
    from ..models import Test, TestVersion
    from sqlalchemy.orm import selectinload
    session = get_session()
    tests = (
        session.query(Test)
        .join(TestVersion, (TestVersion.test_id == Test.id) & TestVersion.is_active.is_(True))
        .filter(Test.is_active.is_(True))
        .order_by(Test.title)
        .all()
    )
    return render_template('assignments.html', tests=[
        {'id': str(t.id), 'title': t.title or '(без названия)'}
        for t in tests
    ])