aps-agent/server/state/packs.py

# ============================================================
# 数据包（moduleId: state-packs, 可重生 ✅）
# 把「一套完整世界数据」封装为可重放的 JSON 包：
#   - 演示厂 = server/data/packs/demo.json（由脚本从种子生成）
#   - 行业种子 / 客户现场数据都可做成包，data.reset 即「重放数据包」
# 日期重基：包内记录 baseDate（世界业务基准日：演示包=生成日，回放包=案例业务日），加载时把所有 YYYY-MM-DD
# 字面日期整体平移到「今天」，保证任意运行日行为一致（黄金测试稳定）。
# ============================================================
from __future__ import annotations

import json
import os
import re
from datetime import datetime, timedelta
from typing import Any

PACKS_DIR = os.path.join("server", "data", "packs")
_DATE_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})")


def export_pack(world: dict[str, Any], path: str, *, name: str = "", description: str = "") -> None:
    """把世界数据导出为数据包（含日期基准，供加载时重基）。"""
    pack = {
        "packVersion": 1,
        "name": name or os.path.splitext(os.path.basename(path))[0],
        "description": description,
        "baseDate": datetime.now().strftime("%Y-%m-%d"),
        "world": world,
    }
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(pack, f, ensure_ascii=False, indent=1)


def _shift_dates(obj: Any, delta_days: int) -> Any:
    """递归平移所有 YYYY-MM-DD 开头的字符串日期。"""
    if delta_days == 0:
        return obj
    if isinstance(obj, dict):
        return {k: _shift_dates(v, delta_days) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_shift_dates(v, delta_days) for v in obj]
    if isinstance(obj, str):
        m = _DATE_RE.match(obj)
        if m:
            try:
                d = datetime.strptime(m.group(1), "%Y-%m-%d") + timedelta(days=delta_days)
                return d.strftime("%Y-%m-%d") + obj[10:]
            except ValueError:
                return obj
    return obj


def load_pack(path: str) -> dict[str, Any]:
    """读取数据包并把日期重基到今天，返回 world dict。"""
    with open(path, "r", encoding="utf-8") as f:
        pack = json.load(f)
    world = pack.get("world") or {}
    base = pack.get("baseDate")
    if base:
        try:
            delta = (datetime.now().date() - datetime.strptime(base, "%Y-%m-%d").date()).days
            world = _shift_dates(world, delta)
        except ValueError:
            pass
    return world


def list_packs() -> list[dict[str, Any]]:
    """列出可用数据包（名称/描述/路径）。"""
    out: list[dict[str, Any]] = []
    if not os.path.isdir(PACKS_DIR):
        return out
    for fn in sorted(os.listdir(PACKS_DIR)):
        if not fn.endswith(".json"):
            continue
        path = os.path.join(PACKS_DIR, fn)
        try:
            with open(path, "r", encoding="utf-8") as f:
                head = json.load(f)
            out.append({"file": fn, "name": head.get("name") or fn,
                        "description": head.get("description") or "",
                        "aliases": list(head.get("aliases") or []),
                        "baseDate": head.get("baseDate") or "", "path": path})
        except (json.JSONDecodeError, OSError):
            continue
    return out


def _pack_ref_key(value: str) -> str:
    """Normalize a user-facing pack reference without imposing a customer name."""
    value = str(value or "").strip().lower()
    value = re.sub(r"(?:数据包?|data\s*pack|数据)$", "", value).strip()
    return re.sub(r"[^0-9a-z\u4e00-\u9fff]+", "", value)


def resolve_pack_reference(reference: str) -> dict[str, Any] | None:
    """Resolve a natural-language pack reference against the registered pack catalog.

    Matching is deliberately limited to metadata (file/name/aliases) so a chat
    command can never turn an arbitrary path into a data load operation.
    """
    key = _pack_ref_key(reference)
    if not key:
        return None
    packs = list_packs()
    exact: list[dict[str, Any]] = []
    loose: list[dict[str, Any]] = []
    for pack in packs:
        labels = [pack.get("file"), os.path.splitext(str(pack.get("file") or ""))[0],
                  pack.get("name"), *(pack.get("aliases") or [])]
        keys = {_pack_ref_key(label) for label in labels if label}
        if key in keys:
            exact.append(pack)
            continue
        if any(key in label_key or label_key in key for label_key in keys if label_key):
            loose.append(pack)
    matches = exact if exact else loose
    return matches[0] if len(matches) == 1 else None