first commit
This commit is contained in:
greyjoy
@@ -0,0 +1,352 @@
|
||||
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
|
||||
|
||||
|
||||
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.getenv("SECRET_KEY", "change-this-secret-key")
|
||||
app.config["SESSION_COOKIE_HTTPONLY"] = True
|
||||
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
|
||||
|
||||
|
||||
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")
|
||||
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"]:
|
||||
conditions.append("(name LIKE ? OR contact LIKE ? OR message LIKE ? OR admin_note LIKE ?)")
|
||||
search_pattern = f"%{filters['q']}%"
|
||||
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"])
|
||||
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.getenv("ADMIN_PASSWORD", "admin123")
|
||||
|
||||
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"))
|
||||
|
||||
|
||||
@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"],
|
||||
consultation["name"],
|
||||
consultation["contact"],
|
||||
consultation["subject"],
|
||||
consultation["exam"],
|
||||
consultation["format"],
|
||||
consultation["preferred_time"],
|
||||
REQUEST_STATUSES.get(consultation["status"], consultation["status"]),
|
||||
consultation["message"] or "",
|
||||
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(debug=True)
|
||||
Reference in New Issue
Block a user