387 lines
12 KiB
Python
387 lines
12 KiB
Python
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": "Математика"}
|
|
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",
|
|
)
|