admin/server.py - main - amne/dca-agent

admin/server.py
601 строка22 KB
amne
feat: Phase 2 — pipeline_steps tracking + API
11 авг 2026, 14:52
11 авг 2026, 14:522b38c25
import logging
import time
import asyncio
import json
import os
from datetime import datetime, timedelta, timezone
from contextlib import asynccontextmanager

import jwt
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse, StreamingResponse
from starlette.middleware.cors import CORSMiddleware

import db
import proxy

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s")
logger = logging.getLogger("dca-admin")

# --- JWT Config ---
JWT_SECRET = os.environ.get("JWT_SECRET", "")
if not JWT_SECRET:
    secret_path = os.path.join(os.path.dirname(__file__), ".jwt_secret")
    if os.path.exists(secret_path):
        with open(secret_path, "r") as f:
            JWT_SECRET = f.read().strip()
if not JWT_SECRET:
    # Fallback: generate ephemeral (sessions invalidated on restart)
    import secrets as _secrets
    JWT_SECRET = _secrets.token_hex(32)
    logger.warning("[security] No .jwt_secret found — using ephemeral secret (sessions will not survive restart)")

JWT_ALGORITHM = "HS256"
JWT_EXPIRE_HOURS = 24

# --- Per-client concurrency limits ---
_client_semaphores: dict[int, asyncio.Semaphore] = {}
MAX_CONCURRENT_PROCESS = 5  # per client

# --- SSE Broadcaster ---
_sse_listeners: list[asyncio.Queue] = []

def sse_broadcast(event: str, data: dict):
    """Send event to all connected SSE clients."""
    msg = json.dumps({"event": event, "data": data}, ensure_ascii=False, default=str)
    dead = []
    for i, q in enumerate(_sse_listeners):
        try:
            q.put_nowait(msg)
        except asyncio.QueueFull:
            dead.append(i)
    for i in reversed(dead):
        _sse_listeners.pop(i)

async def sse_generator(q: asyncio.Queue):
    """Yield SSE events from queue, keep alive with comments."""
    try:
        while True:
            try:
                msg = await asyncio.wait_for(q.get(), timeout=30)
                yield f"data: {msg}\n\n"
            except asyncio.TimeoutError:
                yield ": keepalive\n\n"
    except asyncio.CancelledError:
        pass
    finally:
        if q in _sse_listeners:
            _sse_listeners.remove(q)

# --- Lifespan ---
@asynccontextmanager
async def lifespan(app: FastAPI):
    await db.init_db()
    logger.info("[admin] DB initialized")
    # Shared aiohttp session — reuses TCP connections instead of creating new per request
    import aiohttp
    app.state.http_session = aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=600),
        connector=aiohttp.TCPConnector(limit=20, limit_per_host=10)
    )
    logger.info("[admin] HTTP session pool created")
    # Schedule periodic heartbeat cleanup
    cleanup_task = asyncio.create_task(_heartbeat_cleanup_loop())
    yield
    # Cleanup
    cleanup_task.cancel()
    await app.state.http_session.close()
    logger.info("[admin] shutdown complete")


async def _heartbeat_cleanup_loop():
    """Delete heartbeats older than 30 days, every 24h."""
    while True:
        try:
            await asyncio.sleep(86400)  # 24h
            await db.cleanup_old_heartbeats(days=30)
            logger.info("[cleanup] old heartbeats purged")
        except asyncio.CancelledError:
            break
        except Exception as e:
            logger.error(f"[cleanup] heartbeat cleanup error: {e}")

app = FastAPI(title="DCA Admin", version="1.0", lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://i.vhos.ru",
        "http://localhost:8003",
        "http://127.0.0.1:8003",
    ],
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Client-ID"],
)


# --- Auth ---
async def verify_agent(request: Request):
    auth = request.headers.get("Authorization", "")
    client_id = request.headers.get("X-Client-ID", "")
    if not auth.startswith("Bearer ") or not client_id:
        raise HTTPException(401, "Missing auth headers")
    token = auth[7:]
    client = await db.authenticate_client(client_id, token)
    if not client:
        raise HTTPException(403, "Invalid client credentials")
    return client


import bcrypt

