from fastapi import FastAPI, Request, Form, UploadFile, File, Body, HTTPException from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from starlette.middleware.sessions import SessionMiddleware from passlib.context import CryptContext from pydantic import BaseModel import sqlite3, pathlib, os import re from typing import Optional from fastapi import Query from datetime import datetime, timedelta from typing import Optional, Dict, Any, Annotated from fastapi.responses import JSONResponse from pydantic import BaseModel, Field, StringConstraints import datetime as dt from fastapi import Query from typing import Optional, List from fastapi import HTTPException import hashlib import hmac from typing import List import secrets, time import math import json, secrets as _secrets, string, urllib.parse import datetime as _dt from urllib.parse import urlparse app = FastAPI(title="TheL4FPRO") BASE_DIR = pathlib.Path(__file__).resolve().parent DB_PATH = BASE_DIR / "db.sqlite3" IS_PROD = os.getenv("APP_ENV", "dev").lower() in ("prod", "production") SECRET_KEY = os.getenv("APP_SECRET_KEY") or secrets.token_urlsafe(64) app.add_middleware( SessionMiddleware, secret_key=SECRET_KEY, same_site="lax", https_only=IS_PROD, max_age=60*60*24*7, session_cookie="thelfpro_session" ) pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def rotate_session(request: Request) -> None: request.session.clear() request.session["sid"] = secrets.token_urlsafe(32) request.session["csrf"] = secrets.token_urlsafe(32) request.session["created_at"] = int(time.time()) app.mount("/static", StaticFiles(directory="app/static"), name="static") templates = Jinja2Templates(directory="app/templates") def require_csrf(request: Request, token: str | None): sess = request.session.get("csrf") if not sess or not token or not hmac.compare_digest(str(sess), str(token or "")): raise HTTPException(status_code=403, detail="CSRF token inválido") @app.get("/contabilidad", response_class=HTMLResponse) async def contabilidad(request: Request): require_admin(request) user = current_user(request) return templates.TemplateResponse("contabilidad.html", {"request": request, "user": user}) UNSAFE_METHODS = {"POST", "PUT", "PATCH", "DELETE"} def _enforce_csrf_if_unsafe(request: Request): if request.method in UNSAFE_METHODS: token = request.headers.get("X-CSRF-Token") require_csrf(request, token) def require_admin(request: Request): me = current_user(request) if not me: raise HTTPException(status_code=401, detail="No autenticado") _enforce_csrf_if_unsafe(request) role = (me["role"] if isinstance(me, sqlite3.Row) else str(me.get("role", ""))) or "" if str(role).lower() != "admin": raise HTTPException(status_code=403, detail="Solo administradores") return me def require_docente_or_admin(request: Request): me = current_user(request) if not me: raise HTTPException(status_code=401, detail="No autenticado") _enforce_csrf_if_unsafe(request) role = (me["role"] if isinstance(me, sqlite3.Row) else str(me.get("role", ""))) or "" if str(role).lower() not in ("docente", "admin"): raise HTTPException(status_code=403, detail="Solo docentes o admin") return me def get_conn(): conn = sqlite3.connect(DB_PATH, isolation_level=None, check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON;") conn.execute("PRAGMA journal_mode = WAL;") conn.execute("PRAGMA synchronous = NORMAL;") conn.execute("PRAGMA busy_timeout = 5000;") conn.execute("PRAGMA cache_size = -20000;") conn.execute("PRAGMA temp_store = MEMORY;") try: conn.execute("PRAGMA mmap_size = 268435456;") except Exception: pass return conn def _conn(): return get_conn() def _table_columns(conn, table): try: return {r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()} except Exception: return set() def init_db(): with _conn() as conn: c = conn.cursor() c.execute("PRAGMA foreign_keys=ON") c.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT NOT NULL UNIQUE, name TEXT NOT NULL, first_name TEXT, last_name TEXT, phone TEXT, password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'alumno', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S','now','localtime')) ) """) cols_users = {r[1] for r in c.execute("PRAGMA table_info(users)").fetchall()} if "first_name" not in cols_users: c.execute("ALTER TABLE users ADD COLUMN first_name TEXT") if "last_name" not in cols_users: c.execute("ALTER TABLE users ADD COLUMN last_name TEXT") if "phone" not in cols_users: c.execute("ALTER TABLE users ADD COLUMN phone TEXT") c.execute(""" CREATE TABLE IF NOT EXISTS user_profile ( user_id INTEGER PRIMARY KEY, phone TEXT, avatar TEXT, city TEXT, dni TEXT, birth_date TEXT, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS user_sports ( user_id INTEGER PRIMARY KEY, team TEXT, category TEXT, position TEXT, dominant_foot TEXT, strengths TEXT, weaknesses TEXT, injury_history TEXT, training_type TEXT, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS user_health ( user_id INTEGER PRIMARY KEY, physical_work INTEGER, physical_work_details TEXT, smoking INTEGER, alcohol INTEGER, recovery TEXT, chest_pain INTEGER, discomfort INTEGER, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS user_consent ( user_id INTEGER PRIMARY KEY, whatsapp_content INTEGER, video_permission INTEGER, privacy_acceptance INTEGER, data_confirmation INTEGER, agreement INTEGER, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS user_guardians ( user_id INTEGER PRIMARY KEY, occupation TEXT, study_place TEXT, parent_name TEXT, parent_email TEXT, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS user_marketing ( user_id INTEGER PRIMARY KEY, found_us TEXT, enjoyment TEXT, nerves_confidence TEXT, additional_comments TEXT, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS user_finance ( user_id INTEGER PRIMARY KEY, matricula_eur REAL NOT NULL DEFAULT 0, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS reservations ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S','now','localtime')), date TEXT NOT NULL, time TEXT, duration_minutes INTEGER, price_cents INTEGER NOT NULL DEFAULT 0, name TEXT, email TEXT, phone TEXT, notes TEXT, user_id INTEGER, status TEXT NOT NULL DEFAULT 'confirmada', docente_id INTEGER NOT NULL, paid INTEGER NOT NULL DEFAULT 0, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE SET NULL, FOREIGN KEY(docente_id) REFERENCES users(id) ON DELETE CASCADE ) """) cols_res = {r[1] for r in c.execute("PRAGMA table_info(reservations)").fetchall()} if "created_at" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN created_at TEXT") c.execute("UPDATE reservations SET created_at = strftime('%Y-%m-%dT%H:%M:%S','now','localtime') WHERE created_at IS NULL") if "time" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN time TEXT") if "duration_minutes" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN duration_minutes INTEGER") if "price_cents" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN price_cents INTEGER NOT NULL DEFAULT 0") if "name" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN name TEXT") if "email" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN email TEXT") if "phone" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN phone TEXT") if "notes" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN notes TEXT") if "user_id" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN user_id INTEGER") if "status" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN status TEXT NOT NULL DEFAULT 'confirmada'") if "docente_id" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN docente_id INTEGER NOT NULL DEFAULT 1") if "paid" not in cols_res: c.execute("ALTER TABLE reservations ADD COLUMN paid INTEGER NOT NULL DEFAULT 0") c.execute(""" CREATE TABLE IF NOT EXISTS reservation_students ( reservation_id INTEGER NOT NULL, user_id INTEGER NOT NULL, PRIMARY KEY (reservation_id, user_id), FOREIGN KEY(reservation_id) REFERENCES reservations(id) ON DELETE CASCADE, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS jugadores_gastos ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S','now','localtime')), date TEXT NOT NULL, user_id INTEGER, reservation_id INTEGER, amount_cents INTEGER NOT NULL, category TEXT, concept TEXT, notes TEXT, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE, FOREIGN KEY(reservation_id) REFERENCES reservations(id) ON DELETE CASCADE ) """) c.execute("CREATE INDEX IF NOT EXISTS idx_jg_date ON jugadores_gastos(date)") c.execute("CREATE INDEX IF NOT EXISTS idx_jg_user ON jugadores_gastos(user_id)") c.execute("CREATE INDEX IF NOT EXISTS idx_jg_reservation ON jugadores_gastos(reservation_id)") c.execute(""" CREATE TABLE IF NOT EXISTS notifications ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S','now','localtime')), title TEXT, body TEXT, sender_id INTEGER, recipient_id INTEGER NOT NULL, reservation_id INTEGER, is_read INTEGER NOT NULL DEFAULT 0, deleted_by_sender INTEGER NOT NULL DEFAULT 0, deleted_by_recipient INTEGER NOT NULL DEFAULT 0, FOREIGN KEY(sender_id) REFERENCES users(id) ON DELETE SET NULL, FOREIGN KEY(recipient_id) REFERENCES users(id) ON DELETE CASCADE, FOREIGN KEY(reservation_id) REFERENCES reservations(id) ON DELETE CASCADE ) """) c.execute("CREATE INDEX IF NOT EXISTS idx_notif_rec ON notifications(recipient_id, is_read)") c.execute("CREATE INDEX IF NOT EXISTS idx_notif_send ON notifications(sender_id)") c.execute("CREATE INDEX IF NOT EXISTS idx_notif_created ON notifications(created_at)") c.execute(""" CREATE TABLE IF NOT EXISTS player_ratings ( id INTEGER PRIMARY KEY AUTOINCREMENT, player_id INTEGER NOT NULL, docente_id INTEGER NOT NULL, rating INTEGER NOT NULL, notes TEXT, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S','now','localtime')), FOREIGN KEY(player_id) REFERENCES users(id) ON DELETE CASCADE, FOREIGN KEY(docente_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS valoraciones_jugador ( id INTEGER PRIMARY KEY AUTOINCREMENT, eval_id TEXT NOT NULL, player_id INTEGER NOT NULL, docente_id INTEGER NOT NULL, area TEXT NOT NULL, subescala TEXT NOT NULL, indicador TEXT NOT NULL, score INTEGER NOT NULL, observaciones TEXT, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S','now','localtime')), FOREIGN KEY(player_id) REFERENCES users(id) ON DELETE CASCADE, FOREIGN KEY(docente_id) REFERENCES users(id) ON DELETE CASCADE ) """) c.execute(""" CREATE TABLE IF NOT EXISTS "formulario-registro" ( id INTEGER PRIMARY KEY AUTOINCREMENT, pin TEXT NOT NULL UNIQUE, email TEXT, name TEXT, used INTEGER NOT NULL DEFAULT 0, expires_at TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S','now','localtime')) ) """) c.execute('CREATE INDEX IF NOT EXISTS idx_formreg_used_exp ON "formulario-registro"(used, expires_at)') c.execute("INSERT OR IGNORE INTO user_profile(user_id) SELECT id FROM users") c.execute("INSERT OR IGNORE INTO user_sports(user_id) SELECT id FROM users") c.execute("INSERT OR IGNORE INTO user_health(user_id) SELECT id FROM users") c.execute("INSERT OR IGNORE INTO user_consent(user_id) SELECT id FROM users") c.execute("INSERT OR IGNORE INTO user_guardians(user_id)SELECT id FROM users") c.execute("INSERT OR IGNORE INTO user_marketing(user_id)SELECT id FROM users") c.execute("INSERT OR IGNORE INTO user_finance(user_id, matricula_eur) SELECT id, 0.0 FROM users") conn.commit() def count_admins() -> int: with _conn() as conn: row = conn.execute("SELECT COUNT(1) AS c FROM users WHERE role='admin'").fetchone() return int(row["c"] or 0) def is_admin(me) -> bool: return str(me["role"]) == "admin" def _upsert_satellites_empty(conn, user_id: int): conn.execute("INSERT OR IGNORE INTO user_profile(user_id) VALUES (?)", (user_id,)) conn.execute("INSERT OR IGNORE INTO user_sports(user_id) VALUES (?)", (user_id,)) conn.execute("INSERT OR IGNORE INTO user_health(user_id) VALUES (?)", (user_id,)) conn.execute("INSERT OR IGNORE INTO user_consent(user_id) VALUES (?)", (user_id,)) conn.execute("INSERT OR IGNORE INTO user_guardians(user_id) VALUES (?)", (user_id,)) conn.execute("INSERT OR IGNORE INTO user_marketing(user_id) VALUES (?)", (user_id,)) conn.execute("INSERT OR IGNORE INTO user_finance(user_id, matricula_eur) VALUES (?, 0.0)", (user_id,)) @app.post("/admin/users/delete", name="admin_delete_user") async def admin_delete_user( request: Request, user_id: int = Form(...), next_url: str | None = Form(None), ): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) if not is_admin(me): raise HTTPException(status_code=403, detail="Solo un admin puede eliminar usuarios.") target = get_user_by_id(user_id) if not target: raise HTTPException(status_code=404, detail="Usuario no encontrado.") if int(me["id"]) == int(target["id"]): raise HTTPException(status_code=400, detail="No puedes eliminar tu propia cuenta.") with _conn() as conn: try: conn.execute("BEGIN") conn.execute("DELETE FROM notifications WHERE sender_id = ? OR recipient_id = ?", (user_id, user_id)) conn.execute("DELETE FROM reservation_students WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM reservations WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) conn.commit() except Exception: conn.rollback() raise if next_url: p = urlparse(next_url) if p.scheme == "" and p.netloc == "": return RedirectResponse(url=next_url, status_code=303) return RedirectResponse(url="/admin", status_code=303) def get_user_by_email(email: str): with _conn() as conn: cur = conn.execute(""" SELECT u.id,u.name,u.email,u.password_hash,u.created_at,u.role,u.first_name,u.last_name, p.phone,p.avatar,p.city,p.dni,p.birth_date, s.team,s.category,s.position,s.dominant_foot,s.strengths,s.weaknesses,s.injury_history,s.training_type, h.physical_work,h.physical_work_details,h.smoking,h.alcohol,h.recovery,h.chest_pain,h.discomfort, c.whatsapp_content,c.video_permission,c.privacy_acceptance,c.data_confirmation,c.agreement, g.occupation,g.study_place,g.parent_name,g.parent_email, m.found_us,m.enjoyment,m.nerves_confidence,m.additional_comments, COALESCE(f.matricula_eur,0.0) AS matricula_eur, CAST(ROUND(COALESCE(f.matricula_eur,0.0)*100) AS INTEGER) AS matricula_cents FROM users u LEFT JOIN user_profile p ON p.user_id=u.id LEFT JOIN user_sports s ON s.user_id=u.id LEFT JOIN user_health h ON h.user_id=u.id LEFT JOIN user_consent c ON c.user_id=u.id LEFT JOIN user_guardians g ON g.user_id=u.id LEFT JOIN user_marketing m ON m.user_id=u.id LEFT JOIN user_finance f ON f.user_id=u.id WHERE LOWER(u.email)=LOWER(?) """, (email,)) return cur.fetchone() def get_user_by_id(user_id: int): with _conn() as conn: cur = conn.execute(""" SELECT u.id,u.name,u.email,u.password_hash,u.created_at,u.role,u.first_name,u.last_name, p.phone,p.avatar,p.city,p.dni,p.birth_date, s.team,s.category,s.position,s.dominant_foot,s.strengths,s.weaknesses,s.injury_history,s.training_type, h.physical_work,h.physical_work_details,h.smoking,h.alcohol,h.recovery,h.chest_pain,h.discomfort, c.whatsapp_content,c.video_permission,c.privacy_acceptance,c.data_confirmation,c.agreement, g.occupation,g.study_place,g.parent_name,g.parent_email, m.found_us,m.enjoyment,m.nerves_confidence,m.additional_comments, COALESCE(f.matricula_eur,0.0) AS matricula_eur, CAST(ROUND(COALESCE(f.matricula_eur,0.0)*100) AS INTEGER) AS matricula_cents FROM users u LEFT JOIN user_profile p ON p.user_id=u.id LEFT JOIN user_sports s ON s.user_id=u.id LEFT JOIN user_health h ON h.user_id=u.id LEFT JOIN user_consent c ON c.user_id=u.id LEFT JOIN user_guardians g ON g.user_id=u.id LEFT JOIN user_marketing m ON m.user_id=u.id LEFT JOIN user_finance f ON f.user_id=u.id WHERE u.id=? """, (user_id,)) return cur.fetchone() def create_user(name: str, email: str, password: str, role: str = "alumno", first_name: str | None = None, last_name: str | None = None): password_hash = pwd_context.hash(password) with _conn() as conn: cols = _table_columns(conn, "users") created_at = dt.datetime.utcnow().isoformat() if "first_name" in cols and "last_name" in cols: cur = conn.execute( "INSERT INTO users (name, email, password_hash, created_at, role, first_name, last_name) " "VALUES (?,?,?,?,?,?,?)", (name, email, password_hash, created_at, role, first_name, last_name), ) else: cur = conn.execute( "INSERT INTO users (name, email, password_hash, created_at, role) VALUES (?,?,?,?,?)", (name, email, password_hash, created_at, role), ) uid = cur.lastrowid _upsert_satellites_empty(conn, uid) conn.commit() @app.get("/api/notifications/unread", response_class=JSONResponse) async def api_notifications_unread(request: Request): me = current_user(request) if not me: raise HTTPException(status_code=401, detail="No autenticado") rows = get_unread_notifications_for_user(int(me["id"])) def to_dict(r): return {k: r[k] for k in r.keys()} return [to_dict(r) for r in rows] @app.post("/api/notifications/{notification_id}/mark-read", response_class=JSONResponse) async def api_notifications_mark_read(request: Request, notification_id: int): me = current_user(request) if not me: raise HTTPException(status_code=401, detail="No autenticado") token = request.headers.get("X-CSRF-Token") require_csrf(request, token) with _conn() as conn: conn.execute( """ UPDATE notifications SET is_read = 1 WHERE id = ? AND recipient_id = ? """, (notification_id, int(me["id"])), ) conn.commit() return {"ok": True} @app.get("/api/reservations/{res_id}", response_class=JSONResponse) async def api_reservation_get(request: Request, res_id: int): me = current_user(request) if not me: raise HTTPException(status_code=401, detail="No autenticado") with _conn() as conn: row = conn.execute( """ SELECT id, created_at, date, time, duration_minutes, price_cents, name, email, phone, notes, user_id, status, docente_id, paid FROM reservations WHERE id = ? """, (res_id,) ).fetchone() if not row: raise HTTPException(status_code=404, detail="No encontrada") data = _reservation_row_to_dict(row) return JSONResponse(content=data) @app.get("/api/reservations/{res_id}/students", response_model=dict) async def api_reservation_students(request: Request, res_id: int): me = current_user(request) if not me: raise HTTPException(status_code=401, detail="No autenticado") with _conn() as conn: conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT u.id, u.name, u.email FROM reservation_students rs JOIN users u ON u.id = rs.user_id WHERE rs.reservation_id = ? ORDER BY COALESCE(u.name, u.email) """, (res_id,)).fetchall() if not rows: r = conn.execute("SELECT user_id FROM reservations WHERE id = ?", (res_id,)).fetchone() if r and r["user_id"]: u = conn.execute("SELECT id, name, email FROM users WHERE id = ?", (r["user_id"],)).fetchone() if u: rows = [u] students = [{"id": int(r["id"]), "name": r["name"], "email": r["email"]} for r in rows] return {"reservation_id": res_id, "count": len(students), "students": students} @app.get("/api/reservations/list/paginated") def reservations_paginated( page: int = Query(1, ge=1, description="P�gina (1-based)"), size: int = Query(10, ge=1, le=100, description="Tama�o de p�gina"), user_id: Optional[int] = Query(None), docente_id: Optional[int] = Query(None), status: Optional[str] = Query(None), ): try: offset = (page - 1) * size where_clauses = ["1=1"] params: list = [] if user_id is not None: where_clauses.append("user_id = ?") params.append(user_id) if docente_id is not None: where_clauses.append("docente_id = ?") params.append(docente_id) if status is not None: where_clauses.append("status = ?") params.append(status) where_sql = "WHERE " + " AND ".join(where_clauses) with _conn() as conn: total_row = conn.execute( f"SELECT COUNT(*) AS c FROM reservations {where_sql}", tuple(params), ).fetchone() total = int(total_row["c"] or 0) rows = conn.execute( f""" SELECT id, created_at, date, time, duration_minutes, price_cents, name, email, phone, notes, user_id, status, docente_id FROM reservations {where_sql} ORDER BY datetime(created_at) DESC, id DESC LIMIT ? OFFSET ? """, tuple(params) + (size, offset), ).fetchall() items = [_reservation_row_to_dict(r) for r in rows] pages = max(1, math.ceil(total / size)) if total else 1 return { "items": items, "meta": { "page": page, "size": size, "total": total, "pages": pages, }, } except HTTPException: raise except Exception: raise HTTPException(status_code=500, detail="Error interno listando reservas") def purge_expired_temp_regs(): with _conn() as conn: conn.execute('DELETE FROM "formulario-registro" WHERE used=0 AND expires_at < ?', (_dt.datetime.utcnow().isoformat(),)) conn.commit() def generate_unique_pin(length: int = 6) -> str: digits = string.digits while True: pin = ''.join(_secrets.choice(digits) for _ in range(length)) with _conn() as conn: row = conn.execute( 'SELECT 1 FROM "formulario-registro" WHERE pin = ? AND used = 0 AND expires_at > ?', (pin, _dt.datetime.utcnow().isoformat()) ).fetchone() if not row: return pin def current_user(request: Request): user_id = request.session.get("user_id") if not user_id: return None return get_user_by_id(int(user_id)) def get_students(): with _conn() as conn: cur = conn.execute("SELECT id, name, email FROM users WHERE role='alumno' ORDER BY name") return cur.fetchall() def get_notifications_for_user(user_id: int): with _conn() as conn: cur = conn.execute(""" SELECT n.id, n.created_at, n.title, n.body, s.name AS sender_name, n.sender_id, n.recipient_id, n.reservation_id FROM notifications n LEFT JOIN users s ON s.id = n.sender_id WHERE n.recipient_id = ? AND COALESCE(n.deleted_by_recipient, 0) = 0 ORDER BY n.created_at DESC """, (user_id,)) return cur.fetchall() def get_notifications_sent_by_user(sender_id: int): with _conn() as conn: cur = conn.execute(""" SELECT n.id, n.created_at, n.title, n.body, r.name AS recipient_name, n.sender_id, n.recipient_id, n.reservation_id FROM notifications n LEFT JOIN users r ON r.id = n.recipient_id WHERE n.sender_id = ? AND COALESCE(n.deleted_by_sender, 0) = 0 ORDER BY n.created_at DESC """, (sender_id,)) return cur.fetchall() def get_notifications_sent_by_user_to_user(sender_id: int, recipient_id: int): with _conn() as conn: cur = conn.execute(""" SELECT n.id, n.created_at, n.title, n.body, r.name AS recipient_name, n.sender_id, n.recipient_id, n.reservation_id FROM notifications n LEFT JOIN users r ON r.id = n.recipient_id WHERE n.sender_id = ? AND n.recipient_id = ? AND COALESCE(n.deleted_by_sender, 0) = 0 AND COALESCE(n.deleted_by_recipient, 0) = 0 ORDER BY n.created_at DESC """, (sender_id, recipient_id)) return cur.fetchall() @app.post("/notificaciones/eliminar", name="delete_notification") async def delete_notification( request: Request, notification_id: int = Form(...), next_url: str | None = Form(None), target_recipient_id: int | None = Form(None), csrf: str | None = Form(None), ): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) require_csrf(request, csrf) with _conn() as conn: row = conn.execute( "SELECT id, sender_id, recipient_id FROM notifications WHERE id = ?", (notification_id,), ).fetchone() if not row: raise HTTPException(status_code=404, detail="Notificaci�n no encontrada.") me_id = int(me["id"]) sender_id = int(row["sender_id"]) if row["sender_id"] is not None else None recipient_id = int(row["recipient_id"]) if me_id == recipient_id: conn.execute("UPDATE notifications SET deleted_by_recipient = 1 WHERE id = ?", (notification_id,)) conn.commit() elif sender_id is not None and me_id == sender_id: if target_recipient_id is not None and int(target_recipient_id) == recipient_id: conn.execute("UPDATE notifications SET deleted_by_sender = 1, deleted_by_recipient = 1 WHERE id = ?", (notification_id,)) else: conn.execute("UPDATE notifications SET deleted_by_sender = 1 WHERE id = ?", (notification_id,)) conn.commit() elif me["role"] in ("docente", "admin") and target_recipient_id is not None and int(target_recipient_id) == recipient_id: conn.execute("UPDATE notifications SET deleted_by_recipient = 1 WHERE id = ?", (notification_id,)) conn.commit() else: raise HTTPException(status_code=403, detail="No puedes eliminar esta notificaci�n.") if next_url: parsed = urlparse(next_url) if parsed.scheme == "" and parsed.netloc == "": return RedirectResponse(url=next_url, status_code=303) return RedirectResponse(url="/notificaciones", status_code=303) @app.get("/", response_class=HTMLResponse) async def home(request: Request): me = current_user(request) return templates.TemplateResponse("index.html", {"request": request, "user": me}) @app.get("/about", response_class=HTMLResponse) async def about(request: Request): me = current_user(request) return templates.TemplateResponse("about.html", {"request": request, "user": me}) @app.get("/ticketing", response_class=HTMLResponse) async def ticketing(request: Request): return templates.TemplateResponse("ticketing.html", {"request": request, "user": current_user(request)}) @app.get("/login", name="login", response_class=HTMLResponse) async def login_get(request: Request): return templates.TemplateResponse("login.html", {"request": request, "error": None}) @app.post("/login", name="login_post", response_class=HTMLResponse) async def login_post(request: Request, email: str = Form(...), password: str = Form(...)): email_norm = email.strip().lower() user = get_user_by_email(email_norm) if not user or not pwd_context.verify(password, user["password_hash"]): return templates.TemplateResponse("login.html", {"request": request, "error": "Credenciales inv�lidas"}) rotate_session(request) request.session["user_id"] = int(user["id"]) request.session["role"] = user["role"] if user["role"] == "admin" or email_norm == "admin@admin.es": return RedirectResponse(url=request.url_for("admin"), status_code=303) elif user["role"] == "docente": return RedirectResponse(url=request.url_for("docente"), status_code=303) else: return RedirectResponse(url=request.url_for("dashboard"), status_code=303) @app.get("/docente", name="docente", response_class=HTMLResponse) async def docente_dashboard(request: Request): me = current_user(request) if not me: return RedirectResponse(url=request.url_for("login"), status_code=303) students = [] try: if me["role"] in ("docente", "admin"): students = get_students() except Exception: students = [] return templates.TemplateResponse( "docente.html", { "request": request, "user": me, "students": students } ) @app.get("/valoraciones", name="valoraciones", response_class=HTMLResponse) async def valoraciones_page(request: Request): me = current_user(request) if not me: return RedirectResponse(url=request.url_for("login"), status_code=303) role = str(me["role"]).lower() if "role" in me.keys() else "" if role not in ("docente", "admin"): return RedirectResponse(url=request.url_for("dashboard"), status_code=303) return templates.TemplateResponse("valoraciones.html", {"request": request, "user": me}) @app.get("/api/users/{user_id}", response_class=JSONResponse, name="api_user_detail") async def api_user_detail(user_id: int, request: Request): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) if me["role"] not in ("docente", "admin"): raise HTTPException(status_code=403, detail="Solo docentes o admin") with _conn() as conn: row = conn.execute(""" SELECT u.id,u.name,u.email,u.created_at,u.role, p.phone,p.avatar,p.dni,p.birth_date,p.city, s.dominant_foot,s.team,s.category,s.position,s.injury_history,s.strengths,s.weaknesses,s.training_type, h.physical_work,h.physical_work_details,h.smoking,h.alcohol,h.recovery,h.chest_pain,h.discomfort, c.whatsapp_content,c.video_permission,c.privacy_acceptance,c.data_confirmation, COALESCE(f.matricula_eur,0.0) AS matricula_eur, CAST(ROUND(COALESCE(f.matricula_eur,0.0)*100) AS INTEGER) AS matricula_cents, c.agreement, g.occupation,g.study_place,g.parent_name,g.parent_email, m.found_us,m.enjoyment,m.nerves_confidence,m.additional_comments FROM users u LEFT JOIN user_profile p ON p.user_id=u.id LEFT JOIN user_sports s ON s.user_id=u.id LEFT JOIN user_health h ON h.user_id=u.id LEFT JOIN user_consent c ON c.user_id=u.id LEFT JOIN user_guardians g ON g.user_id=u.id LEFT JOIN user_marketing m ON m.user_id=u.id LEFT JOIN user_finance f ON f.user_id=u.id WHERE u.id = ? """, (user_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="Usuario no encontrado") return dict(row) @app.get("/register", name="register", response_class=HTMLResponse) async def register_get(request: Request): me = current_user(request) ctx_user = me or {"name": "", "email": "", "phone": ""} return templates.TemplateResponse("register.html", {"request": request, "error": None, "user": ctx_user}) @app.post("/register", name="register_post", response_class=HTMLResponse) async def register_post(request: Request, name: str = Form(...), email: str = Form(...), password: str = Form(...)): email_norm = email.strip().lower() if get_user_by_email(email_norm): return templates.TemplateResponse( "register.html", {"request": request, "error": "Ese email ya existe", "user": {"name": name.strip(), "email": email_norm, "phone": ""}} ) if len(password) < 6: return templates.TemplateResponse( "register.html", {"request": request, "error": "La contrase�a debe tener 6+ caracteres", "user": {"name": name.strip(), "email": email_norm, "phone": ""}} ) create_user(name.strip(), email_norm, password, role="alumno") user = get_user_by_email(email_norm) request.session["user_id"] = int(user["id"]) request.session["role"] = user["role"] if user["role"] == "admin" or email_norm == "admin@admin.es": return RedirectResponse(url=request.url_for("admin"), status_code=303) else: return RedirectResponse(url=request.url_for("dashboard"), status_code=303) @app.get("/academy-register", name="academy_register_get", response_class=HTMLResponse) async def academy_register_get(request: Request): me = current_user(request) return templates.TemplateResponse( "academy-register.html", {"request": request, "user": me} ) UPLOADS_DIR = BASE_DIR / "static" / "uploads" / "academy" @app.post("/academy-register", name="academy_register_post") async def academy_register_post(request: Request): print("="*80) print("DEBUG: academy_register_post function called") print("="*80) form = await request.form() print(f"DEBUG: Form keys: {list(form.keys())}") print(f"DEBUG: Form has playerPhoto: {'playerPhoto' in form}") email = (form.get("email") or "").strip().lower() name = (form.get("name") or "").strip() password = (form.get("password") or "") first_name = (form.get("firstName") or "").strip() last_name = (form.get("lastName") or "").strip() print(f"DEBUG: email={email}, name={name}, first_name={first_name}, last_name={last_name}") if not email: raise HTTPException(status_code=400, detail="El email es obligatorio.") if not first_name or len(first_name) < 2: raise HTTPException(status_code=400, detail="Indica tu nombre.") if not last_name or len(last_name) < 2: raise HTTPException(status_code=400, detail="Indica tus apellidos.") if not name or len(name) < 2: raise HTTPException(status_code=400, detail="Indica tu nombre (mínimo 2 caracteres).") if not password or len(password) < 6: raise HTTPException(status_code=400, detail="La contraseña debe tener al menos 6 caracteres.") existing = get_user_by_email(email) if existing: raise HTTPException(status_code=409, detail="Ya existe un usuario con ese email. Inicia sesión.") # Store photo content temporarily (before user creation) photo_content = None photo_ext = None player_photo = form.get("playerPhoto") print(f"DEBUG: player_photo type: {type(player_photo)}") is_upload = hasattr(player_photo, 'filename') and hasattr(player_photo, 'read') print(f"DEBUG: is_upload (hasattr check): {is_upload}") if is_upload: print(f"DEBUG: player_photo.filename: {player_photo.filename}") if is_upload and player_photo.filename: try: photo_content = await player_photo.read() print(f"DEBUG: content length: {len(photo_content) if photo_content else 0}") if photo_content: photo_ext = os.path.splitext(player_photo.filename)[1].lower() or ".jpg" print(f"DEBUG: Photo stored in memory, will save after user creation") except Exception as e: print(f"ERROR: Failed to read photo: {e}") import traceback traceback.print_exc() # Create user first create_user(name.strip(), email, password, role="alumno", first_name=first_name, last_name=last_name) user = get_user_by_email(email) user_id = int(user["id"]) # Now save photo with user_ID.ext format avatar_path = None if photo_content and photo_ext: try: uploads_dir = BASE_DIR / "static" / "uploads" uploads_dir.mkdir(parents=True, exist_ok=True) filename = f"user_{user_id}{photo_ext}" file_path = uploads_dir / filename print(f"DEBUG: Saving photo to: {file_path}") with open(file_path, "wb") as f: f.write(photo_content) print(f"DEBUG: Photo saved successfully to {file_path}") avatar_path = f"/static/uploads/{filename}" print(f"DEBUG: avatar_path: {avatar_path}") except Exception as e: print(f"ERROR: Failed to save photo to disk: {e}") import traceback traceback.print_exc() data: dict[str, str] = {} for k, v in form.items(): if isinstance(v, UploadFile): continue if v not in (None, "", "null"): data[k] = str(v) def pick(*keys): for k in keys: val = data.get(k) if val not in (None, "", "null"): return val return None profile_vals = { "phone": pick("phone", "whatsapp"), "dni": pick("dni"), "birth_date": pick("birthDate", "birth_date"), "city": pick("city"), "avatar": avatar_path or pick("playerPhoto", "avatar") } sports_vals = { "team": pick("currentTeam", "team", "club"), "category": pick("category"), "position": pick("currentPosition", "preferredPosition", "position"), "dominant_foot": pick("dominantFoot"), "strengths": pick("strengths"), "weaknesses": pick("weaknesses"), "training_type": pick("trainingType"), "injury_history": pick("injuries", "injuryHistory", "injury_history") } health_vals = { "physical_work": pick("physicalWork"), "physical_work_details": pick("physicalWorkDetails"), "smoking": pick("smoking"), "alcohol": pick("alcohol"), "recovery": pick("recovery"), "chest_pain": pick("chestPain"), "discomfort": pick("discomfort") } consent_vals = { "whatsapp_content": pick("whatsappContent"), "video_permission": pick("videoPermission"), "privacy_acceptance": pick("privacyAcceptance"), "data_confirmation": pick("dataConfirmation"), "agreement": pick("agreement") } guardians_vals = { "occupation": pick("occupation"), "study_place": pick("studyPlace", "study_place"), "parent_name": pick("parentName", "parent_name"), "parent_email": pick("parentEmail", "parent_email") } marketing_vals = { "found_us": pick("foundUs", "found_us"), "enjoyment": pick("enjoyment"), "nerves_confidence": pick("nervesConfidence", "nerves_confidence"), "additional_comments": pick("additionalComments", "additional_comments") } with _conn() as conn: _upsert_satellites_empty(conn, user_id) if any(v is not None for v in profile_vals.values()): sets = ", ".join([f"{k}=COALESCE(?, {k})" for k in profile_vals.keys()]) conn.execute(f"UPDATE user_profile SET {sets} WHERE user_id=?", (*profile_vals.values(), user_id)) if any(v is not None for v in sports_vals.values()): sets = ", ".join([f"{k}=COALESCE(?, {k})" for k in sports_vals.keys()]) conn.execute(f"UPDATE user_sports SET {sets} WHERE user_id=?", (*sports_vals.values(), user_id)) if any(v is not None for v in health_vals.values()): sets = ", ".join([f"{k}=COALESCE(?, {k})" for k in health_vals.keys()]) conn.execute(f"UPDATE user_health SET {sets} WHERE user_id=?", (*health_vals.values(), user_id)) if any(v is not None for v in consent_vals.values()): sets = ", ".join([f"{k}=COALESCE(?, {k})" for k in consent_vals.keys()]) conn.execute(f"UPDATE user_consent SET {sets} WHERE user_id=?", (*consent_vals.values(), user_id)) if any(v is not None for v in guardians_vals.values()): sets = ", ".join([f"{k}=COALESCE(?, {k})" for k in guardians_vals.keys()]) conn.execute(f"UPDATE user_guardians SET {sets} WHERE user_id=?", (*guardians_vals.values(), user_id)) if any(v is not None for v in marketing_vals.values()): sets = ", ".join([f"{k}=COALESCE(?, {k})" for k in marketing_vals.keys()]) conn.execute(f"UPDATE user_marketing SET {sets} WHERE user_id=?", (*marketing_vals.values(), user_id)) conn.commit() print("="*80) print("DEBUG: About to return JSON response: {'ok': True, 'redirect': '/login'}") print("="*80) return JSONResponse({"ok": True, "redirect": "/login"}) @app.get("/info-usuarios", response_class=HTMLResponse) async def info_usuario(request: Request): me = current_user(request) if not me: return RedirectResponse(url=request.url_for("login"), status_code=303) return templates.TemplateResponse("info-usuarios.html", { "request": request, "user": me }) @app.get("/gestion-usuarios", response_class=HTMLResponse, name="gestion_usuarios") async def gestion_usuarios(request: Request): me = current_user(request) if not me: return RedirectResponse(url=request.url_for("login"), status_code=303) if isinstance(me, sqlite3.Row): me = dict(me) role = str((me.get("role") if isinstance(me, dict) else me["role"]) or "").lower() if role != "admin": return RedirectResponse(url=request.url_for("dashboard"), status_code=303) with _conn() as conn: cur = conn.execute(""" SELECT u.id, u.name, u.email, u.role, u.created_at, LOWER(COALESCE(s.category, '')) AS category FROM users u LEFT JOIN user_sports s ON s.user_id = u.id ORDER BY u.created_at DESC, u.id DESC """) users = [dict(r) for r in cur.fetchall()] return templates.TemplateResponse( "gestion-usuarios.html", { "request": request, "user": me, "users": users } ) @app.get("/dashboard", name="dashboard", response_class=HTMLResponse) async def dashboard(request: Request): me = current_user(request) if not me: return RedirectResponse(url=request.url_for("login"), status_code=303) me_id = int(me["id"]) if isinstance(me, sqlite3.Row) else int(me.get("id")) me_email = (me["email"] if isinstance(me, sqlite3.Row) else me.get("email") or "").strip() with _conn() as conn: row = conn.execute(""" SELECT COUNT(DISTINCT r.id) AS c FROM reservations r LEFT JOIN reservation_students rs ON rs.reservation_id = r.id WHERE rs.user_id = ? OR r.user_id = ? OR LOWER(COALESCE(r.email,'')) = LOWER(?) """, (me_id, me_id, me_email)).fetchone() reservation_count = int(row["c"] if row and row["c"] is not None else 0) return templates.TemplateResponse( "dashboard.html", { "request": request, "user": me, "reservation_count": reservation_count, } ) @app.get("/admin", response_class=HTMLResponse) async def admin(request: Request): user = current_user(request) if not user: return RedirectResponse(url="/login", status_code=303) return templates.TemplateResponse("admin.html", {"request": request, "user": user}) # ----------------------------------------------------------- # 🚫 Código desactivado intencionalmente (NO usar en producción) # Motivo: vulnerabilidad crítica de escalada de privilegios. # Cualquier usuario autenticado podía hacerse admin con solo # visitar /make-me-admin, sin validación ni autorización. # ----------------------------------------------------------- # @app.get("/make-me-admin") # async def make_me_admin(request: Request): # # Obtiene el usuario actual desde la sesión # me = current_user(request) # # # Verifica si hay sesión activa # if not me: # return JSONResponse({"error": "No autenticado"}, status_code=401) # # # ❌ Peligro: eleva el rol del usuario a 'admin' sin permisos # # Esto permitía que cualquier usuario se convirtiera en administrador. # with _conn() as conn: # conn.execute("UPDATE users SET role='admin' WHERE id = ?", (int(me["id"]),)) # conn.commit() # # # Respuesta de confirmación (también insegura, revela el nombre del usuario) # return JSONResponse({"message": f"Usuario {me['name']} ahora es admin"}) # # ----------------------------------------------------------- # Fin del bloque inseguro — mantenido solo como referencia histórica. # ----------------------------------------------------------- @app.get("/notificaciones", name="notificaciones", response_class=HTMLResponse) async def notificaciones(request: Request, user_id: int | None = None): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) target_user_id = int(me["id"]) view_user = None if user_id is not None: try: user_id = int(user_id) except Exception: user_id = None if user_id and me["role"] in ("docente", "admin"): view_user = get_user_by_id(user_id) if view_user: target_user_id = int(view_user["id"]) inbox = get_notifications_for_user(target_user_id) sent = [] if me["role"] in ("docente", "admin"): if view_user: sent = get_notifications_sent_by_user_to_user(int(me["id"]), target_user_id) else: sent = get_notifications_sent_by_user(int(me["id"])) students = [] if me["role"] in ("docente", "admin"): students = get_students() def _row_to_dict(r): try: return {k: r[k] for k in r.keys()} except Exception: return dict(r) notifications_json = [_row_to_dict(n) for n in (inbox or [])] appointments = [] appointments_map = {} appointments_json = [] if view_user: with _conn() as conn: appointments = conn.execute(""" SELECT id, user_id, date, time, duration_minutes, status, created_at, name, email, phone, notes, price_cents FROM reservations WHERE user_id = ? ORDER BY date ASC, time ASC """, (target_user_id,)).fetchall() appointments_json = [_row_to_dict(a) for a in (appointments or [])] appointments_map = {int(a["id"]): _row_to_dict(a) for a in (appointments or [])} return templates.TemplateResponse( "notificaciones.html", { "request": request, "user": me, "view_user": dict(view_user) if view_user else None, "notifications": inbox, "sent_notifications": sent, "students": students, "notifications_json": notifications_json, "appointments": appointments, "appointments_json": appointments_json, "appointments_map": appointments_map, } ) from typing import List @app.get("/api/expenses", response_model=List[dict]) async def api_expenses_list( request: Request, date_from: Optional[str] = None, date_to: Optional[str] = None, user_id: Optional[int] = None, ): require_admin(request) def _ok_date(s: str): try: datetime.strptime(s, "%Y-%m-%d") except Exception: raise HTTPException(status_code=400, detail=f"Fecha inv�lida: {s} (YYYY-MM-DD)") where = [] params: list[Any] = [] if date_from: _ok_date(date_from); where.append("jg.date >= ?"); params.append(date_from) if date_to: _ok_date(date_to); where.append("jg.date <= ?"); params.append(date_to) if user_id is not None: where.append("jg.user_id = ?"); params.append(int(user_id)) sql = """ SELECT jg.id, jg.created_at, jg.date, jg.user_id, jg.reservation_id, jg.amount_cents, jg.category, jg.concept, jg.notes, u.name AS user_name, u.email AS user_email FROM jugadores_gastos jg LEFT JOIN users u ON u.id = jg.user_id """ if where: sql += " WHERE " + " AND ".join(where) sql += " ORDER BY jg.date ASC, jg.id ASC" with _conn() as conn: conn.row_factory = sqlite3.Row rows = conn.execute(sql, tuple(params)).fetchall() out = [] for r in rows: out.append({ "id": r["id"], "created_at": r["created_at"], "date": r["date"], "user_id": r["user_id"], "user_name": r["user_name"], "user_email": r["user_email"], "reservation_id": r["reservation_id"], "amount_cents": int(r["amount_cents"] or 0), "category": r["category"], "concept": r["concept"], "notes": r["notes"], }) return out class ExpenseIn(BaseModel): date: str amount_cents: int = Field(ge=0) category: Optional[str] = None concept: Optional[str] = None notes: Optional[str] = None user_id: Optional[int] = None reservation_id: Optional[int] = None @app.post("/api/expenses", response_model=dict) async def api_expenses_create(request: Request, payload: ExpenseIn): require_admin(request) try: datetime.strptime(payload.date, "%Y-%m-%d") except Exception: raise HTTPException(status_code=400, detail="Fecha inválida (YYYY-MM-DD).") created_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with _conn() as conn: cur = conn.execute(""" INSERT INTO jugadores_gastos (created_at, date, user_id, reservation_id, amount_cents, category, concept, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( created_at, payload.date.strip(), int(payload.user_id) if payload.user_id is not None else None, int(payload.reservation_id) if payload.reservation_id else None, int(payload.amount_cents or 0), (payload.category or "").strip() or None, (payload.concept or "").strip() or None, (payload.notes or "").strip() or None, )) gid = cur.lastrowid return {"ok": True, "id": gid} @app.delete("/api/expenses/{gid}", response_model=dict) async def api_expenses_delete(request: Request, gid: int): require_admin(request) with _conn() as conn: conn.execute("DELETE FROM jugadores_gastos WHERE id = ?", (gid,)) return {"ok": True} @app.post("/notificaciones", name="create_notification") async def create_notification( request: Request, recipient_id: int = Form(...), title: str = Form(...), body: str = Form(...), next_url: str | None = Form(None), reservation_id: int | None = Form(None), csrf: str | None = Form(None), ): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) if me["role"] not in ("docente", "admin"): raise HTTPException(status_code=403, detail="Solo docentes o admin pueden enviar notificaciones.") require_csrf(request, csrf) title = (title or "").strip() body = (body or "").strip() if not title or not body: raise HTTPException(status_code=400, detail="T�tulo y mensaje son obligatorios.") try: recipient_id = int(recipient_id) except Exception: raise HTTPException(status_code=400, detail="Destinatario inv�lido.") res_id_to_save = None if reservation_id is not None and str(reservation_id).strip() != "": try: res_id_to_save = int(reservation_id) except Exception: raise HTTPException(status_code=400, detail="Reserva inv�lida.") with _conn() as conn: res = conn.execute( "SELECT id, user_id FROM reservations WHERE id = ?", (res_id_to_save,) ).fetchone() if not res: raise HTTPException(status_code=404, detail="Reserva no encontrada.") if int(res["user_id"] or 0) != recipient_id: raise HTTPException(status_code=400, detail="La reserva no pertenece al alumno seleccionado.") with _conn() as conn: conn.execute( """ INSERT INTO notifications (created_at, sender_id, recipient_id, title, body, is_read, reservation_id) VALUES (?, ?, ?, ?, ?, 0, ?) """, (dt.datetime.utcnow().isoformat(), int(me["id"]), recipient_id, title, body, res_id_to_save) ) conn.commit() if next_url: parsed = urlparse(next_url) if parsed.scheme == "" and parsed.netloc == "": return RedirectResponse(url=next_url, status_code=303) return RedirectResponse(url="/notificaciones", status_code=303) @app.get("/api/teachers", response_class=JSONResponse, name="api_teachers") async def api_teachers(request: Request): user = current_user(request) if not user: return RedirectResponse(url="/login", status_code=303) with _conn() as conn: cur = conn.execute( "SELECT id, name, email, created_at, role FROM users WHERE role = 'docente' ORDER BY name ASC" ) data = [dict(r) for r in cur.fetchall()] return data ALLOWED_ROLES = {"alumno", "docente", "admin"} ALLOWED_CATEGORIES = { "prebenjamin", "benjamin", "alevines", "infantiles", "cadetes", "juvenil", "senior" } class CategoryUpdate(BaseModel): category: Optional[str] = None class RoleUpdate(BaseModel): role: str def set_user_role(user_id: int, role: str): with _conn() as conn: conn.execute("UPDATE users SET role = ? WHERE id = ?", (role, user_id)) conn.commit() ALLOWED_CATEGORIES = { "prebenjamin", "benjamin", "alevines", "infantiles", "cadetes", "juvenil", "senior" } class CategoryUpdate(BaseModel): category: Optional[str] = None @app.post("/api/users/{user_id}/category", name="api_users_set_category") async def api_users_set_category(user_id: int, payload: CategoryUpdate, request: Request): require_admin(request) target = get_user_by_id(user_id) if not target: raise HTTPException(status_code=404, detail="Usuario no encontrado") cat = (payload.category or "").strip().lower() if cat and cat not in ALLOWED_CATEGORIES: raise HTTPException(status_code=400, detail="Categoría no válida") with _conn() as conn: conn.execute("INSERT OR IGNORE INTO user_sports(user_id) VALUES (?)", (user_id,)) conn.execute("UPDATE user_sports SET category = ? WHERE user_id = ?", (cat if cat else None, user_id)) conn.commit() return {"ok": True, "user_id": user_id, "category": cat} @app.get("/api/users", response_class=JSONResponse, name="api_users") async def api_users( request: Request, roles: Optional[str] = Query(None), role: Optional[str] = Query(None) ): user = current_user(request) if not user: return RedirectResponse(url="/login", status_code=303) roles_list: List[str] = [] if roles: roles_list = [r.strip().lower() for r in roles.split(",") if r.strip()] elif role: roles_list = [role.strip().lower()] where = "" params: list = [] if roles_list: placeholders = ",".join("?" * len(roles_list)) where = f"WHERE lower(u.role) IN ({placeholders})" params.extend(roles_list) with _conn() as conn: cur = conn.execute(f""" SELECT u.id,u.name,u.email,u.created_at,u.role,u.first_name,u.last_name, p.phone,p.avatar, s.team,s.category,s.position,s.injury_history, p.dni,p.birth_date,p.city, s.dominant_foot,s.strengths,s.weaknesses,s.training_type, h.physical_work,h.physical_work_details,h.smoking,h.alcohol,h.recovery,h.chest_pain,h.discomfort, c.whatsapp_content,c.video_permission,c.privacy_acceptance,c.data_confirmation, COALESCE(f.matricula_eur,0.0) AS matricula_eur, CAST(ROUND(COALESCE(f.matricula_eur,0.0)*100) AS INTEGER) AS matricula_cents, c.agreement, g.occupation,g.study_place,g.parent_name,g.parent_email, m.found_us,m.enjoyment,m.nerves_confidence,m.additional_comments FROM users u LEFT JOIN user_profile p ON p.user_id=u.id LEFT JOIN user_sports s ON s.user_id=u.id LEFT JOIN user_health h ON h.user_id=u.id LEFT JOIN user_consent c ON c.user_id=u.id LEFT JOIN user_guardians g ON g.user_id=u.id LEFT JOIN user_marketing m ON m.user_id=u.id LEFT JOIN user_finance f ON f.user_id=u.id {where} ORDER BY u.id DESC """, params) data = [dict(r) for r in cur.fetchall()] return data @app.post("/api/users/{user_id}/role", name="api_users_set_role") async def api_users_set_role(user_id: int, payload: RoleUpdate, request: Request): require_admin(request) target = get_user_by_id(user_id) if not target: raise HTTPException(status_code=404, detail="Usuario no encontrado") new_role = (payload.role or "").strip().lower() if new_role not in ALLOWED_ROLES: raise HTTPException(status_code=400, detail="Rol inv�lido") current_role = str(target["role"]).lower() was_admin = current_role == "admin" is_demoting_admin = was_admin and new_role != "admin" if current_role == new_role: return {"ok": True, "user_id": user_id, "role": new_role, "changed": False} if is_demoting_admin: admins_count = count_admins() if admins_count <= 1: raise HTTPException( status_code=400, detail="Debe existir al menos un usuario con rol admin. No puedes degradar al �ltimo admin." ) set_user_role(user_id, new_role) return {"ok": True, "user_id": user_id, "role": new_role, "changed": True} DateStr = Annotated[str, StringConstraints(pattern=r"^\d{4}-\d{2}-\d{2}$")] TimeStr = Annotated[str, StringConstraints(pattern=r"^\d{2}:\d{2}$")] class ReservationIn(BaseModel): date: DateStr time: TimeStr duration_minutes: int = Field(60, ge=5, le=24*60) price_cents: int = Field(0, ge=0) user_ids: List[int] = Field(default_factory=list) user_id: Optional[int] = None name: Optional[str] = None email: Optional[str] = None phone: Optional[str] = None status: Optional[str] = "confirmada" notes: Optional[str] = None docente_id: Optional[int] = None paid: bool = False class ReservationUpdate(BaseModel): date: Optional[DateStr] = None time: Optional[TimeStr] = None duration_minutes: Optional[int] = Field(None, ge=5, le=24*60) price_cents: Optional[int] = Field(None, ge=0) user_ids: Optional[List[int]] = None user_id: Optional[int] = None name: Optional[str] = None email: Optional[str] = None phone: Optional[str] = None status: Optional[str] = None notes: Optional[str] = None docente_id: Optional[int] = None paid: Optional[bool] = None class RatingIn(BaseModel): player_id: int score: int = Field(ge=1, le=5) comment: Optional[str] = None def _reservation_row_to_dict(row): base = { "id": row["id"], "created_at": row["created_at"], "date": row["date"], "time": row["time"], "duration_minutes": row["duration_minutes"], "price_cents": int(row["price_cents"] or 0), "price_eur": (int(row["price_cents"] or 0) / 100.0), "name": row["name"], "email": row["email"], "phone": row["phone"], "notes": row["notes"], "user_id": row["user_id"], "status": row["status"], "docente_id": row["docente_id"] if "docente_id" in row.keys() else None, "paid": bool(row["paid"]) if "paid" in row.keys() and row["paid"] is not None else False, } with _conn() as c: srows = c.execute(""" SELECT u.id, u.name, u.email FROM reservation_students rs JOIN users u ON u.id = rs.user_id WHERE rs.reservation_id = ? ORDER BY u.name """, (row["id"],)).fetchall() base["student_ids"] = [int(r["id"]) for r in srows] base["students"] = [{"id": int(r["id"]), "name": r["name"], "email": r["email"]} for r in srows] return base @app.patch("/api/reservations/{res_id}", response_model=dict) async def api_reservations_update(request: Request, res_id: int, payload: ReservationUpdate): require_docente_or_admin(request) updates = payload.dict(exclude_unset=True) if hasattr(payload, "dict") else (payload or {}) try: raw = await request.json() except Exception: raw = {} if "duration" in raw and "duration_minutes" not in updates: try: updates["duration_minutes"] = int(raw.get("duration") or 0) except Exception: pass if "teacher_id" in raw and "docente_id" not in updates: updates["docente_id"] = raw.get("teacher_id") if "price_eur" in raw and "price_cents" not in updates: try: updates["price_cents"] = int(round(float(str(raw["price_eur"]).replace(",", ".")) * 100)) except Exception: pass if "price" in raw and "price_cents" not in updates: try: updates["price_cents"] = int(round(float(str(raw["price"]).replace(",", ".")) * 100)) except Exception: pass has_other_updates = bool(updates) with _conn() as conn: conn.row_factory = sqlite3.Row prev_row = conn.execute("SELECT * FROM reservations WHERE id = ?", (res_id,)).fetchone() if not prev_row: raise HTTPException(status_code=404, detail="Reserva no encontrada") try: prev_paid = bool(prev_row["paid"]) except Exception: prev_paid = False new_paid = prev_paid if "paid" in updates and updates["paid"] is not None: new_paid = bool(updates["paid"]) else: val = raw.get("paid", raw.get("has_paid")) if val is not None: if isinstance(val, bool): new_paid = val elif isinstance(val, (int, str)): s = str(val).strip().lower() if s in ("1", "true", "sí", "si", "on"): new_paid = True elif s in ("0", "false", "no", "off"): new_paid = False if not has_other_updates and new_paid == prev_paid: raise HTTPException(status_code=400, detail="Nada que actualizar") updates["paid"] = 1 if new_paid else 0 if "date" in updates and updates["date"] is not None: try: dt.datetime.strptime(updates["date"], "%Y-%m-%d") except Exception: raise HTTPException(status_code=400, detail="Fecha inválida. Usa formato YYYY-MM-DD.") if "time" in updates and updates["time"] is not None: try: t = str(updates["time"]).strip()[:5] hh, mm = map(int, t.split(":")) except Exception: raise HTTPException(status_code=400, detail="Hora inválida. Usa formato HH:MM.") if not (0 <= hh < 24 and 0 <= mm < 60): raise HTTPException(status_code=400, detail="Hora inválida.") if hh == 0 and mm == 0: raise HTTPException(status_code=400, detail="No se admiten reservas a medianoche (00:00).") if not (8 <= hh < 22): raise HTTPException(status_code=400, detail="Hora fuera de horario (08:00–22:00).") updates["time"] = f"{hh:02d}:{mm:02d}" if "duration_minutes" in updates and updates["duration_minutes"] is not None: dmin = int(updates["duration_minutes"]) if dmin <= 0 or dmin > 24*60: raise HTTPException(status_code=400, detail="Duración inválida.") if "status" in updates and updates["status"] is not None: status_norm = str(updates["status"]).strip().lower() if status_norm not in ALLOWED_STATUSES: raise HTTPException(status_code=400, detail=f"Estado no permitido: {status_norm}") updates["status"] = status_norm if "docente_id" in updates and updates["docente_id"] is not None: docente_id = updates["docente_id"] with _conn() as conn: d = conn.execute( "SELECT id FROM users WHERE id=? AND LOWER(COALESCE(role,''))='docente'", (docente_id,) ).fetchone() if not d: raise HTTPException(status_code=400, detail="docente_id no corresponde a un docente válido.") if "price_eur" in updates and "price_cents" not in updates: try: updates["price_cents"] = int(round(float(str(updates["price_eur"]).replace(",", ".")) * 100)) except Exception: updates["price_cents"] = 0 if "price_cents" in updates: try: updates["price_cents"] = int(updates["price_cents"] or 0) except Exception: updates["price_cents"] = 0 if updates["price_cents"] < 0: raise HTTPException(status_code=400, detail="Precio inválido.") allowed = { "date","time","duration_minutes","price_cents","name","email","phone","notes","user_id","status","docente_id","paid" } set_parts = [] args = [] for k, v in updates.items(): if k in allowed: set_parts.append(f"{k} = ?") args.append(v) if not set_parts: raise HTTPException(status_code=400, detail="Ningún campo permitido para actualizar") try: effective_price_cents = int(updates.get("price_cents") if "price_cents" in updates else (prev_row["price_cents"] or 0)) except Exception: effective_price_cents = 0 do_charge_now = (not prev_paid) and new_paid args.append(res_id) with _conn() as conn: conn.row_factory = sqlite3.Row if do_charge_now and effective_price_cents > 0: rows = conn.execute( "SELECT user_id FROM reservation_students WHERE reservation_id = ?", (res_id,) ).fetchall() student_ids = [int(r["user_id"]) for r in rows if r["user_id"] is not None] if not student_ids and prev_row["user_id"]: try: student_ids = [int(prev_row["user_id"])] except Exception: student_ids = [] per_student_cents = {} try: psc = raw.get("per_student_cents") or {} if isinstance(psc, dict): tmp = {} for k, v in psc.items(): try: kk = int(k) if isinstance(k, str) else int(k) except Exception: continue try: vv = int(v or 0) except Exception: vv = 0 tmp[kk] = max(0, vv) per_student_cents = tmp except Exception: per_student_cents = {} if student_ids: conn.executemany( "INSERT OR IGNORE INTO user_finance(user_id, matricula_eur) VALUES (?, 0.0)", [(sid,) for sid in student_ids] ) shares = {} if per_student_cents: for sid in student_ids: shares[sid] = max(0, int(per_student_cents.get(sid, effective_price_cents) or 0)) else: for sid in student_ids: shares[sid] = max(0, int(effective_price_cents or 0)) now_txt = dt.datetime.now().strftime("%Y-%m-%dT%H:%M:%S") date_txt = updates.get("date") or prev_row["date"] time_txt = (updates.get("time") or prev_row["time"] or "00:00")[:5] concept_base = f"Reserva {date_txt} {time_txt}" to_ins = [] for sid, share in shares.items(): share_eur = round(share / 100.0, 2) conn.execute( "UPDATE user_finance SET matricula_eur = ROUND(COALESCE(matricula_eur,0.0) - ?, 2) WHERE user_id = ?", (share_eur, sid) ) if share > 0: to_ins.append(( now_txt, date_txt, sid, res_id, int(share), "reserva", concept_base, None )) if to_ins: conn.executemany(""" INSERT INTO jugadores_gastos (created_at, date, user_id, reservation_id, amount_cents, category, concept, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, to_ins) conn.execute(f"UPDATE reservations SET {', '.join(set_parts)} WHERE id = ?", tuple(args)) conn.commit() return {"ok": True, "id": res_id} @app.delete("/api/reservations/{rid}", response_model=dict) async def api_reservations_delete(request: Request, rid: int): require_docente_or_admin(request) def _table_exists(conn, name: str) -> bool: row = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name = ?", (name,), ).fetchone() return bool(row) def _get_uids_from_bridge(conn, rid: int) -> list[int]: for t in ("reservation_users", "reservation_students", "reservation_alumnos"): if _table_exists(conn, t): try: rows = conn.execute( f"SELECT user_id FROM {t} WHERE reservation_id = ?", (rid,), ).fetchall() if rows: return [int(r[0]) for r in rows if r[0] is not None] except Exception: pass return [] with _conn() as conn: conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM reservations WHERE id = ?", (rid,)).fetchone() if not row: raise HTTPException(status_code=404, detail="Reserva no encontrada.") paid_flag = False try: paid_flag = bool(row["paid"]) except Exception: paid_flag = False if paid_flag: cols = set(row.keys()) price_cents = int(row["price_cents"] or 0) if "price_cents" in cols else 0 uids: list[int] = [] if "user_ids_json" in cols and row["user_ids_json"]: try: uids = [int(x) for x in json.loads(row["user_ids_json"]) or []] except Exception: uids = [] elif "user_id" in cols and row["user_id"]: try: uids = [int(row["user_id"])] except Exception: uids = [] if not uids: uids = _get_uids_from_bridge(conn, rid) uids = sorted({u for u in uids if isinstance(u, int)}) per_map: dict[int, int] = {} if "per_student_cents_json" in cols and row["per_student_cents_json"]: try: j = json.loads(row["per_student_cents_json"]) or {} for k, v in j.items(): try: sid = int(k) if not uids or sid in uids: per_map[sid] = int(v or 0) except Exception: continue except Exception: per_map = {} if not per_map: sids = uids n = len(sids) if n > 0: base, rem = divmod(abs(price_cents), n) tmp = {sid: base for sid in sids} for sid in sids[:rem]: tmp[sid] += 1 if price_cents < 0: for k in list(tmp.keys()): tmp[k] = -tmp[k] per_map = tmp for sid, cents in per_map.items(): eur = (cents or 0) / 100.0 cur = conn.execute( "SELECT user_id FROM user_finance WHERE user_id = ?", (sid,), ).fetchone() if cur: conn.execute( "UPDATE user_finance SET matricula_eur = COALESCE(matricula_eur,0) + ? WHERE user_id = ?", (eur, sid), ) else: conn.execute( "INSERT INTO user_finance (user_id, matricula_eur) VALUES (?, ?)", (sid, eur), ) conn.execute("DELETE FROM jugadores_gastos WHERE reservation_id = ?", (rid,)) conn.execute("DELETE FROM reservations WHERE id = ?", (rid,)) conn.commit() return {"ok": True} ALLOWED_STATUSES = {"confirmada", "pendiente", "cancelada"} @app.post("/api/reservations", response_model=dict) async def api_reservations_create(request: Request, payload: ReservationIn): require_docente_or_admin(request) try: raw = await request.json() except Exception: raw = {} paid_flag = False if hasattr(payload, "paid") and payload.paid is not None: paid_flag = bool(payload.paid) else: val = raw.get("paid", raw.get("has_paid")) if isinstance(val, bool): paid_flag = val elif isinstance(val, (int, str)): s = str(val).strip().lower() if s in ("1", "true", "sí", "si", "on"): paid_flag = True date_txt = payload.date.strip() time_txt = payload.time.strip()[:5] try: dt.date.fromisoformat(date_txt) except Exception: raise HTTPException(status_code=400, detail="Fecha inválida (usa YYYY-MM-DD).") try: hh, mm = map(int, time_txt.split(":")) except Exception: raise HTTPException(status_code=400, detail="Hora inválida (usa HH:MM).") if not (0 <= hh < 24 and 0 <= mm < 60): raise HTTPException(status_code=400, detail="Hora inválida.") if hh == 0 and mm == 0: raise HTTPException(status_code=400, detail="No se admiten reservas a medianoche (00:00).") if not (8 <= hh < 22): raise HTTPException(status_code=400, detail="Hora fuera de horario (08:00-22:00).") time_txt = f"{hh:02d}:{mm:02d}" dmin = int(payload.duration_minutes) if dmin <= 0 or dmin > 24*60: raise HTTPException(status_code=400, detail="Duración inválida.") price_cents = int(payload.price_cents or 0) price_eur_raw = getattr(payload, "price_eur", None) if price_eur_raw is None: price_eur_raw = raw.get("price_eur", raw.get("price")) if (not price_cents) and (price_eur_raw is not None and str(price_eur_raw) != ""): try: price_cents = int(round(float(str(price_eur_raw).replace(",", ".")) * 100)) except Exception: price_cents = 0 if price_cents < 0: raise HTTPException(status_code=400, detail="Precio inválido.") status_norm = (payload.status or "confirmada").strip().lower() if status_norm not in ALLOWED_STATUSES: raise HTTPException(status_code=400, detail=f"Estado no permitido: {status_norm}") name = (payload.name or None) email = (payload.email or None) phone = (payload.phone or None) student_ids = list(payload.user_ids or []) if payload.user_id: student_ids.append(payload.user_id) try: student_ids = sorted(set(int(x) for x in student_ids if x is not None)) except Exception: raise HTTPException(status_code=400, detail="IDs de alumnos inválidos.") if student_ids: with _conn() as conn: rows = conn.execute( f"SELECT id, LOWER(COALESCE(role,'')) AS role, name, email, phone FROM users WHERE id IN ({','.join('?'*len(student_ids))})", tuple(student_ids) ).fetchall() found = {int(r["id"]): r for r in rows} bad = [sid for sid in student_ids if sid not in found or found[sid]["role"] != "alumno"] if bad: raise HTTPException(status_code=400, detail=f"IDs no v\ufffdlidos o no alumnos: {bad}") if len(student_ids) == 1: u1 = found[student_ids[0]] if not name: name = u1["name"] if not email: email = u1["email"] if not phone: phone = u1["phone"] docente_id = payload.docente_id if payload.docente_id is not None else None if docente_id is None and "teacher_id" in raw: docente_id = raw.get("teacher_id") if docente_id is not None: try: docente_id = int(docente_id) except Exception: raise HTTPException(status_code=400, detail="docente_id inválido.") if docente_id > 0: with _conn() as conn: d = conn.execute( "SELECT id FROM users WHERE id=? AND LOWER(COALESCE(role,''))='docente'", (docente_id,) ).fetchone() if not d: raise HTTPException(status_code=400, detail="docente_id no corresponde a un docente válido.") else: docente_id = None legacy_user_id = student_ids[0] if len(student_ids) == 1 else None now_txt = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S") per_student_cents = {} try: psc = raw.get("per_student_cents") or {} if isinstance(psc, dict): tmp = {} for k, v in psc.items(): try: kk = int(k) if isinstance(k, str) else int(k) except Exception: continue try: vv = int(v or 0) except Exception: vv = 0 tmp[kk] = max(0, vv) per_student_cents = tmp except Exception: per_student_cents = {} if student_ids: shares = {} if per_student_cents: for sid in student_ids: shares[sid] = max(0, int(per_student_cents.get(sid, price_cents) or 0)) else: for sid in student_ids: shares[sid] = max(0, int(price_cents or 0)) total_price_cents = sum(shares.values()) else: shares = {} total_price_cents = int(price_cents or 0) try: with _conn() as conn: cur = conn.execute(""" INSERT INTO reservations (created_at, date, time, duration_minutes, price_cents, name, email, phone, notes, user_id, status, docente_id, paid) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (now_txt, date_txt, time_txt, dmin, total_price_cents, name, email, phone, payload.notes, legacy_user_id, status_norm, docente_id, 1 if paid_flag else 0)) new_id = cur.lastrowid if student_ids: conn.executemany( "INSERT OR IGNORE INTO reservation_students(reservation_id, user_id) VALUES (?, ?)", [(new_id, sid) for sid in student_ids] ) conn.executemany( "INSERT OR IGNORE INTO user_finance(user_id, matricula_eur) VALUES (?, 0.0)", [(sid,) for sid in student_ids] ) if paid_flag: for sid, share in shares.items(): share_eur = round(share / 100.0, 2) conn.execute( "UPDATE user_finance SET matricula_eur = ROUND(COALESCE(matricula_eur,0.0) - ?, 2) WHERE user_id = ?", (share_eur, sid) ) to_ins = [] concept_base = f"Reserva {date_txt} {time_txt}" for sid, share in shares.items(): if int(share) > 0: to_ins.append(( now_txt, date_txt, sid, new_id, int(share), "reserva", concept_base, None )) if to_ins: conn.executemany(""" INSERT INTO jugadores_gastos (created_at, date, user_id, reservation_id, amount_cents, category, concept, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, to_ins) conn.commit() created_reservation_id = new_id except sqlite3.IntegrityError: with _conn() as conn: row = conn.execute( "SELECT id FROM reservations WHERE date=? AND time=? AND COALESCE(docente_id,0)=COALESCE(?,0)", (date_txt, time_txt, docente_id) ).fetchone() if not row: raise HTTPException(status_code=409, detail="Ya existe una reserva en ese horario.") existing_id = int(row["id"]) current = conn.execute( "SELECT user_id FROM reservation_students WHERE reservation_id=?", (existing_id,) ).fetchall() current_ids = [int(r["user_id"]) for r in current] to_add = [sid for sid in student_ids if sid not in current_ids] if not to_add: created_reservation_id = existing_id else: conn.executemany( "INSERT OR IGNORE INTO reservation_students(reservation_id, user_id) VALUES (?, ?)", [(existing_id, sid) for sid in to_add] ) conn.executemany( "INSERT OR IGNORE INTO user_finance(user_id, matricula_eur) VALUES (?, 0.0)", [(sid,) for sid in to_add] ) to_ins = [] concept_base = f"Reserva {date_txt} {time_txt}" if paid_flag: for sid in to_add: cents = per_student_cents.get(sid, price_cents) try: cents = int(cents or 0) except Exception: cents = 0 share_eur = round(max(0, cents) / 100.0, 2) conn.execute( "UPDATE user_finance SET matricula_eur = ROUND(COALESCE(matricula_eur,0.0) - ?, 2) WHERE user_id = ?", (share_eur, sid) ) if cents > 0: to_ins.append(( now_txt, date_txt, sid, existing_id, int(cents), "reserva", concept_base, None )) if to_ins: conn.executemany(""" INSERT INTO jugadores_gastos (created_at, date, user_id, reservation_id, amount_cents, category, concept, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, to_ins) conn.commit() created_reservation_id = existing_id try: with _conn() as conn2: row = conn2.execute("SELECT * FROM reservations WHERE id = ?", (created_reservation_id,)).fetchone() srows = conn2.execute(""" SELECT u.id, u.name, u.email FROM reservation_students rs JOIN users u ON u.id = rs.user_id WHERE rs.reservation_id = ? ORDER BY u.name """, (created_reservation_id,)).fetchall() recipients = list(srows) if srows else [] if not recipients: recip = None if row["user_id"]: recip = conn2.execute( "SELECT id, name, email FROM users WHERE id = ?", (row["user_id"],) ).fetchone() if not recip and row["email"]: recip = conn2.execute( "SELECT id, name, email FROM users WHERE lower(email) = lower(?)", (row["email"],) ).fetchone() if recip: recipients = [recip] docente_name = None if row["docente_id"] is not None: d = conn2.execute("SELECT name FROM users WHERE id = ?", (row["docente_id"],)).fetchone() docente_name = d["name"] if d else None if recipients: titulo = "Reserva creada" cuerpo = ( f"Se ha creado una reserva para el {row['date']} a las {row['time']}" + (f" con {docente_name}" if docente_name else "") + "." ) try: sender_id = int(current_user(request)["id"]) except Exception: sender_id = 0 now_iso = dt.datetime.utcnow().isoformat(timespec="seconds") conn2.executemany(""" INSERT INTO notifications (created_at, sender_id, recipient_id, title, body, is_read, reservation_id) VALUES (?, ?, ?, ?, ?, 0, ?) """, [ (now_iso, sender_id, int(r["id"]), titulo, cuerpo, created_reservation_id) for r in recipients ]) conn2.commit() except Exception: pass return {"ok": True, "id": created_reservation_id} @app.get("/api/reservations", response_model=List[dict]) async def api_reservations_list( request: Request, date_from: Optional[str] = None, date_to: Optional[str] = None, user_id: Optional[int] = None, status: Optional[str] = None, docente_id: Optional[int] = None, ): require_docente_or_admin(request) def _check_date(s: str) -> None: try: datetime.strptime(s, "%Y-%m-%d") except Exception: raise HTTPException(status_code=400, detail=f"Fecha inválida: {s}. Usa formato YYYY-MM-DD.") if date_from: _check_date(date_from) if date_to: _check_date(date_to) if date_from and date_to and date_from > date_to: date_from, date_to = date_to, date_from status_norm: Optional[str] = None if status: status_norm = status.strip().lower() if status_norm not in ALLOWED_STATUSES: raise HTTPException( status_code=400, detail=f"Estado inválido. Usa uno de: {', '.join(sorted(ALLOWED_STATUSES))}." ) where = [] params: list = [] if date_from: where.append("date >= ?") params.append(date_from) if date_to: where.append("date <= ?") params.append(date_to) if user_id is not None: where.append("user_id = ?") params.append(user_id) if status_norm: where.append("LOWER(status) = ?") params.append(status_norm) if docente_id is not None: where.append("docente_id = ?") params.append(docente_id) clause = (" WHERE " + " AND ".join(where)) if where else "" sql = f""" SELECT id, created_at, date, time, duration_minutes, price_cents, name, email, phone, notes, user_id, status, docente_id, paid FROM reservations {clause} ORDER BY date ASC, time ASC, id ASC """ with _conn() as conn: rows = conn.execute(sql, params).fetchall() return [_reservation_row_to_dict(r) for r in rows] @app.get("/perfil", response_class=HTMLResponse) async def perfil(request: Request): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) me_id = int(me["id"]) if isinstance(me, sqlite3.Row) else int(me.get("id")) me_email = (me["email"] if isinstance(me, sqlite3.Row) else me.get("email") or "").strip() with _conn() as conn: row = conn.execute( """ SELECT COUNT(*) AS c FROM reservations r WHERE r.user_id = ? OR LOWER(COALESCE(r.email,'')) = LOWER(?) """, (me_id, me_email) ).fetchone() reservation_count = row["c"] if row else 0 return templates.TemplateResponse( "profile.html", { "request": request, "user": me, "reservation_count": reservation_count, } ) @app.get("/ajustes", response_class=HTMLResponse) async def ajustes_get(request: Request): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) with _conn() as conn: row_core = conn.execute("SELECT id, name, email, created_at FROM users WHERE id=?", (me["id"],)).fetchone() row_profile = conn.execute("SELECT phone, avatar FROM user_profile WHERE user_id=?", (me["id"],)).fetchone() row_sports = conn.execute("SELECT team, category, position, injury_history FROM user_sports WHERE user_id=?", (me["id"],)).fetchone() ctx = dict(row_core) if row_core else {} if row_profile: ctx.update(dict(row_profile)) if row_sports: ctx.update(dict(row_sports)) return templates.TemplateResponse( "settings.html", { "request": request, "user": ctx, "csrf": request.session.get("csrf"), } ) @app.post("/ajustes", response_class=HTMLResponse) async def ajustes_post( request: Request, name: str = Form(...), email: str = Form(...), phone: str = Form(None), avatar: UploadFile = File(None), team: str = Form(None), category: str = Form(None), position: str = Form(None), injury_history: str = Form(None), csrf: str = Form(...), ): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) require_csrf(request, csrf) name = (name or "").strip() email = (email or "").strip() phone = ((phone or "").strip() or None) team = ((team or "").strip() or None) category = ((category or "").strip() or None) position = ((position or "").strip() or None) injury_history = ((injury_history or "").strip() or None) with _conn() as conn: row = conn.execute("SELECT avatar FROM user_profile WHERE user_id = ?", (me["id"],)).fetchone() current_avatar = row["avatar"] if row else None avatar_path = current_avatar if avatar and getattr(avatar, "filename", None): ext = os.path.splitext(avatar.filename)[1].lower() if ext not in [".png", ".jpg", ".jpeg", ".webp"]: return templates.TemplateResponse("settings.html", { "request": request, "user": me, "error": "Formato de imagen no permitido. Usa PNG, JPG, JPEG o WebP.", "csrf": request.session.get("csrf"), }) uploads_dir = BASE_DIR / "static" / "uploads" uploads_dir.mkdir(parents=True, exist_ok=True) file_path = uploads_dir / f"user_{me['id']}{ext}" with open(file_path, "wb") as f: f.write(await avatar.read()) avatar_path = f"/static/uploads/user_{me['id']}{ext}" try: with _conn() as conn: conn.execute("UPDATE users SET name=?, email=? WHERE id=?", (name, email, me["id"])) conn.execute("INSERT OR IGNORE INTO user_profile(user_id) VALUES (?)", (me["id"],)) conn.execute("UPDATE user_profile SET phone=?, avatar=? WHERE user_id=?", (phone, avatar_path, me["id"])) conn.execute("INSERT OR IGNORE INTO user_sports(user_id) VALUES (?)", (me["id"],)) conn.execute("UPDATE user_sports SET team=?, category=?, position=?, injury_history=? WHERE user_id=?", (team, category, position, injury_history, me["id"])) conn.commit() except sqlite3.IntegrityError: return templates.TemplateResponse("settings.html", { "request": request, "user": me, "error": "El correo ya est� en uso por otra cuenta.", "csrf": request.session.get("csrf"), }) return RedirectResponse(url="/perfil", status_code=303) @app.get("/notes", response_class=HTMLResponse) async def notes(request: Request): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) sql = """ SELECT r.*, su.name AS student_name, su.email AS student_email, du.name AS docente_name, du.email AS docente_email FROM reservations r LEFT JOIN users su ON su.id = r.user_id LEFT JOIN users du ON du.id = r.docente_id WHERE r.notes IS NOT NULL AND TRIM(r.notes) <> '' """ params = [] role = str(me["role"]) if "role" in me.keys() and me["role"] is not None else "" if role == "admin": pass elif role == "docente": sql += " AND r.docente_id = ?" params.append(int(me["id"])) else: sql += " AND r.user_id = ?" params.append(int(me["id"])) sql += " ORDER BY r.created_at DESC, r.id DESC" with _conn() as conn: rows = conn.execute(sql, params).fetchall() def has(row, key): try: return key in row.keys() except Exception: return False def pick(row, *names): for n in names: if has(row, n): val = row[n] if val is not None and str(val).strip() != "": return val return None def student_label(row): return pick(row, "student_name") or pick(row, "name") or pick(row, "email") or "\u2014" def docente_label(row): return pick(row, "docente_name") or pick(row, "docente_email") or "\u2014" items = [] for row in rows: date_txt = pick(row, "date", "fecha") or "" time_txt = pick(row, "time", "hour", "hora") or "" if not date_txt and has(row, "created_at") and row["created_at"]: parts = str(row["created_at"]).split() if len(parts) >= 1: date_txt = parts[0] if len(parts) >= 2: time_txt = parts[1] dt_display = (f"{date_txt} {time_txt}").strip() if date_txt or time_txt else (row["created_at"] if has(row, "created_at") else "") items.append({ "id": row["id"] if has(row, "id") else None, "date": date_txt, "time": time_txt, "dt_display": dt_display, "created_at": row["created_at"] if has(row, "created_at") else None, "notes": row["notes"], "student": student_label(row), "docente": docente_label(row), }) return templates.TemplateResponse( "notes.html", { "request": request, "user": me, "items": items, } ) try: from passlib.hash import bcrypt _HAS_BCRYPT = True except Exception: _HAS_BCRYPT = False def hash_password(plain: str) -> str: plain = plain or "" if _HAS_BCRYPT: return bcrypt.hash(plain) return "sha256$" + hashlib.sha256(plain.encode("utf-8")).hexdigest() def verify_password(plain: str, stored_hash: str) -> bool: plain = plain or "" stored_hash = stored_hash or "" if stored_hash.startswith("$2a$") or stored_hash.startswith("$2b$") or stored_hash.startswith("$2y$"): if _HAS_BCRYPT: try: return bcrypt.verify(plain, stored_hash) except Exception: return False return False if stored_hash.startswith("sha256$"): expected = stored_hash.split("sha256$", 1)[1] digest = hashlib.sha256(plain.encode("utf-8")).hexdigest() return hmac.compare_digest(digest, expected) return False @app.post("/ajustes/password", response_class=HTMLResponse) async def ajustes_password( request: Request, current_password: str = Form(...), new_password: str = Form(...), confirm_password: str = Form(...), csrf: str = Form(...), ): me = current_user(request) if not me: return RedirectResponse(url="/login", status_code=303) require_csrf(request, csrf) new_password = (new_password or "").strip() confirm_password = (confirm_password or "").strip() if new_password != confirm_password: return templates.TemplateResponse("settings.html", { "request": request, "user": me, "error": "Las contrase�as no coinciden.", }) if len(new_password) < 8: return templates.TemplateResponse("settings.html", { "request": request, "user": me, "error": "La nueva contrase�a debe tener al menos 8 caracteres.", }) stored_hash = me.get("password_hash") if isinstance(me, dict) else me["password_hash"] if not verify_password(current_password, stored_hash): return templates.TemplateResponse("settings.html", { "request": request, "user": me, "error": "La contrase�a actual no es correcta.", }) new_hash = hash_password(new_password) with _conn() as conn: conn.execute("UPDATE users SET password_hash=? WHERE id=?", (new_hash, me["id"])) conn.commit() return RedirectResponse(url="/ajustes", status_code=303) class MatriculaPayload(BaseModel): matricula_cents: int = Field(ge=0) mode: Optional[str] = "add" @app.patch("/api/users/{user_id}/matricula", response_model=dict) async def api_users_update_matricula(request: Request, user_id: int, payload: dict = Body(...)): require_admin(request) mode = str(payload.get("mode", "set")).lower() if mode not in {"set", "add"}: raise HTTPException(status_code=400, detail="mode debe ser 'set' o 'add'.") has_eur = "matricula_eur" in payload has_cents = "matricula_cents" in payload if not has_eur and not has_cents: raise HTTPException(status_code=400, detail="Falta matricula_eur.") if has_eur: try: euros = round(float(str(payload["matricula_eur"]).replace(",", ".")), 2) except Exception: raise HTTPException(status_code=400, detail="matricula_eur inv�lido.") else: try: cents = int(payload["matricula_cents"]) except Exception: raise HTTPException(status_code=400, detail="matricula_cents inv�lido.") euros = round(cents / 100.0, 2) with _conn() as conn: row = conn.execute("SELECT matricula_eur FROM user_finance WHERE user_id = ?", (user_id,)).fetchone() if not row: conn.execute("INSERT INTO user_finance (user_id, matricula_eur) VALUES (?, ?)", (user_id, 0.0)) current = 0.0 else: current = float(row[0] or 0.0) if mode == "set": new_val = euros else: new_val = round(current + euros, 2) conn.execute("UPDATE user_finance SET matricula_eur = ? WHERE user_id = ?", (new_val, user_id)) return { "ok": True, "matricula_eur": new_val, "matricula_cents": int(round(new_val * 100)) } from fastapi import Path VALID_ROLES = {"alumno", "docente", "admin"} @app.patch("/api/users/{user_id}/role", response_model=dict) async def api_users_update_role(request: Request, user_id: int = Path(...), payload: dict = Body(...)): require_admin(request) target_role = (payload.get("role") or "").strip().lower() if target_role not in VALID_ROLES: raise HTTPException(status_code=400, detail="Rol inválido") with _conn() as conn: row = conn.execute("SELECT id, role FROM users WHERE id = ?", (user_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="Usuario no encontrado") conn.execute("UPDATE users SET role = ? WHERE id = ?", (target_role, user_id)) conn.commit() updated = conn.execute("SELECT id, name, email, role, created_at FROM users WHERE id = ?", (user_id,)).fetchone() return {"ok": True, "user": dict(updated)} @app.post("/api/ratings", response_class=JSONResponse, name="api_ratings_create") async def api_ratings_create(request: Request, payload: RatingIn): me = require_docente_or_admin(request) player = get_user_by_id(payload.player_id) if not player: raise HTTPException(status_code=404, detail="Jugador no encontrado") if str(player["role"]).lower() != "alumno": raise HTTPException(status_code=400, detail="Solo se pueden valorar alumnos") score = int(payload.score) comment = (payload.comment or "").strip() now = datetime.utcnow().isoformat(timespec="seconds") with _conn() as conn: conn.execute( "INSERT INTO player_ratings (player_id, docente_id, score, comment, created_at) VALUES (?,?,?,?,?)", (int(payload.player_id), int(me["id"]), score, comment, now) ) return {"ok": True} @app.get("/api/ratings", response_class=JSONResponse, name="api_ratings_list") async def api_ratings_list(request: Request, player_id: Optional[int] = Query(None)): me = require_docente_or_admin(request) params = [] where = "" if player_id: where = "WHERE r.player_id = ?" params.append(int(player_id)) with _conn() as conn: cur = conn.execute(f""" SELECT r.id, r.player_id, r.docente_id, r.score, r.comment, r.created_at, p.name AS player_name, d.name AS docente_name FROM player_ratings r JOIN users p ON p.id = r.player_id JOIN users d ON d.id = r.docente_id {where} ORDER BY r.id DESC """, params) data = [dict(x) for x in cur.fetchall()] return data @app.get("/logout") async def logout(request: Request): request.session.clear() return RedirectResponse(url="/", status_code=303) @app.get("/api/health") async def health(): return {"status": "ok"} from pydantic import BaseModel, Field from typing import List, Optional import secrets class ValoracionItem(BaseModel): area: str subescala: str indicador: str score: int = Field(ge=1, le=5) observaciones: Optional[str] = None class ValoracionJugadorIn(BaseModel): player_id: int items: List[ValoracionItem] @app.get("/api/valoraciones_jugador", response_model=list[dict]) async def api_valoraciones_jugador_list(request: Request, player_id: int = Query(...)): require_docente(request) with _conn() as conn: rows = conn.execute(""" SELECT eval_id, created_at, docente_id, area, subescala, indicador, score, COALESCE(observaciones,'') AS observaciones, (SELECT name FROM users WHERE id = docente_id) AS docente_name FROM valoraciones_jugador WHERE player_id = ? ORDER BY created_at DESC, eval_id, area, subescala, indicador """, (player_id,)).fetchall() return [dict(r) for r in rows] @app.delete("/api/users/{user_id}", response_model=dict) async def api_users_delete(request: Request, user_id: int): require_admin(request) me = current_user(request) me_id = None try: me_id = int(me["id"]) except Exception: try: me_id = int(dict(me).get("id")) except Exception: me_id = None if me_id is not None and me_id == int(user_id): raise HTTPException(status_code=400, detail="No puedes eliminar tu propio usuario.") with _conn() as conn: conn.row_factory = sqlite3.Row row = conn.execute("SELECT id FROM users WHERE id = ?", (user_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="Usuario no encontrado.") try: conn.execute("DELETE FROM reservation_students WHERE user_id = ?", (user_id,)) except Exception: pass try: conn.execute("DELETE FROM notifications WHERE sender_id = ? OR recipient_id = ?", (user_id, user_id)) except Exception: pass try: conn.execute("DELETE FROM user_profile WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM user_sports WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM user_health WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM user_consent WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM user_guardians WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM user_marketing WHERE user_id = ?", (user_id,)) conn.execute("DELETE FROM user_finance WHERE user_id = ?", (user_id,)) except Exception: pass conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) return {"ok": True, "deleted": int(user_id)} from fastapi import HTTPException, Request from fastapi import HTTPException, Request import sqlite3 def require_docente(request: Request): me = current_user(request) if not me: raise HTTPException(status_code=403, detail="Se requiere rol docente.") try: role = me["role"] except Exception: role = getattr(me, "role", None) if role not in ("docente", "admin"): raise HTTPException(status_code=403, detail="Se requiere rol docente.") return me def require_csrf_header(request: Request): token = request.headers.get("X-CSRF-Token") or request.headers.get("X-Csrf-Token") or "" require_csrf(request, token) @app.post("/api/valoraciones_jugador", response_model=dict) async def api_valoraciones_jugador_create(request: Request, payload: ValoracionJugadorIn): me = require_docente(request) require_csrf_header(request) if not payload.items: raise HTTPException(status_code=400, detail="Sin �tems en la evaluaci�n.") eval_id = secrets.token_hex(8) with _conn() as conn: cur = conn.cursor() for it in payload.items: cur.execute(""" INSERT INTO valoraciones_jugador (eval_id, player_id, docente_id, area, subescala, indicador, score, observaciones) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( eval_id, payload.player_id, me["id"], it.area.strip(), it.subescala.strip(), it.indicador.strip(), int(it.score), (it.observaciones or "").strip() or None )) conn.commit() return {"ok": True, "eval_id": eval_id} from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request: Request, exc: StarletteHTTPException): try: user_ctx = current_user(request) except Exception: user_ctx = None if exc.status_code == HTTP_404_NOT_FOUND: return templates.TemplateResponse("404.html", {"request": request, "user": user_ctx}, status_code=HTTP_404_NOT_FOUND) if exc.status_code == HTTP_403_FORBIDDEN: return templates.TemplateResponse("403.html", {"request": request, "user": user_ctx}, status_code=HTTP_403_FORBIDDEN) return JSONResponse({"detail": exc.detail}, status_code=exc.status_code) def get_unread_notifications_for_user(user_id: int): with _conn() as conn: rows = conn.execute(""" SELECT id, created_at, title, body, sender_id, recipient_id, reservation_id FROM notifications WHERE recipient_id=? AND COALESCE(is_read,0)=0 AND COALESCE(deleted_by_recipient,0)=0 ORDER BY created_at DESC """, (user_id,)).fetchall() return rows @app.on_event("startup") def on_startup(): try: with _conn() as conn: conn.execute(""" UPDATE reservations SET name = (SELECT name FROM users WHERE users.id = reservations.user_id) WHERE (name IS NULL OR name='') AND user_id IS NOT NULL """) conn.commit() except Exception: pass init_db() try: with _conn() as conn: conn.execute("PRAGMA optimize;") conn.execute("PRAGMA wal_checkpoint(TRUNCATE);") except Exception: pass purge_expired_temp_regs()