120 lines
4.6 KiB
Python
120 lines
4.6 KiB
Python
# ============================================================
|
||
# 数据包（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
|