def _create_jwt(email: str, name: str) -> str:
    """Create a signed JWT for an admin user."""
    payload = {
        "email": email,
        "name": name,
        "exp": datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS),
        "iat": datetime.now(timezone.utc),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)

async def verify_admin(request: Request):
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(401, "Missing admin token")
    token = auth[7:]
    
    # Decode and verify JWT
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(403, "Invalid admin token")
    
    email = payload.get("email")
    if not email:
        raise HTTPException(403, "Invalid token payload")
    
    # Single query by email — no iteration, no bcrypt on every request
    import aiosqlite
    async with aiosqlite.connect(db.DB_PATH) as conn:
        conn.row_factory = aiosqlite.Row
        cur = await conn.execute("SELECT * FROM admin_users WHERE email = ?", (email,))
        row = await cur.fetchone()
    
    if not row:
        raise HTTPException(403, "Admin user not found")
    return dict(row)


# ==========================================
# Agent API
# ==========================================

@app.post("/api/agent/register")
async def agent_register(request: Request, client: dict = Depends(verify_agent)):
    config = await db.get_pipeline_config(client["id"])
    if not config:
        raise HTTPException(404, "No pipeline config")
    return _config_response(config)


@app.post("/api/agent/config")
async def agent_config(request: Request, client: dict = Depends(verify_agent)):
    config = await db.get_pipeline_config(client["id"])
    if not config:
        raise HTTPException(404, "No pipeline config")
    return _config_response(config)


@app.post("/api/agent/process")
async def agent_process(request: Request, client: dict = Depends(verify_agent)):
    """
    Agent sends audio file, server proxies to Nexara + LLM, returns result.
    Agent never touches external APIs directly.
    """
    # Get pipeline config for this client
    config = await db.get_pipeline_config(client["id"])
    if not config:
        raise HTTPException(404, "No pipeline config")

    if not config.get("nexara_url") or not config.get("nexara_key"):
        raise HTTPException(400, "STT provider not configured")
    if not config.get("llm_url") or not config.get("llm_key"):
        raise HTTPException(400, "LLM provider not configured")

    # --- Daily call limit check ---
    max_calls = config.get("max_calls_per_day", 100)
    today_count = await db.get_today_call_count(client["id"])
    if today_count >= max_calls:
        logger.warning(f"[agent] client {client['id']} hit daily limit: {today_count}/{max_calls}")
        raise HTTPException(429, f"Daily call limit reached ({max_calls})")

    # --- Per-client concurrency limit ---
    cid = client["id"]
    if cid not in _client_semaphores:
        _client_semaphores[cid] = asyncio.Semaphore(MAX_CONCURRENT_PROCESS)
    sem = _client_semaphores[cid]
    if sem.locked():
        raise HTTPException(429, "Too many concurrent requests — try again shortly")
    async with sem:
        # Read audio from multipart form
        form = await request.form()
        audio_file = form.get("file")
        if not audio_file:
            raise HTTPException(400, "No audio file in request")

        filename = form.get("filename", audio_file.filename or "audio.mp3")
        audio_data = await audio_file.read()

        if not audio_data:
            raise HTTPException(400, "Empty audio file")

        max_mb = config.get("max_file_size_mb", 50)
        if len(audio_data) > max_mb * 1024 * 1024:
            raise HTTPException(400, f"File too large (max {max_mb}MB)")

        logger.info(f"[agent] processing audio from client {client['id']}: {filename} "
                    f"({len(audio_data)/1024/1024:.1f}MB)")

        try:
            result = await proxy.process_audio(request.app.state.http_session, config, audio_data, filename, client_id=client["id"])
        except Exception as e:
            logger.error(f"[agent] processing failed for {filename}: {e}")
            await db.save_call_stat(client["id"], {
                "filename": filename,
                "duration_sec": 0,
                "tokens_in": 0,
                "tokens_out": 0,
                "status": "error",
                "error": str(e),
            })
            sse_broadcast("call_error", {"client_id": client["id"], "filename": filename, "error": str(e)})
            raise HTTPException(500, f"Processing failed: {e}")

        await db.save_call_stat(client["id"], {
            "filename": filename,
            "duration_sec": result["duration"],
            "tokens_in": result["tokens_in"],
            "tokens_out": result["tokens_out"],
            "status": "done",
        })

        # Save full result on server (safety net against client DB loss)
        try:
            await db.save_call_result(client["id"], filename, result)
        except Exception as e:
            logger.warning(f"[agent] failed to save call result: {e}")

        sse_broadcast("call_processed", {"client_id": client["id"], "filename": filename, "status": "done"})

        logger.info(f"[agent] done {filename}: dur={result['duration']:.1f}s "
                    f"tokens={result['tokens_in']}+{result['tokens_out']}")

        return JSONResponse(result)


