av_Site_repetitor/app.py

import csv
import io
import os
import sqlite3
from datetime import datetime
from functools import wraps
from hmac import compare_digest
from pathlib import Path

from flask import Flask, flash, make_response, redirect, render_template, request, session, url_for
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_wtf import CSRFProtect


BASE_DIR = Path(__file__).resolve().parent
DATABASE_PATH = BASE_DIR / "consultations.db"
SUBJECTS = {"math": "Математика", "physics": "Физика"}
EXAMS = {"oge": "ОГЭ", "ege": "ЕГЭ", "other": "Пока не определились"}
FORMATS = {"online": "Онлайн", "offline": "Очно", "other": "Пока не определились"}
PREFERRED_TIMES = {
    "flexible": "Гибко",
    "morning": "Утро",
    "day": "День",
    "evening": "Вечер",
}
REQUEST_STATUSES = {
    "new": "Новая",
    "contacted": "Связались",
    "scheduled": "Запланирована",
    "closed": "Завершена",
}

app = Flask(__name__)
app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY")
if not app.config["SECRET_KEY"]:
    raise RuntimeError("SECRET_KEY не задан. Установите переменную окружения SECRET_KEY.")
app.config["SESSION_COOKIE_HTTPONLY"] = True
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
app.config["SESSION_COOKIE_SECURE"] = os.environ.get("COOKIE_SECURE", "0") == "1"

csrf = CSRFProtect(app)

limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per hour"],
    storage_uri="memory://",
)


def get_connection():
    connection = sqlite3.connect(DATABASE_PATH)
    connection.row_factory = sqlite3.Row
    return connection


def init_database():
    with get_connection() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS consultation_requests (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                contact TEXT NOT NULL,
                subject TEXT NOT NULL,
                exam TEXT NOT NULL,
                format TEXT NOT NULL,
                preferred_time TEXT NOT NULL DEFAULT 'flexible',
                message TEXT,
                status TEXT NOT NULL DEFAULT 'new',
                admin_note TEXT NOT NULL DEFAULT '',
                created_at TEXT NOT NULL,
                updated_at TEXT
            )
            """
        )
        existing_columns = {
            row["name"]
            for row in connection.execute("PRAGMA table_info(consultation_requests)").fetchall()
        }
        migrations = {
            "preferred_time": "TEXT NOT NULL DEFAULT 'flexible'",
            "status": "TEXT NOT NULL DEFAULT 'new'",
            "admin_note": "TEXT NOT NULL DEFAULT ''",
            "updated_at": "TEXT",
        }
        for column, definition in migrations.items():
            if column not in existing_columns:
                connection.execute(
                    f"ALTER TABLE consultation_requests ADD COLUMN {column} {definition}"
                )


@app.context_processor
def inject_site_data():
    return {"current_year": datetime.now().year}


@app.get("/")
def index():
    return render_template("index.html")


@app.post("/submit")
@limiter.limit("10 per minute")
def submit_request():
    name = request.form.get("name", "").strip()
    contact = request.form.get("contact", "").strip()
    subject_key = request.form.get("subject", "").strip()
    exam_key = request.form.get("exam", "").strip()
    format_key = request.form.get("format", "").strip()
    preferred_time_key = request.form.get("preferred_time", "").strip()
    message = request.form.get("message", "").strip()[:2000]

    if (
        not name
        or not contact
        or subject_key not in SUBJECTS
        or exam_key not in EXAMS
        or format_key not in FORMATS
        or preferred_time_key not in PREFERRED_TIMES
    ):
        flash("Пожалуйста, заполните обязательные поля формы.", "error")
        return redirect(url_for("index") + "#contact")

    created_at = datetime.now().isoformat(timespec="minutes")
    with get_connection() as connection:
        connection.execute(
            """
            INSERT INTO consultation_requests
                (name, contact, subject, exam, format, preferred_time, message, status, admin_note, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                name[:120],
                contact[:160],
                SUBJECTS[subject_key],
                EXAMS[exam_key],
                FORMATS[format_key],
                PREFERRED_TIMES[preferred_time_key],
                message,
                "new",
                "",
                created_at,
                created_at,
            ),
        )

    return redirect(url_for("success", name=name))


@app.get("/success")
def success():
    return render_template("success.html", name=request.args.get("name", ""))


def admin_required(view):
    @wraps(view)
    def wrapped_view(*args, **kwargs):
        if not session.get("is_admin"):
            return redirect(url_for("admin_login"))
        return view(*args, **kwargs)

    return wrapped_view


def get_request_filters():
    status = request.args.get("status", "").strip()
    subject = request.args.get("subject", "").strip()
    search = request.args.get("q", "").strip()

    return {
        "status": status if status in REQUEST_STATUSES else "",
        "subject": subject if subject in SUBJECTS.values() else "",
        "q": search[:100],
    }


def build_request_query(filters):
    conditions = []
    parameters = []

    if filters["status"]:
        conditions.append("status = ?")
        parameters.append(filters["status"])
    if filters["subject"]:
        conditions.append("subject = ?")
        parameters.append(filters["subject"])
    if filters["q"]:
        escaped = filters["q"].replace("%", "\\%").replace("_", "\\_")
        search_pattern = f"%{escaped}%"
        conditions.append(
            "(name LIKE ? ESCAPE '\\' OR contact LIKE ? ESCAPE '\\' "
            "OR message LIKE ? ESCAPE '\\' OR admin_note LIKE ? ESCAPE '\\')"
        )
        parameters.extend([search_pattern] * 4)

    where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""
    return where_clause, parameters


