Agentic_Rag/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 orders (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ordered_at TEXT NOT NULL,
    order_id TEXT NOT NULL UNIQUE,
    user_id TEXT NOT NULL,
    product_name TEXT NOT NULL,
    category TEXT NOT NULL,
    amount REAL NOT NULL,
    status TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_orders_user_time
ON orders(user_id, ordered_at);
CREATE INDEX IF NOT EXISTS idx_orders_product
ON orders(product_name);
"""

DEFAULT_SOURCE_PATH = PROJECT_ROOT / "data" / "source" / "orders.csv"
ORDER_ID_PATTERN = re.compile(r"^O\d{7}$")
USER_ID_PATTERN = re.compile(r"^U\d{4}$")
ALLOWED_STATUSES = {"paid", "shipped", "completed", "cancelled", "refunded"}
ACTIVE_STATUSES = ("paid", "shipped", "completed")


def load_orders(source_path: Path) -> list[tuple[date, str, str, str, str, float, str]]:
    if not source_path.exists():
        raise FileNotFoundError(f"SQLite 原始订单不存在：{source_path}")

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

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

            order_id = item["order_id"].strip().upper()
            user_id = item["user_id"].strip().upper()
            product_name = item["product_name"].strip()
            category = item["category"].strip().lower()
            status = item["status"].strip().lower()
            if not ORDER_ID_PATTERN.fullmatch(order_id):
                raise ValueError(f"第 {line_number} 行 order_id 非法")
            if not USER_ID_PATTERN.fullmatch(user_id):
                raise ValueError(f"第 {line_number} 行 user_id 非法")
            if not product_name or not category:
                raise ValueError(f"第 {line_number} 行商品信息为空")
            if status not in ALLOWED_STATUSES:
                raise ValueError(f"第 {line_number} 行 status 非法")

            rows.append(
                (
                    today - timedelta(days=days_ago),
                    order_id,
                    user_id,
                    product_name,
                    category,
                    amount,
                    status,
                )
            )

    if not rows:
        raise ValueError("orders.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_orders(source_path or DEFAULT_SOURCE_PATH)
    with sqlite3.connect(path) as connection:
        connection.executescript(SCHEMA_SQL)
        current = connection.execute("SELECT COUNT(*) FROM orders").fetchone()[0]
        if current == 0:
            connection.executemany(
                """
                INSERT INTO orders(
                    ordered_at, order_id, user_id, product_name,
                    category, amount, status
                )
                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 orders").fetchone()[0]


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

    def summarize_recent(
        self,
        user_id: str,
        days: int = 30,
        product_keyword: str = "",
    ) -> tuple[dict, Evidence]:
        if not self.path.exists():
            raise RuntimeError(
                "业务数据库不存在，请先执行 uv run python scripts/prepare_data.py"
            )
        normalized_user_id = user_id.strip().upper()
        if not USER_ID_PATTERN.fullmatch(normalized_user_id):
            raise ValueError("user_id 必须是 U 加四位数字")
        if not 1 <= days <= 365:
            raise ValueError("days 必须在 1 到 365 之间")
        if len(product_keyword) > 100:
            raise ValueError("product_keyword 过长")

        start_date = (date.today() - timedelta(days=days)).isoformat()
        normalized_product = product_keyword.strip()
        # SQL 模板固定并使用占位符；模型不能生成 SQL 或拼接 WHERE 表达式。
        sql = """
            SELECT COUNT(*) AS order_count,
                   COALESCE(SUM(amount), 0) AS total_paid,
                   SUM(CASE WHEN amount < 99 THEN 1 ELSE 0 END) AS below_99_count,
                   GROUP_CONCAT(order_id) AS order_ids
            FROM orders
            WHERE user_id = ?
              AND ordered_at >= ?
              AND status IN (?, ?, ?)
              AND (? = '' OR product_name LIKE ?)
        """
        product_pattern = f"%{normalized_product}%" if normalized_product else ""
        parameters = (
            normalized_user_id,
            start_date,
            *ACTIVE_STATUSES,
            normalized_product,
            product_pattern,
        )
        with sqlite3.connect(self.path) as connection:
            connection.row_factory = sqlite3.Row
            row = connection.execute(sql, parameters).fetchone()

        result = {
            "user_id": normalized_user_id,
            "days": days,
            "product_keyword": normalized_product,
            "order_count": int(row["order_count"]),
            "total_paid": round(float(row["total_paid"]), 2),
            "below_99_count": int(row["below_99_count"] or 0),
            "order_ids": row["order_ids"].split(",") if row["order_ids"] else [],
        }
        product_scope = f"、商品包含“{normalized_product}”" if normalized_product else ""
        evidence = Evidence(
            source_type="sql",
            source="orders",
            content=(
                f"用户 {normalized_user_id} 最近 {days} 天{product_scope}的有效订单共 "
                f"{result['order_count']} 单，实付总额 {result['total_paid']:.2f} 元，"
                f"其中低于 99 元 {result['below_99_count']} 单。"
            ),
            metadata={"query_name": "summarize_recent_orders", **result},
        )
        return result, evidence