@app.post("/api/agent/report")
async def agent_report(request: Request, client: dict = Depends(verify_agent)):
    """Agent sends billing stats only (no report text)"""
    body = await request.json()
    await db.save_call_stat(client["id"], body)
    logger.info(f"[agent] stat from client {client['id']}: {body.get('filename', '?')} "
                f"dur={body.get('duration_sec', 0):.0f}s tokens={body.get('tokens_in', 0)}+{body.get('tokens_out', 0)}")
    return {"ok": True}


@app.post("/api/agent/heartbeat")
async def agent_heartbeat(request: Request, client: dict = Depends(verify_agent)):
    body = await request.json()
    await db.save_heartbeat(client["id"], body)
    sse_broadcast("agent_heartbeat", {"client_id": client["id"], "data": body})
    return {"ok": True}


@app.get("/api/agent/subscription")
async def agent_subscription(client: dict = Depends(verify_agent)):
    """Return subscription status: days/minutes used vs limits."""
    stats = await db.get_client_stats(client["id"])
    config = await db.get_pipeline_config(client["id"])
    max_calls = config.get("max_calls_per_day", 100) if config else 100

    # Estimate: each call ~2 min average
    minutes_used = int(stats.get("total_min", 0))
    today_calls = stats.get("today_calls", 0)

    # Days left: based on daily limit and remaining in billing cycle
    # For now: no hard expiry — return generous defaults
    return {
        "status": "active" if client.get("is_active", 1) else "suspended",
        "days_left": 365,
        "days_limit": 365,
        "minutes_used": minutes_used,
        "minutes_limit": max_calls * 2,  # rough estimate
    }


@app.get("/api/agent/pipeline-status/{filename:path}")
async def pipeline_status(filename: str, client: dict = Depends(verify_agent)):
    """Get pipeline step statuses for a file."""
    steps = await db.get_pipeline_steps(filename)
    if not steps:
        raise HTTPException(404, "No pipeline steps found for this file")
    return {"filename": filename, "steps": steps}


@app.post("/api/agent/weekly-report")
async def agent_weekly_report(request: Request, client: dict = Depends(verify_agent)):
    """Agent sends aggregated call data, server generates markdown via LLM."""
    try:
        body = await request.json()
    except Exception:
        raise HTTPException(400, "Invalid JSON body")

    report_type = body.get("report_type", "clinic")
    date_from = body.get("date_from", "")
    date_to = body.get("date_to", "")
    operator = body.get("operator", "")
    calls = body.get("calls", [])

    if not calls:
        raise HTTPException(400, "No call data provided")

    config = await db.get_pipeline_config(client["id"])
    if not config:
        raise HTTPException(400, "Pipeline not configured")

    # Build LLM prompt based on report type
    if report_type == "operator":
        system_prompt = (
            "Ты — аналитик стоматологической клиники. Создай еженедельный отчёт "
            "по работе оператора на основе данных о звонках. "
            "Отчёт должен включать: общую оценку, сильные стороны, зоны роста, "
            "конкретные рекомендации. Пиши на русском, в формате markdown."
        )
    else:
        system_prompt = (
            "Ты — аналитик стоматологической клиники. Создай еженедельный отчёт "
            "по клинике на основе данных о звонках. "
            "Отчёт должен включать: общую статистику, распределение по категориям, "
            "красные флаги, рекомендации. Пиши на русском, в формате markdown."
        )

    # Build user message from call data
    import json as _json
    user_msg = f"Период: {date_from} — {date_to}\n"
    if operator:
        user_msg += f"Оператор: {operator}\n"
    user_msg += f"Звонков в отчёте: {len(calls)}\n\n"
    user_msg += "Данные по звонкам (JSON):\n"
    user_msg += _json.dumps(calls, ensure_ascii=False, indent=2)[:50000]  # cap at 50K chars

    # Call LLM
    llm_url = config.get("llm_url", "")
    llm_key = config.get("llm_key", "")
    llm_model = config.get("llm_model", "deepseek-v4-pro")
    llm_temp = config.get("llm_temperature", 0.3)

    if not llm_url or not llm_key:
        raise HTTPException(400, "LLM not configured")

    try:
        markdown, _, _ = await proxy.call_llm(
            request.app.state.http_session, llm_url, llm_key, llm_model, llm_temp,
            system_prompt, user_msg, max_tokens=20000
        )
    except Exception as e:
        logger.error(f"[agent] weekly report LLM failed: {e}")
        raise HTTPException(500, f"Report generation failed: {e}")

    logger.info(f"[agent] weekly report generated for client {client['id']}: "
                f"{report_type}, {len(calls)} calls, {len(markdown)} chars")

    return {"markdown": markdown}


