# ============================================================ # 数据包(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