homework/05_agentic_rag/homework/app/sql_store.py

from __future__ import annotations

import csv
import re
import sqlite3
from datetime import date, timedelta
from pathlib import Path

from app.config import PROJECT_ROOT
from app.schemas import Evidence


SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS incidents (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    occurred_at TEXT NOT NULL,
    incident_id TEXT NOT NULL UNIQUE,
    service_name TEXT NOT NULL,
    error_type TEXT NOT NULL,
    client_os TEXT NOT NULL,
    status TEXT NOT NULL,
    resolution TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_incidents_service_time
ON incidents(service_name, occurred_at);
CREATE INDEX IF NOT EXISTS idx_incidents_resolution
ON incidents(resolution);
CREATE INDEX IF NOT EXISTS idx_incidents_error_type
ON incidents(error_type);
"""

DEFAULT_SOURCE_PATH = PROJECT_ROOT / "data" / "source" / "incidents.csv"
INCIDENT_ID_PATTERN = re.compile(r"^INC\d+$")
SERVICE_NAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{1,63}$")
ERROR_TYPE_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_]{1,63}$")
CLIENT_OS_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._/+-]{1,63}$")
ALLOWED_STATUSES = {"resolved", "open", "in_progress", "pending", "closed", "escalated"}
ACTIVE_STATUSES = ("open", "in_progress", "resolved")


def load_incidents(
    source_path: Path,
) -> list[tuple[date, str, str, str, str, str, str]]:
    """从脱敏 CSV 读取工单记录并转换为 (date, ...) 元组。

    返回字段顺序与 `SCHEMA_SQL` 列顺序一致（除自增 id），
    便于 `initialize_database` 直接复用 executemany。
    """
    if not source_path.exists():
        raise FileNotFoundError(f"SQLite 原始工单不存在：{source_path}")

    # 相对天数在导入时转换，保证"最近 30 天"样例不会随课程日期过期。
    today = date.today()
    rows: list[tuple[date, str, str, str, str, str, str]] = []
    with source_path.open("r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file)
        required_fields = {
            "days_ago",
            "incident_id",
            "service_name",
            "error_type",
            "client_os",
            "status",
            "resolution",
        }
        if not reader.fieldnames or not required_fields <= set(reader.fieldnames):
            raise ValueError("incidents.csv 缺少必需字段")

        for line_number, item in enumerate(reader, start=2):
            try:
                days_ago = int(item["days_ago"])
            except (TypeError, ValueError) as exc:
                raise ValueError(f"第 {line_number} 行 days_ago 非法") from exc
            if not 0 <= days_ago <= 3650:
                raise ValueError(f"第 {line_number} 行 days_ago 超出范围")

            incident_id = item["incident_id"].strip().upper()
            service_name = item["service_name"].strip()
            error_type = item["error_type"].strip()
            client_os = item["client_os"].strip()
            status = item["status"].strip().lower()
            resolution = item["resolution"].strip()

            if not INCIDENT_ID_PATTERN.fullmatch(incident_id):
                raise ValueError(f"第 {line_number} 行 incident_id 非法")
            if not SERVICE_NAME_PATTERN.fullmatch(service_name):
                raise ValueError(f"第 {line_number} 行 service_name 非法")
            if not ERROR_TYPE_PATTERN.fullmatch(error_type):
                raise ValueError(f"第 {line_number} 行 error_type 非法")
            if not CLIENT_OS_PATTERN.fullmatch(client_os):
                raise ValueError(f"第 {line_number} 行 client_os 非法")
            if not resolution:
                raise ValueError(f"第 {line_number} 行 resolution 为空")
            if len(resolution) > 64:
                raise ValueError(f"第 {line_number} 行 resolution 过长")
            if status not in ALLOWED_STATUSES:
                raise ValueError(f"第 {line_number} 行 status 非法")

            rows.append(
                (
                    today - timedelta(days=days_ago),
                    incident_id,
                    service_name,
                    error_type,
                    client_os,
                    status,
                    resolution,
                )
            )

    if not rows:
        raise ValueError("incidents.csv 没有数据")
    return rows


def initialize_database(
    path: Path,
    source_path: Path | None = None,
    reset: bool = False,
) -> int:
    path.parent.mkdir(parents=True, exist_ok=True)
    if reset and path.exists():
        path.unlink()

    # CSV 是可审计真源，SQLite 只负责课程中的精确过滤和聚合。
    rows = load_incidents(source_path or DEFAULT_SOURCE_PATH)
    with sqlite3.connect(path) as connection:
        connection.executescript(SCHEMA_SQL)
        current = connection.execute("SELECT COUNT(*) FROM incidents").fetchone()[0]
        if current == 0:
            connection.executemany(
                """
                INSERT INTO incidents(
                    occurred_at, incident_id, service_name,
                    error_type, client_os, status, resolution
                )
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                [
                    (
                        item[0].isoformat(),
                        item[1],
                        item[2],
                        item[3],
                        item[4],
                        item[5],
                        item[6],
                    )
                    for item in rows
                ],
            )
        return connection.execute("SELECT COUNT(*) FROM incidents").fetchone()[0]


