ticket/web/api/app.py

from __future__ import annotations

from datetime import date, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from typing import Any, Literal
import base64
import hashlib
import hmac
import os
import secrets
import sys

import psycopg2
from psycopg2.extras import DictCursor
from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response, status
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field


ROOT_DIR = Path(__file__).resolve().parents[2]
API_DIR = ROOT_DIR / "api"
CONN_DIR = ROOT_DIR / "conn"
for path in (API_DIR, CONN_DIR):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

import ai_retriever
import config
import mysql5
from bootstrap_db import ensure_bot_schema
from ai_index import ensure_ai_schema


COOKIE_NAME = "ticket_web_session"
SESSION_DAYS = int(os.getenv("TICKET_WEB_SESSION_DAYS", "7"))
STATIC_DIR = Path(os.getenv("TICKET_WEB_STATIC_DIR", ROOT_DIR / "web" / "static")).resolve()

WEB_SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS ticket_web_users (
    id BIGSERIAL PRIMARY KEY,
    username TEXT NOT NULL UNIQUE,
    password_hash TEXT NOT NULL,
    display_name TEXT NOT NULL DEFAULT '',
    role TEXT NOT NULL DEFAULT 'admin',
    created_at TIMESTAMP NOT NULL DEFAULT now(),
    updated_at TIMESTAMP NOT NULL DEFAULT now(),
    CHECK (role IN ('admin'))
);

CREATE TABLE IF NOT EXISTS ticket_web_sessions (
    token_hash TEXT PRIMARY KEY,
    user_id BIGINT NOT NULL REFERENCES ticket_web_users(id) ON DELETE CASCADE,
    created_at TIMESTAMP NOT NULL DEFAULT now(),
    expires_at TIMESTAMP NOT NULL
);

CREATE INDEX IF NOT EXISTS ticket_web_sessions_user_idx ON ticket_web_sessions (user_id);
CREATE INDEX IF NOT EXISTS ticket_web_sessions_expires_idx ON ticket_web_sessions (expires_at);
"""

BOT_COMPATIBILITY_SQL = """
ALTER TABLE users ADD COLUMN IF NOT EXISTS user_admin BOOLEAN NOT NULL DEFAULT FALSE;
ALTER TABLE users ADD COLUMN IF NOT EXISTS created_at TIMESTAMP NOT NULL DEFAULT now();
ALTER TABLE users ADD COLUMN IF NOT EXISTS updated_at TIMESTAMP NOT NULL DEFAULT now();