def _config_response(config: dict) -> dict:
    """Return non-sensitive config to agent. No API keys!"""
    return {
        "llm_model": config.get("llm_model", "deepseek-v4-pro"),
        "llm_temperature": config.get("llm_temperature", 0.2),
        "normalize_template": config.get("normalize_template", ""),
        "report_template": config.get("report_template", ""),
        "max_file_size_mb": config.get("max_file_size_mb", 50),
        "max_calls_per_day": config.get("max_calls_per_day", 100),
    }


# ==========================================
# Admin API
# ==========================================

@app.post("/api/admin/login")
async def admin_login(request: Request):
    body = await request.json()
    email = body.get("email", "")
    password = body.get("password", "")
    import aiosqlite
    async with aiosqlite.connect(db.DB_PATH) as conn:
        conn.row_factory = aiosqlite.Row
        cur = await conn.execute("SELECT * FROM admin_users WHERE email = ?", (email,))
        row = await cur.fetchone()
    if not row:
        raise HTTPException(403, "Invalid credentials")
    if not bcrypt.checkpw(password.encode(), row["password_hash"].encode()):
        raise HTTPException(403, "Invalid credentials")
    # Return JWT — NOT the raw password
    token = _create_jwt(row["email"], row["display_name"])
    return {"token": token, "email": row["email"], "name": row["display_name"]}


@app.get("/api/admin/clients")
async def admin_list_clients(admin: dict = Depends(verify_admin)):
    clients = await db.list_clients()
    for c in clients:
        c["stats"] = await db.get_client_stats(c["id"])
        c["last_heartbeat"] = await db.get_latest_heartbeat(c["id"])
    return {"clients": clients}


@app.post("/api/admin/clients")
async def admin_create_client(request: Request, admin: dict = Depends(verify_admin)):
    body = await request.json()
    name = body.get("name", "")
    slug = body.get("slug", "")
    tariff = body.get("tariff", "basic")
    price = body.get("price_per_call", 20.0)
    if not name:
        raise HTTPException(400, "name is required")
    client = await db.create_client(name, slug, tariff, price)
    logger.info(f"[admin] created client: {client['name']} ({client['slug']})")
    return client


@app.get("/api/admin/clients/{client_id}")
async def admin_get_client(client_id: int, admin: dict = Depends(verify_admin)):
    client = await db.get_client(client_id)
    if not client:
        raise HTTPException(404, "Client not found")
    config = await db.get_pipeline_config(client_id)
    stats = await db.get_client_stats(client_id)
    hb = await db.get_latest_heartbeat(client_id)
    return {"client": client, "config": config, "stats": stats, "heartbeat": hb}


@app.put("/api/admin/clients/{client_id}/config")
async def admin_update_config(client_id: int, request: Request, admin: dict = Depends(verify_admin)):
    body = await request.json()
    config = await db.update_pipeline_config(client_id, body)
    logger.info(f"[admin] updated config for client {client_id}")
    return config