class IncidentRepository:
    """企业 IT 工单聚合查询，统一返回结构化结果与可引用 Evidence。"""

    def __init__(self, path: Path) -> None:
        self.path = path

    def summarize_recent(
        self,
        service_name: str,
        days: int = 30,
        error_type: str = "",
        client_os: str = "",
    ) -> tuple[dict, Evidence]:
        if not self.path.exists():
            raise RuntimeError(
                "业务数据库不存在，请先执行 uv run python scripts/prepare_data.py"
            )
        normalized_service = service_name.strip()
        if not SERVICE_NAME_PATTERN.fullmatch(normalized_service):
            raise ValueError("service_name 只能包含字母数字、下划线和连字符")
        if not 1 <= days <= 365:
            raise ValueError("days 必须在 1 到 365 之间")

        normalized_error = error_type.strip()
        if normalized_error and not ERROR_TYPE_PATTERN.fullmatch(normalized_error):
            raise ValueError("error_type 只能包含字母数字和下划线")
        if len(normalized_error) > 64:
            raise ValueError("error_type 过长")

        normalized_os = client_os.strip()
        if normalized_os and not CLIENT_OS_PATTERN.fullmatch(normalized_os):
            raise ValueError("client_os 非法")
        if len(normalized_os) > 64:
            raise ValueError("client_os 过长")

        start_date = (date.today() - timedelta(days=days)).isoformat()

        # SQL 模板固定并使用占位符；模型不能生成 SQL 或拼接 WHERE 表达式。
        # 单次查询按 resolution 分组返回，每组自带该组的 incident_ids 与 resolved 数；
        # Python 端再做二次聚合得到总数 / 已解决 / 仍需跟进，避免两次往返。
        sql = """
            SELECT resolution,
                   COUNT(*) AS count,
                   SUM(CASE WHEN status = 'resolved' THEN 1 ELSE 0 END) AS resolved_count,
                   GROUP_CONCAT(incident_id) AS incident_ids
            FROM incidents
            WHERE service_name = ?
              AND occurred_at >= ?
              AND status IN (?, ?, ?)
              AND (? = '' OR error_type = ?)
              AND (? = '' OR client_os = ?)
            GROUP BY resolution
            ORDER BY count DESC, resolution ASC
        """
        parameters = (
            normalized_service,
            start_date,
            *ACTIVE_STATUSES,
            normalized_error,
            normalized_error,
            normalized_os,
            normalized_os,
        )
        with sqlite3.connect(self.path) as connection:
            connection.row_factory = sqlite3.Row
            rows = connection.execute(sql, parameters).fetchall()

        incident_ids: list[str] = []
        resolution_counts: dict[str, int] = {}
        total_count = 0
        resolved_count = 0
        for row in rows:
            count = int(row["count"])
            total_count += count
            resolved_count += int(row["resolved_count"] or 0)
            resolution = str(row["resolution"])
            resolution_counts[resolution] = count
            if row["incident_ids"]:
                incident_ids.extend(row["incident_ids"].split(","))

        # 仍需跟进：在已过滤 ACTIVE_STATUSES 之内，排除已解决即为仍需跟进。
        open_count = total_count - resolved_count

        result = {
            "service_name": normalized_service,
            "days": days,
            "error_type": normalized_error,
            "client_os": normalized_os,
            "incident_count": total_count,
            "resolved_count": resolved_count,
            "open_count": open_count,
            "resolution_counts": resolution_counts,
            "incident_ids": incident_ids,
        }

        scope_clauses: list[str] = []
        if normalized_error:
            scope_clauses.append(f"错误类型为“{normalized_error}”")
        if normalized_os:
            scope_clauses.append(f"客户端为“{normalized_os}”")
        scope_clause = (
            f"、{' 且 '.join(scope_clauses)}的" if scope_clauses else ""
        )
        if resolution_counts:
            resolution_summary = "、".join(
                f"{name}({count} 次)" for name, count in resolution_counts.items()
            )
        else:
            resolution_summary = "暂无解决记录"
        evidence = Evidence(
            source_type="sql",
            source="incidents",
            content=(
                f"服务 {normalized_service} 最近 {days} 天{scope_clause}工单共 "
                f"{total_count} 起，其中已解决 {resolved_count} 起、"
                f"仍需跟进 {open_count} 起；"
                f"采取的解决方案包括：{resolution_summary}。"
            ),
            metadata={"query_name": "summarize_recent_incidents", **result},
        )
        return result, evidence