hechakuaa_bot/admin/app.py

"""
admin/app.py — Panel de administración interno (FastAPI).

Solo escucha en 127.0.0.1. Acceso via SSH tunnel:
    ssh -L 9090:127.0.0.1:9090 pedrito@<IP_VPS>
    Luego: http://localhost:9090

Autenticación: header X-Admin-Token en cada request.
"""
from __future__ import annotations

import json
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any

import aiosqlite
from fastapi import Depends, FastAPI, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates

from admin.auth import require_token

_db_path: str = ""
_admin_token: str = ""
_data_dir: Path = Path("./data")

templates = Jinja2Templates(directory="admin/templates")


@asynccontextmanager
async def _lifespan(app: FastAPI):
    app.state.db = await aiosqlite.connect(_db_path)
    app.state.db.row_factory = aiosqlite.Row
    yield
    await app.state.db.close()


def create_app(db_path: str, admin_token: str, data_dir: Path) -> FastAPI:
    global _db_path, _admin_token, _data_dir
    _db_path = db_path
    _admin_token = admin_token
    _data_dir = data_dir

    app = FastAPI(lifespan=_lifespan, docs_url=None, redoc_url=None)

    auth = Depends(require_token(admin_token))

    # ── helpers ───────────────────────────────────────────────────────────────

    def _mask_cedula(credentials_enc: str | None) -> str:
        """Muestra '***' — la cédula está cifrada y no se expone en el panel."""
        if not credentials_enc:
            return "—"
        return "***"

    def _parse_detalle(detalle_json: str | None) -> str:
        if not detalle_json:
            return ""
        try:
            return json.dumps(json.loads(detalle_json), ensure_ascii=False, indent=None)
        except Exception:
            return detalle_json or ""

    # ── rutas ─────────────────────────────────────────────────────────────────

    @app.get("/", include_in_schema=False)
    async def root():
        return RedirectResponse(url="/tenants")

    @app.get("/tenants", response_class=HTMLResponse, dependencies=[auth])
    async def lista_tenants(request: Request):
        db: aiosqlite.Connection = request.app.state.db
        async with db.execute("SELECT * FROM tenants ORDER BY created_at DESC") as cur:
            rows = [dict(r) for r in await cur.fetchall()]
        for r in rows:
            r["cedula_masked"] = _mask_cedula(r.get("credentials_enc"))
            r.pop("credentials_enc", None)
        return templates.TemplateResponse("tenants.html", {"request": request, "tenants": rows})

    @app.get("/tenants/{chat_id}", response_class=HTMLResponse, dependencies=[auth])
    async def detalle_tenant(request: Request, chat_id: int):
        db: aiosqlite.Connection = request.app.state.db
        async with db.execute("SELECT * FROM tenants WHERE chat_id = ?", (chat_id,)) as cur:
            row = await cur.fetchone()
        if not row:
            return HTMLResponse("<h1>Tenant no encontrado</h1>", status_code=404)
        tenant: dict[str, Any] = dict(row)
        tenant["cedula_masked"] = _mask_cedula(tenant.get("credentials_enc"))
        tenant.pop("credentials_enc", None)

        async with db.execute(
            "SELECT * FROM audit_log WHERE chat_id = ? ORDER BY timestamp DESC LIMIT 20",
            (chat_id,),
        ) as cur:
            logs = [dict(r) for r in await cur.fetchall()]
        for entry in logs:
            entry["detalle_str"] = _parse_detalle(entry.get("detalle"))

        return templates.TemplateResponse(
            "tenant.html", {"request": request, "tenant": tenant, "logs": logs}
        )

    @app.post("/tenants/{chat_id}/resetear", dependencies=[auth])
    async def resetear_tenant(request: Request, chat_id: int):
        db: aiosqlite.Connection = request.app.state.db
        await db.execute("DELETE FROM tenants WHERE chat_id = ?", (chat_id,))
        await db.execute("DELETE FROM audit_log WHERE chat_id = ?", (chat_id,))
        await db.commit()
        snapshot = _data_dir / f"snapshot_{chat_id}.json"
        if snapshot.exists():
            snapshot.unlink()
        return RedirectResponse(url="/tenants", status_code=303)

    @app.get("/audit", response_class=HTMLResponse, dependencies=[auth])
    async def audit_log(request: Request, chat_id: str = "", page: int = 1):
        db: aiosqlite.Connection = request.app.state.db
        limit = 200
        offset = (page - 1) * limit
        if chat_id.strip():
            try:
                cid = int(chat_id.strip())
                async with db.execute(
                    "SELECT * FROM audit_log WHERE chat_id = ? ORDER BY timestamp DESC LIMIT ? OFFSET ?",
                    (cid, limit, offset),
                ) as cur:
                    logs = [dict(r) for r in await cur.fetchall()]
                async with db.execute(
                    "SELECT COUNT(*) FROM audit_log WHERE chat_id = ?", (cid,)
                ) as cur:
                    total = (await cur.fetchone())[0]
            except ValueError:
                logs, total = [], 0
        else:
            async with db.execute(
                "SELECT * FROM audit_log ORDER BY timestamp DESC LIMIT ? OFFSET ?",
                (limit, offset),
            ) as cur:
                logs = [dict(r) for r in await cur.fetchall()]
            async with db.execute("SELECT COUNT(*) FROM audit_log") as cur:
                total = (await cur.fetchone())[0]

        for entry in logs:
            entry["detalle_str"] = _parse_detalle(entry.get("detalle"))

        has_more = (offset + limit) < total
        return templates.TemplateResponse(
            "audit.html",
            {
                "request": request,
                "logs": logs,
                "chat_id_filter": chat_id,
                "page": page,
                "has_more": has_more,
                "total": total,
            },
        )

    return app