analytics.py

from __future__ import annotations

import json
import uuid
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from config import CLARIFICATION_LOG_PATH, USAGE_LOG_PATH


def record_usage_event(
    question: str,
    answer: str,
    status: str,
    clause_eval: dict[str, Any] | None = None,
) -> None:
    clause_eval = clause_eval or {}
    record = {
        "event_id": str(uuid.uuid4()),
        "created_at": datetime.now(timezone.utc).isoformat(),
        "question": (question or "").strip(),
        "status": status,
        "answer_chars": len(answer or ""),
        "document_ids": clause_eval.get("document_ids", [])[:8],
        "source_ids": clause_eval.get("source_ids", [])[:8],
        "scores": clause_eval.get("scores", [])[:8],
        "top_source": (clause_eval.get("source_ids", []) or [""])[0] if clause_eval.get("source_ids") else "",
        "top_document": (clause_eval.get("document_ids", []) or [""])[0] if clause_eval.get("document_ids") else "",
    }
    _append_jsonl(USAGE_LOG_PATH, record)


def read_usage_events(limit: int | None = None) -> list[dict[str, Any]]:
    records = _read_jsonl(USAGE_LOG_PATH)
    if limit is None:
        return records
    return records[-limit:]


def read_clarification_events(limit: int | None = None) -> list[dict[str, Any]]:
    records = _read_jsonl(CLARIFICATION_LOG_PATH)
    if limit is None:
        return records
    return records[-limit:]


def record_clarification_event(
    event_type: str,
    source_question: str = "",
    selected_key: str = "",
    resolved_question: str = "",
    options: dict[str, str] | None = None,
    raw_message: str = "",
) -> None:
    record = {
        "event_id": str(uuid.uuid4()),
        "created_at": datetime.now(timezone.utc).isoformat(),
        "event_type": event_type,
        "source_question": (source_question or "").strip(),
        "selected_key": str(selected_key or ""),
        "resolved_question": (resolved_question or "").strip(),
        "raw_message": (raw_message or "").strip(),
        "options": options or {},
    }
    _append_jsonl(CLARIFICATION_LOG_PATH, record)


def usage_summary() -> dict[str, Any]:
    records = read_usage_events()
    statuses = Counter(str(record.get("status", "unknown")) for record in records)
    top_sources = Counter(str(record.get("top_source", "")) for record in records if record.get("top_source"))
    top_documents = Counter(str(record.get("top_document", "")) for record in records if record.get("top_document"))
    recent = records[-10:]
    return {
        "total_events": len(records),
        "status_counts": dict(statuses.most_common()),
        "top_sources": top_sources.most_common(8),
        "top_documents": top_documents.most_common(8),
        "recent": recent,
    }


def _append_jsonl(path: Path, record: dict[str, Any]) -> None:
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("a", encoding="utf-8") as file:
            file.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        pass


def _read_jsonl(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        return []
    records = []
    try:
        with path.open("r", encoding="utf-8") as file:
            for line in file:
                line = line.strip()
                if not line:
                    continue
                try:
                    records.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
    except Exception:
        return []
    return records