ALTER TABLE filters ADD COLUMN IF NOT EXISTS created_at TIMESTAMP NOT NULL DEFAULT now();
ALTER TABLE tickets ADD COLUMN IF NOT EXISTS created_at TIMESTAMP NOT NULL DEFAULT now();
"""

FILTER_TYPES = {
    "A": {"name": "Area", "table": "aree", "label_column": "AREA"},
    "S": {"name": "Sottoarea", "table": "sottoaree", "label_column": "SOTTOAREA"},
    "C": {"name": "Competenza", "table": "competenze", "label_column": "COMPETENZA"},
    "P": {"name": "Prodotto", "table": "prodotti", "label_column": "PRODOTTO"},
}


class LoginRequest(BaseModel):
    username: str = Field(min_length=1, max_length=120)
    password: str = Field(min_length=1, max_length=256)


class UserPatch(BaseModel):
    status: Literal["P", "A"] | None = None
    user_admin: bool | None = None


class FilterCreate(BaseModel):
    sort: Literal["A", "S", "C", "P"]
    sort_value: str = Field(min_length=1, max_length=80)


class SettingsPatch(BaseModel):
    notifications_enabled: bool | None = None
    quiet_start_hour: int | None = Field(default=None, ge=0, le=23)
    quiet_end_hour: int | None = Field(default=None, ge=0, le=23)
    digest_minutes: int | None = Field(default=None, ge=0, le=1440)
    high_gravity_only: bool | None = None


class FeedbackCreate(BaseModel):
    suggested_ticket_id: int
    query_ticket_id: int | None = None
    feedback: Literal["useful", "not_useful", "favorite", "opened"]
    note: str | None = Field(default=None, max_length=500)


app = FastAPI(
    title="Ticket Web Console",
    version="1.0.0",
    docs_url="/api/docs",
    openapi_url="/api/openapi.json",
)


def _env_bool(name: str, default: bool = False) -> bool:
    value = os.getenv(name)
    if value is None or value == "":
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _pg_connection():
    return psycopg2.connect(**config.POSTGRES_CONFIG)


def _serialize(value: Any) -> Any:
    if isinstance(value, dict):
        return {key: _serialize(item) for key, item in value.items()}
    if isinstance(value, list):
        return [_serialize(item) for item in value]
    if isinstance(value, tuple):
        return [_serialize(item) for item in value]
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Decimal):
        return float(value)
    return value


def _hash_password(password: str, salt: bytes | None = None, iterations: int = 390000) -> str:
    salt = salt or secrets.token_bytes(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
    salt_text = base64.urlsafe_b64encode(salt).decode("ascii")
    digest_text = base64.urlsafe_b64encode(digest).decode("ascii")
    return f"pbkdf2_sha256${iterations}${salt_text}${digest_text}"


def _verify_password(password: str, stored_hash: str) -> bool:
    try:
        scheme, iterations_text, salt_text, digest_text = stored_hash.split("$", 3)
        if scheme != "pbkdf2_sha256":
            return False
        iterations = int(iterations_text)
        salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
        expected = base64.urlsafe_b64decode(digest_text.encode("ascii"))
    except Exception:
        return False
    actual = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
    return hmac.compare_digest(actual, expected)


def _hash_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def _web_actor_id(web_user: dict[str, Any]) -> int:
    return -int(web_user["id"])


def _public_web_user(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": row["id"],
        "username": row["username"],
        "display_name": row.get("display_name") or row["username"],
        "role": row["role"],
    }


def _model_fields_set(model: BaseModel) -> set[str]:
    return getattr(model, "model_fields_set", getattr(model, "__fields_set__", set()))


def ensure_web_schema() -> None:
    ensure_bot_schema()
    ensure_ai_schema()

    connection = _pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(BOT_COMPATIBILITY_SQL)
    cursor.execute(WEB_SCHEMA_SQL)
    cursor.execute("SELECT COUNT(*) AS count FROM ticket_web_users")
    count = int(cursor.fetchone()["count"])
    if count == 0:
        username = os.getenv("TICKET_WEB_ADMIN_USERNAME", "admin")
        password = os.getenv("TICKET_WEB_ADMIN_PASSWORD", "change-me-now")
        display_name = os.getenv("TICKET_WEB_ADMIN_DISPLAY_NAME", "Web administrator")
        cursor.execute(
            """
            INSERT INTO ticket_web_users (username, password_hash, display_name, role)
            VALUES (%s, %s, %s, 'admin')
            """,
            (username, _hash_password(password), display_name),
        )
    cursor.execute("DELETE FROM ticket_web_sessions WHERE expires_at < now()")
    connection.commit()
    cursor.close()
    connection.close()


@app.on_event("startup")
def _startup() -> None:
    ensure_web_schema()


def current_web_user(request: Request) -> dict[str, Any]:
    token = request.cookies.get(COOKIE_NAME)
    if not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")

    web_user = _find_web_user_by_token(token)
    if not web_user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Session expired")
    return web_user


def optional_web_user(request: Request) -> dict[str, Any] | None:
    token = request.cookies.get(COOKIE_NAME)
    if not token:
        return None
    return _find_web_user_by_token(token)


def _find_web_user_by_token(token: str) -> dict[str, Any] | None:
    connection = _pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        """
        SELECT u.id, u.username, u.display_name, u.role
        FROM ticket_web_sessions s
        JOIN ticket_web_users u ON u.id = s.user_id
        WHERE s.token_hash = %s
          AND s.expires_at > now()
        """,
        (_hash_token(token),),
    )
    row = cursor.fetchone()
    cursor.close()
    connection.close()
    return dict(row) if row else None


def _set_session_cookie(response: Response, token: str) -> None:
    response.set_cookie(
        COOKIE_NAME,
        token,
        max_age=SESSION_DAYS * 24 * 60 * 60,
        httponly=True,
        secure=_env_bool("TICKET_WEB_COOKIE_SECURE", False),
        samesite="lax",
    )


def _clear_session_cookie(response: Response) -> None:
    response.delete_cookie(COOKIE_NAME, httponly=True, samesite="lax")


def _assert_select(sql: str) -> None:
    if not sql.lstrip().upper().startswith("SELECT"):
        raise ValueError("MySQL web queries must be read-only SELECT statements")


def _mysql_fetch_all(sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
    _assert_select(sql)
    connection = mysql5.getMySqlConnection()
    try:
        try:
            connection.start_transaction(readonly=True)
        except Exception:
            pass
        cursor = connection.cursor(dictionary=True)
        cursor.execute(sql, params)
        rows = cursor.fetchall()
        cursor.close()
        try:
            connection.rollback()
        except Exception:
            pass
        return _serialize(rows)
    finally:
        connection.close()


def _mysql_fetch_one(sql: str, params: tuple[Any, ...] = ()) -> dict[str, Any] | None:
    rows = _mysql_fetch_all(sql, params)
    return rows[0] if rows else None


def _pg_fetch_all(sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
    connection = _pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(sql, params)
    rows = [dict(row) for row in cursor.fetchall()]
    cursor.close()
    connection.close()
    return _serialize(rows)


def _pg_fetch_one(sql: str, params: tuple[Any, ...] = ()) -> dict[str, Any] | None:
    rows = _pg_fetch_all(sql, params)
    return rows[0] if rows else None


def _ticket_link(ticket_id: int | str) -> str:
    return f"https://tsnew.sanmarcoweb.com/it/ticket/index/index/operation/view/id/{ticket_id}"


def _normalize_filter_value(value: str) -> str:
    value = str(value).strip()
    if value.startswith("_"):
        return "-" + value[1:]
    return value


def _as_int(value: Any) -> int | None:
    try:
        if value is None or value == "":
            return None
        return int(value)
    except (TypeError, ValueError):
        return None


def _filter_label(sort: str, sort_value: str) -> str | None:
    metadata = FILTER_TYPES.get(sort)
    value = _as_int(sort_value)
    if not metadata or value is None:
        return None

    sql = f"""
        SELECT {metadata["label_column"]} AS label
        FROM {metadata["table"]}
        WHERE ID = %s
        LIMIT 1
    """
    row = _mysql_fetch_one(sql, (value,))
    return str(row["label"]) if row and row.get("label") is not None else None


def _decorate_filter(row: dict[str, Any]) -> dict[str, Any]:
    sort = str(row.get("sort") or "")
    value = str(row.get("sort_value") or "")
    row["sort_name"] = FILTER_TYPES.get(sort, {}).get("name", sort)
    try:
        row["label"] = _filter_label(sort, value)
    except Exception:
        row["label"] = None
    return row


def _search_filters_from_params(
    product_id: int | None,
    area_id: int | None,
    subarea_id: int | None,
    competence_id: int | None,
    client_id: int | None,
    recent_months: int | None,
    min_confidence: str | None,
    require_code_match: bool,
) -> dict[str, Any]:
    filters: dict[str, Any] = {}
    if product_id is not None:
        filters["product_ids"] = [product_id]
    if area_id is not None:
        filters["area_ids"] = [area_id]
    if subarea_id is not None:
        filters["subarea_ids"] = [subarea_id]
    if competence_id is not None:
        filters["competence_ids"] = [competence_id]
    if client_id is not None:
        filters["client_ids"] = [client_id]
    if recent_months is not None and recent_months > 0:
        filters["recent_months"] = recent_months
    if min_confidence:
        filters["min_confidence"] = min_confidence
    if require_code_match:
        filters["require_code_match"] = True
    return filters


@app.get("/api/health")
def health() -> dict[str, Any]:
    return {"ok": True, "time": datetime.utcnow().isoformat()}


@app.post("/api/auth/login")
def login(payload: LoginRequest, response: Response) -> dict[str, Any]:
    connection = _pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        """
        SELECT id, username, password_hash, display_name, role
        FROM ticket_web_users
        WHERE username = %s
        """,
        (payload.username,),
    )
    row = cursor.fetchone()
    if not row or not _verify_password(payload.password, row["password_hash"]):
        cursor.close()
        connection.close()
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid username or password")

    token = secrets.token_urlsafe(48)
    expires_at = datetime.utcnow() + timedelta(days=SESSION_DAYS)
    cursor.execute(
        """
        INSERT INTO ticket_web_sessions (token_hash, user_id, expires_at)
        VALUES (%s, %s, %s)
        """,
        (_hash_token(token), row["id"], expires_at),
    )
    connection.commit()
    cursor.close()
    connection.close()
    _set_session_cookie(response, token)
    return {"user": _public_web_user(dict(row))}


@app.get("/api/auth/me")
def me(web_user: dict[str, Any] | None = Depends(optional_web_user)) -> dict[str, Any]:
    return {"user": _public_web_user(web_user) if web_user else None}


@app.post("/api/auth/logout")
def logout(
    request: Request,
    response: Response,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    token = request.cookies.get(COOKIE_NAME)
    if token:
        connection = _pg_connection()
        cursor = connection.cursor()
        cursor.execute("DELETE FROM ticket_web_sessions WHERE token_hash = %s", (_hash_token(token),))
        connection.commit()
        cursor.close()
        connection.close()
    _clear_session_cookie(response)
    return {"ok": True, "user_id": web_user["id"]}


@app.get("/api/dashboard")
def dashboard(web_user: dict[str, Any] = Depends(current_web_user)) -> dict[str, Any]:
    stats = _serialize(ai_retriever.get_ai_stats())
    bot_counts = _pg_fetch_one(
        """
        SELECT
            COUNT(*) AS users,
            COALESCE(SUM(CASE WHEN status = 'P' THEN 1 ELSE 0 END), 0) AS pending_users,
            COALESCE(SUM(CASE WHEN status = 'A' THEN 1 ELSE 0 END), 0) AS approved_users
        FROM users
        """
    ) or {}
    filter_count = _pg_fetch_one("SELECT COUNT(*) AS filters FROM filters") or {"filters": 0}
    settings_count = _pg_fetch_one(
        """
        SELECT
            COUNT(*) AS configured_users,
            COALESCE(SUM(CASE WHEN notifications_enabled THEN 0 ELSE 1 END), 0) AS paused_notifications,
            COALESCE(SUM(CASE WHEN high_gravity_only THEN 1 ELSE 0 END), 0) AS high_gravity_users
        FROM ticket_bot_user_settings
        """
    ) or {}
    recent_feedback = _pg_fetch_all(
        """
        SELECT
            f.created_at,
            f.user_id,
            f.query_ticket_id,
            f.suggested_ticket_id,
            f.feedback,
            i.subject_text,
            i.product,
            i.area
        FROM ticket_ai_feedback f
        LEFT JOIN ticket_ai_index i ON i.ticket_id = f.suggested_ticket_id
        ORDER BY f.created_at DESC
        LIMIT 8
        """
    )

    mysql_status: dict[str, Any] = {"ok": True, "error": None}
    recent_tickets: list[dict[str, Any]] = []
    try:
        recent_tickets = _mysql_fetch_all(
            """
            SELECT
                s.TICKET AS ticket_id,
                cl.INTESTATARIO AS client,
                s.OGGETTO AS subject,
                g.GRAVITA AS gravity,
                c.COMPETENZA AS competence,
                p.PRODOTTO AS product,
                st.STATO AS status,
                s.APERTURA AS opened_at
            FROM segnalazioni s
            LEFT JOIN clienti cl ON cl.ID = s.CLIENTE
            LEFT JOIN gravita g ON g.ID = s.GRAVITA
            LEFT JOIN competenze c ON c.ID = s.COMPETENZA
            LEFT JOIN prodotti p ON p.ID = s.PRODOTTO
            LEFT JOIN stati st ON st.ID = s.STATO
            WHERE s.APERTURA >= %s
            ORDER BY s.APERTURA DESC
            LIMIT 8
            """,
            (date.today() - timedelta(days=14),),
        )
    except Exception as exc:
        mysql_status = {"ok": False, "error": str(exc)}

    return {
        "ai": stats,
        "bot": {**bot_counts, **filter_count, **settings_count},
        "recent_feedback": recent_feedback,
        "recent_tickets": recent_tickets,
        "mysql": mysql_status,
        "viewer": _public_web_user(web_user),
    }


@app.get("/api/tickets")
def list_tickets(
    q: str | None = Query(default=None, max_length=120),
    date_from: date | None = None,
    date_to: date | None = None,
    product_id: int | None = None,
    competence_id: int | None = None,
    area_id: int | None = None,
    subarea_id: int | None = None,
    status_id: int | None = None,
    client_id: int | None = None,
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    where_parts: list[str] = []
    params: list[Any] = []
    protected_range_applied = False

    query = (q or "").strip()
    if query:
        if query.isdigit():
            where_parts.append("s.TICKET = %s")
            params.append(int(query))
        else:
            like = f"%{query[:80]}%"
            where_parts.append("(s.OGGETTO LIKE %s OR cl.INTESTATARIO LIKE %s)")
            params.extend([like, like])
    elif date_from is None and date_to is None:
        date_from = date.today() - timedelta(days=30)
        protected_range_applied = True

    if date_from is not None:
        where_parts.append("s.APERTURA >= %s")
        params.append(date_from)
    if date_to is not None:
        where_parts.append("s.APERTURA < %s")
        params.append(date_to + timedelta(days=1))

    filter_columns = {
        "s.PRODOTTO": product_id,
        "s.COMPETENZA": competence_id,
        "s.AREA": area_id,
        "s.SOTTOAREA": subarea_id,
        "s.STATO": status_id,
        "s.CLIENTE": client_id,
    }
    for column, value in filter_columns.items():
        if value is not None:
            where_parts.append(f"{column} = %s")
            params.append(value)

    where_sql = " AND ".join(where_parts) if where_parts else "TRUE"
    page_size = int(limit) + 1
    rows = _mysql_fetch_all(
        f"""
        SELECT
            s.TICKET AS ticket_id,
            cl.INTESTATARIO AS client,
            s.OGGETTO AS subject,
            g.GRAVITA AS gravity,
            c.COMPETENZA AS competence,
            p.PRODOTTO AS product,
            a.AREA AS area,
            sa.SOTTOAREA AS subarea,
            st.STATO AS status,
            s.APERTURA AS opened_at,
            s.CHIUSURA AS closed_at
        FROM segnalazioni s
        LEFT JOIN clienti cl ON cl.ID = s.CLIENTE
        LEFT JOIN gravita g ON g.ID = s.GRAVITA
        LEFT JOIN competenze c ON c.ID = s.COMPETENZA
        LEFT JOIN prodotti p ON p.ID = s.PRODOTTO
        LEFT JOIN aree a ON a.ID = s.AREA
        LEFT JOIN sottoaree sa ON sa.ID = s.SOTTOAREA
        LEFT JOIN stati st ON st.ID = s.STATO
        WHERE {where_sql}
        ORDER BY s.APERTURA DESC
        LIMIT %s OFFSET %s
        """,
        tuple(params + [page_size, offset]),
    )
    items = rows[:limit]
    for item in items:
        item["source_link"] = _ticket_link(item["ticket_id"])
    return {
        "items": items,
        "has_more": len(rows) > limit,
        "limit": limit,
        "offset": offset,
        "protected_range_applied": protected_range_applied,
        "viewer_id": web_user["id"],
    }


@app.get("/api/tickets/{ticket_id}")
def ticket_detail(
    ticket_id: int,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    ticket = _mysql_fetch_one(
        """
        SELECT
            s.TICKET AS ticket_id,
            s.OGGETTO AS subject,
            s.PROBLEMA AS problem,
            s.SOLUZIONE AS solution,
            s.COMPETENZA AS competence_id,
            c.COMPETENZA AS competence,
            s.PRODOTTO AS product_id,
            p.PRODOTTO AS product,
            s.AREA AS area_id,
            a.AREA AS area,
            s.SOTTOAREA AS subarea_id,
            sa.SOTTOAREA AS subarea,
            s.CLIENTE AS client_id,
            cl.INTESTATARIO AS client,
            s.GRAVITA AS gravity_id,
            g.GRAVITA AS gravity,
            s.STATO AS status_id,
            st.STATO AS status,
            s.APERTURA AS opened_at,
            s.CHIUSURA AS closed_at,
            s.PROGRAMMA AS program,
            s.TIPO_PROGRAMMA AS program_type,
            s.VERSIONE AS version_id,
            s.PATCH AS patch_id,
            s.CLASSIFICAZIONE AS classification_id,
            s.SISTEMA_OPERATIVO AS os_id
        FROM segnalazioni s
        LEFT JOIN competenze c ON c.ID = s.COMPETENZA
        LEFT JOIN prodotti p ON p.ID = s.PRODOTTO
        LEFT JOIN aree a ON a.ID = s.AREA
        LEFT JOIN sottoaree sa ON sa.ID = s.SOTTOAREA
        LEFT JOIN clienti cl ON cl.ID = s.CLIENTE
        LEFT JOIN gravita g ON g.ID = s.GRAVITA
        LEFT JOIN stati st ON st.ID = s.STATO
        WHERE s.TICKET = %s
        LIMIT 1
        """,
        (ticket_id,),
    )
    if not ticket:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ticket not found")
    data = _serialize(ticket)
    data["source_link"] = _ticket_link(ticket_id)
    return {"ticket": data, "viewer_id": web_user["id"]}


@app.get("/api/search/text")
def search_text(
    q: str = Query(min_length=2, max_length=1000),
    product_id: int | None = None,
    area_id: int | None = None,
    subarea_id: int | None = None,
    competence_id: int | None = None,
    client_id: int | None = None,
    recent_months: int | None = Query(default=None, ge=1, le=120),
    min_confidence: Literal["media", "alta"] | None = None,
    require_code_match: bool = False,
    limit: int = Query(default=8, ge=1, le=20),
    offset: int = Query(default=0, ge=0),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    filters = _search_filters_from_params(
        product_id,
        area_id,
        subarea_id,
        competence_id,
        client_id,
        recent_months,
        min_confidence,
        require_code_match,
    )
    rows = ai_retriever.search_similar_text(
        q,
        user_id=_web_actor_id(web_user),
        limit=limit,
        offset=offset,
        filters=filters,
    )
    return {"items": _serialize(rows), "filters": filters, "limit": limit, "offset": offset}


@app.get("/api/search/ticket/{ticket_id}")
def search_ticket(
    ticket_id: int,
    product_id: int | None = None,
    area_id: int | None = None,
    subarea_id: int | None = None,
    competence_id: int | None = None,
    client_id: int | None = None,
    recent_months: int | None = Query(default=None, ge=1, le=120),
    min_confidence: Literal["media", "alta"] | None = None,
    require_code_match: bool = False,
    limit: int = Query(default=8, ge=1, le=20),
    offset: int = Query(default=0, ge=0),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    filters = _search_filters_from_params(
        product_id,
        area_id,
        subarea_id,
        competence_id,
        client_id,
        recent_months,
        min_confidence,
        require_code_match,
    )
    rows = ai_retriever.search_similar_tickets(
        ticket_id,
        user_id=_web_actor_id(web_user),
        limit=limit,
        offset=offset,
        filters=filters,
    )
    return {"items": _serialize(rows), "filters": filters, "limit": limit, "offset": offset}


@app.post("/api/feedback")
def create_feedback(
    payload: FeedbackCreate,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    actor_id = _web_actor_id(web_user)
    if payload.feedback == "favorite":
        inserted = ai_retriever.record_favorite(
            actor_id,
            payload.suggested_ticket_id,
            query_ticket_id=payload.query_ticket_id,
            note=payload.note,
        )
        return {"ok": True, "inserted": inserted}

    if payload.query_ticket_id is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="query_ticket_id is required")

    ai_retriever.record_feedback(
        actor_id,
        payload.query_ticket_id,
        payload.suggested_ticket_id,
        payload.feedback,
        note=payload.note,
    )
    return {"ok": True}


@app.get("/api/users")
def list_users(
    status_filter: Literal["P", "A"] | None = Query(default=None, alias="status"),
    q: str | None = Query(default=None, max_length=120),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    where_parts: list[str] = []
    params: list[Any] = []
    if status_filter:
        where_parts.append("u.status = %s")
        params.append(status_filter)
    if q:
        like = f"%{q.strip()[:80]}%"
        where_parts.append("(CAST(u.tid AS TEXT) LIKE %s OR u.username ILIKE %s OR u.surname ILIKE %s)")
        params.extend([like, like, like])
    where_sql = "WHERE " + " AND ".join(where_parts) if where_parts else ""
    rows = _pg_fetch_all(
        f"""
        SELECT
            u.tid::bigint AS tid,
            u.username,
            u.surname,
            u.status,
            u.user_admin,
            u.created_at,
            u.updated_at,
            COUNT(f.fid)::integer AS filter_count,
            COALESCE(s.notifications_enabled, TRUE) AS notifications_enabled,
            COALESCE(s.high_gravity_only, FALSE) AS high_gravity_only
        FROM users u
        LEFT JOIN filters f ON f.tid = u.tid
        LEFT JOIN ticket_bot_user_settings s ON s.user_id = u.tid
        {where_sql}
        GROUP BY
            u.tid,
            u.username,
            u.surname,
            u.status,
            u.user_admin,
            u.created_at,
            u.updated_at,
            s.notifications_enabled,
            s.high_gravity_only
        ORDER BY u.created_at DESC, u.tid DESC
        LIMIT 300
        """,
        tuple(params),
    )
    return {"items": rows, "viewer_id": web_user["id"]}


@app.patch("/api/users/{tid}")
def update_user(
    tid: int,
    payload: UserPatch,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    fields = _model_fields_set(payload)
    assignments: list[str] = []
    params: list[Any] = []
    if "status" in fields and payload.status is not None:
        assignments.append("status = %s")
        params.append(payload.status)
    if "user_admin" in fields and payload.user_admin is not None:
        assignments.append("user_admin = %s")
        params.append(payload.user_admin)
    if not assignments:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No changes supplied")

    params.append(tid)
    connection = _pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute(
        f"""
        UPDATE users
        SET {", ".join(assignments)},
            updated_at = now()
        WHERE tid = %s
        RETURNING tid::bigint AS tid, username, surname, status, user_admin, created_at, updated_at
        """,
        tuple(params),
    )
    row = cursor.fetchone()
    connection.commit()
    cursor.close()
    connection.close()
    if not row:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return {"user": _serialize(dict(row)), "viewer_id": web_user["id"]}


@app.get("/api/users/{tid}/filters")
def list_user_filters(
    tid: int,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    rows = _pg_fetch_all(
        """
        SELECT fid::bigint AS fid, tid::bigint AS tid, sort, sort_value, created_at
        FROM filters
        WHERE tid = %s
        ORDER BY sort, created_at DESC
        """,
        (tid,),
    )
    return {"items": [_decorate_filter(row) for row in rows], "viewer_id": web_user["id"]}


@app.post("/api/users/{tid}/filters", status_code=status.HTTP_201_CREATED)
def add_user_filter(
    tid: int,
    payload: FilterCreate,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    sort_value = _normalize_filter_value(payload.sort_value)
    if _as_int(sort_value) is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Filter value must be a numeric id")

    connection = _pg_connection()
    cursor = connection.cursor(cursor_factory=DictCursor)
    cursor.execute("SELECT 1 FROM users WHERE tid = %s", (tid,))
    if not cursor.fetchone():
        cursor.close()
        connection.close()
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    cursor.execute(
        """
        INSERT INTO filters (tid, sort, sort_value)
        VALUES (%s, %s, %s)
        ON CONFLICT (tid, sort, sort_value) DO NOTHING
        RETURNING fid::bigint AS fid, tid::bigint AS tid, sort, sort_value, created_at
        """,
        (tid, payload.sort, sort_value),
    )
    row = cursor.fetchone()
    inserted = row is not None
    if not row:
        cursor.execute(
            """
            SELECT fid::bigint AS fid, tid::bigint AS tid, sort, sort_value, created_at
            FROM filters
            WHERE tid = %s AND sort = %s AND sort_value = %s
            """,
            (tid, payload.sort, sort_value),
        )
        row = cursor.fetchone()
    connection.commit()
    cursor.close()
    connection.close()
    return {"filter": _decorate_filter(_serialize(dict(row))), "inserted": inserted, "viewer_id": web_user["id"]}


@app.delete("/api/users/{tid}/filters/{fid}")
def delete_user_filter(
    tid: int,
    fid: int,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    connection = _pg_connection()
    cursor = connection.cursor()
    cursor.execute("DELETE FROM filters WHERE tid = %s AND fid = %s", (tid, fid))
    deleted = cursor.rowcount
    connection.commit()
    cursor.close()
    connection.close()
    if deleted == 0:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Filter not found")
    return {"ok": True, "deleted": deleted, "viewer_id": web_user["id"]}


@app.get("/api/users/{tid}/settings")
def get_user_settings(
    tid: int,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    return {"settings": _serialize(ai_retriever.get_user_bot_settings(tid)), "viewer_id": web_user["id"]}


@app.patch("/api/users/{tid}/settings")
def update_user_settings(
    tid: int,
    payload: SettingsPatch,
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    fields = _model_fields_set(payload)
    updates = {field: getattr(payload, field) for field in fields}
    if not updates:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No changes supplied")
    settings = ai_retriever.update_user_bot_settings(tid, **updates)
    return {"settings": _serialize(settings), "viewer_id": web_user["id"]}


@app.get("/api/taxonomy/{kind}")
def taxonomy(
    kind: Literal["products", "competences", "areas", "subareas", "statuses"],
    q: str | None = Query(default=None, max_length=120),
    area_id: int | None = None,
    limit: int = Query(default=300, ge=1, le=1000),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    table_map = {
        "products": ("prodotti", "PRODOTTO"),
        "competences": ("competenze", "COMPETENZA"),
        "areas": ("aree", "AREA"),
        "subareas": ("sottoaree", "SOTTOAREA"),
        "statuses": ("stati", "STATO"),
    }
    table, label_column = table_map[kind]
    where_parts: list[str] = []
    params: list[Any] = []
    if q:
        where_parts.append(f"{label_column} LIKE %s")
        params.append(f"%{q.strip()[:80]}%")
    if kind == "subareas" and area_id is not None:
        where_parts.append("AREA = %s")
        params.append(area_id)
    where_sql = "WHERE " + " AND ".join(where_parts) if where_parts else ""
    rows = _mysql_fetch_all(
        f"""
        SELECT ID AS id, {label_column} AS label
        FROM {table}
        {where_sql}
        ORDER BY {label_column}
        LIMIT %s
        """,
        tuple(params + [limit]),
    )
    return {"items": rows, "viewer_id": web_user["id"]}


@app.get("/api/reports/no-results")
def no_result_report(
    limit: int = Query(default=20, ge=1, le=100),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    return {"items": _serialize(ai_retriever.get_no_result_searches(limit)), "viewer_id": web_user["id"]}


@app.get("/api/reports/negative-feedback")
def negative_feedback_report(
    limit: int = Query(default=20, ge=1, le=100),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    return {"items": _serialize(ai_retriever.get_negative_feedback(limit)), "viewer_id": web_user["id"]}


@app.get("/api/reports/top-useful")
def top_useful_report(
    limit: int = Query(default=20, ge=1, le=100),
    web_user: dict[str, Any] = Depends(current_web_user),
) -> dict[str, Any]:
    return {"items": _serialize(ai_retriever.get_top_useful(limit)), "viewer_id": web_user["id"]}


def _serve_spa_path(full_path: str) -> FileResponse:
    if full_path.startswith("api/"):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
    candidate = (STATIC_DIR / full_path).resolve()
    if candidate.is_file() and STATIC_DIR in candidate.parents:
        return FileResponse(candidate)
    index = STATIC_DIR / "index.html"
    if index.is_file():
        return FileResponse(index)
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Web UI was not built")


@app.get("/", include_in_schema=False)
def root() -> FileResponse:
    return _serve_spa_path("index.html")


@app.get("/{full_path:path}", include_in_schema=False)
def spa(full_path: str) -> FileResponse:
    return _serve_spa_path(full_path)