def fetch_requests(connection, filters):
    where_clause, parameters = build_request_query(filters)
    return connection.execute(
        f"""
        SELECT id, name, contact, subject, exam, format, preferred_time,
               message, status, admin_note, created_at, updated_at
        FROM consultation_requests
        {where_clause}
        ORDER BY id DESC
        """,
        parameters,
    ).fetchall()


def fetch_stats(connection):
    stats = connection.execute(
        """
        SELECT
            COUNT(*) AS total,
            COALESCE(SUM(CASE WHEN status = 'new' THEN 1 ELSE 0 END), 0) AS new,
            COALESCE(SUM(CASE WHEN status = 'contacted' THEN 1 ELSE 0 END), 0) AS contacted,
            COALESCE(SUM(CASE WHEN status = 'scheduled' THEN 1 ELSE 0 END), 0) AS scheduled,
            COALESCE(SUM(CASE WHEN status = 'closed' THEN 1 ELSE 0 END), 0) AS closed
        FROM consultation_requests
        """
    ).fetchone()
    return dict(stats)


@app.route("/admin/login", methods=["GET", "POST"])
@limiter.limit("10 per minute")
def admin_login():
    if session.get("is_admin"):
        return redirect(url_for("admin_dashboard"))

    if request.method == "POST":
        password = request.form.get("password", "")
        admin_password = os.environ.get("ADMIN_PASSWORD")
        if not admin_password:
            raise RuntimeError("ADMIN_PASSWORD не задан. Установите переменную окружения.")

        if compare_digest(password, admin_password):
            session.clear()
            session["is_admin"] = True
            return redirect(url_for("admin_dashboard"))

        flash("Неверный пароль.", "error")

    return render_template("admin_login.html")


@app.get("/admin")
@admin_required
def admin_dashboard():
    filters = get_request_filters()
    with get_connection() as connection:
        consultation_requests = fetch_requests(connection, filters)
        stats = fetch_stats(connection)

    return render_template(
        "admin.html",
        consultation_requests=consultation_requests,
        filters=filters,
        stats=stats,
        statuses=REQUEST_STATUSES,
        subjects=SUBJECTS.values(),
    )


@app.post("/admin/request/<int:request_id>/status")
@admin_required
def admin_update_status(request_id):
    status = request.form.get("status", "").strip()
    if status not in REQUEST_STATUSES:
        flash("Неизвестный статус заявки.", "error")
        return redirect(url_for("admin_dashboard"))

    with get_connection() as connection:
        connection.execute(
            "UPDATE consultation_requests SET status = ?, updated_at = ? WHERE id = ?",
            (status, datetime.now().isoformat(timespec="minutes"), request_id),
        )

    flash("Статус заявки обновлён.", "success")
    return redirect(url_for("admin_dashboard"))


@app.post("/admin/request/<int:request_id>/note")
@admin_required
def admin_update_note(request_id):
    admin_note = request.form.get("admin_note", "").strip()[:2000]
    with get_connection() as connection:
        connection.execute(
            "UPDATE consultation_requests SET admin_note = ?, updated_at = ? WHERE id = ?",
            (admin_note, datetime.now().isoformat(timespec="minutes"), request_id),
        )

    flash("Заметка сохранена.", "success")
    return redirect(url_for("admin_dashboard"))


@app.post("/admin/request/<int:request_id>/delete")
@admin_required
def admin_delete_request(request_id):
    with get_connection() as connection:
        connection.execute("DELETE FROM consultation_requests WHERE id = ?", (request_id,))

    flash("Заявка удалена.", "success")
    return redirect(url_for("admin_dashboard"))


def csv_safe(value):
    """Защита от CSV-инъекции: значения с формулами экранируются апострофом."""
    if value and value[0] in ("=", "+", "-", "@", "\t", "\r"):
        return "'" + value
    return value


@app.get("/admin/export.csv")
@admin_required
def admin_export():
    filters = get_request_filters()
    with get_connection() as connection:
        consultation_requests = fetch_requests(connection, filters)

    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(
        [
            "Номер",
            "Имя",
            "Контакт",
            "Предмет",
            "Экзамен",
            "Формат",
            "Желаемое время",
            "Статус",
            "Комментарий",
            "Заметка преподавателя",
            "Создана",
            "Обновлена",
        ]
    )
    for consultation in consultation_requests:
        writer.writerow(
            [
                consultation["id"],
                csv_safe(consultation["name"]),
                csv_safe(consultation["contact"]),
                consultation["subject"],
                consultation["exam"],
                consultation["format"],
                consultation["preferred_time"],
                REQUEST_STATUSES.get(consultation["status"], consultation["status"]),
                csv_safe(consultation["message"] or ""),
                csv_safe(consultation["admin_note"] or ""),
                consultation["created_at"],
                consultation["updated_at"] or "",
            ]
        )

    response = make_response("\ufeff" + output.getvalue())
    response.headers["Content-Type"] = "text/csv; charset=utf-8"
    response.headers["Content-Disposition"] = (
        f"attachment; filename=consultations-{datetime.now().strftime('%Y-%m-%d')}.csv"
    )
    return response


@app.get("/admin/logout")
def admin_logout():
    session.clear()
    return redirect(url_for("admin_login"))


init_database()


if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port=5000,
        debug=os.environ.get("FLASK_DEBUG") == "1",
    )