@app.put("/api/admin/clients/{client_id}/toggle")
async def admin_toggle_client(client_id: int, request: Request, admin: dict = Depends(verify_admin)):
    client = await db.get_client(client_id)
    if not client:
        raise HTTPException(404, "Client not found")
    new_state = 0 if client["is_active"] else 1
    import aiosqlite
    async with aiosqlite.connect(db.DB_PATH) as conn:
        await conn.execute("UPDATE clients SET is_active = ? WHERE id = ?", (new_state, client_id))
        await conn.commit()
    return {"is_active": new_state}


@app.put("/api/admin/clients/{client_id}")
async def admin_update_client(client_id: int, request: Request, admin: dict = Depends(verify_admin)):
    body = await request.json()
    fields = []
    values = []
    for k in ["name", "slug", "tariff", "price_per_call"]:
        if k in body:
            fields.append(f"{k} = ?")
            values.append(body[k])
    if not fields:
        raise HTTPException(400, "No fields to update")
    values.append(client_id)
    import aiosqlite
    async with aiosqlite.connect(db.DB_PATH) as conn:
        await conn.execute(f"UPDATE clients SET {', '.join(fields)} WHERE id = ?", values)
        await conn.commit()
    return await db.get_client(client_id)


@app.get("/api/admin/clients/{client_id}/stats")
async def admin_client_stats(client_id: int, limit: int = 50, offset: int = 0, admin: dict = Depends(verify_admin)):
    stats = await db.list_call_stats(client_id, limit, offset)
    summary = await db.get_client_stats(client_id)
    return {"summary": summary, "records": stats}


# ==========================================
# Pipelines
# ==========================================

@app.get("/api/admin/pipelines")
async def admin_list_pipelines(admin: dict = Depends(verify_admin)):
    pipelines = await db.list_pipelines()
    return {"pipelines": pipelines}

@app.post("/api/admin/pipelines")
async def admin_create_pipeline(request: Request, admin: dict = Depends(verify_admin)):
    body = await request.json()
    if not body.get("name"):
        raise HTTPException(400, "name is required")
    p = await db.create_pipeline(body)
    return p


@app.post("/api/admin/pipelines/{pipeline_id}/copy")
async def admin_copy_pipeline(pipeline_id: int, request: Request, admin: dict = Depends(verify_admin)):
    body = await request.json()
    new_name = body.get("name", "")
    if not new_name:
        raise HTTPException(400, "name is required")
    p = await db.copy_pipeline(pipeline_id, new_name)
    if not p:
        raise HTTPException(404, "Pipeline not found")
    logger.info(f"[admin] copied pipeline {pipeline_id} -> {p.get('id')} ({new_name})")
    return p

@app.get("/api/admin/pipelines/{pipeline_id}")
async def admin_get_pipeline(pipeline_id: int, admin: dict = Depends(verify_admin)):
    p = await db.get_pipeline(pipeline_id)
    if not p:
        raise HTTPException(404, "Pipeline not found")
    return p

@app.put("/api/admin/pipelines/{pipeline_id}")
async def admin_update_pipeline(pipeline_id: int, request: Request, admin: dict = Depends(verify_admin)):
    body = await request.json()
    p = await db.update_pipeline(pipeline_id, body)
    if not p:
        raise HTTPException(404, "Pipeline not found")
    return p

@app.delete("/api/admin/pipelines/{pipeline_id}")
async def admin_delete_pipeline(pipeline_id: int, admin: dict = Depends(verify_admin)):
    await db.delete_pipeline(pipeline_id)
    return {"ok": True}


# ==========================================
# SSE (Server-Sent Events)
# ==========================================

@app.get("/api/events")
async def sse_events(request: Request):
    q = asyncio.Queue(maxsize=100)
    _sse_listeners.append(q)
    return StreamingResponse(sse_generator(q), media_type="text/event-stream", headers={
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
    })


# ==========================================
# SPA
# ==========================================

@app.get("/", response_class=HTMLResponse)
async def spa_index():
    return FileResponse("static/index.html")


app.mount("/static", StaticFiles(directory="static"), name="static")


@app.get("/healthz")
async def healthz():
    """Health check for Caddy / monitoring."""
    return {"status": "ok", "ts": int(time.time())}


if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", "8003"))
    uvicorn.run("server:app", host="0.0.0.0", port=port, reload=False)