aps-agent/server/state/store.py

# ============================================================
# 世界状态存储器（moduleId: state-store, 可重生 ✅）
# M1 形态：单文件 JSON + 原子写（tmp + os.replace），结构与 data.js 对齐
# 后续可整体替换为 PostgreSQL 实现而不影响上层（契约=方法签名）
# ============================================================
from __future__ import annotations                    # 前向类型引用

import json                                            # JSON 序列化
import os                                              # 路径与原子替换
import re                                              # worldKey 清洗
import tempfile                                        # 原子写的临时文件
import threading                                       # 进程内互斥锁
from typing import Any                                 # 类型标注

from server.state.seed import seed_world               # 种子数据工厂


def _safe_scope(value: str, fallback: str) -> str:
    cleaned = re.sub(r"[^\w\-]+", "_", (value or fallback).strip())[:64]
    return cleaned or fallback


def world_path_for(world_key: str | None, tenant_uuid: str | None = None) -> str:
    """Resolve a tenant/project namespaced world path."""
    key = (world_key or "default").strip() or "default"
    key = _safe_scope(key, "default")
    try:
        from server.aps_home import default_world_path
        base = default_world_path()
    except Exception:
        base = os.environ.get("APS_WORLD_PATH", "server/data/world.json")
    tenant = _safe_scope(tenant_uuid or "platform", "platform")
    if tenant == "platform":
        if key in ("default", "", "__personal__"):
            return base
        root = os.path.dirname(base) or "."
        return os.path.join(root, f"world-{key}.json")
    root = os.path.dirname(base) or "."
    return os.path.join(root, "tenants", tenant, "projects", key, "world.json")


class WorldStore:
    """世界状态存储器：加载/保存/重置/发号（进程内可按 worldKey 切换）。"""

    def __init__(self, path: str | None = None, world_key: str = "default", tenant_uuid: str | None = None) -> None:
        """初始化：确定存储路径并加载（无文件则播种）。权力等级 P0（构造只读加载）。"""
        self.world_key = (world_key or "default").strip() or "default"
        self.tenant_uuid = tenant_uuid or "platform"
        if path:
            self.path = path
        else:
            self.path = world_path_for(self.world_key, self.tenant_uuid)

        self._lock = threading.Lock()                  # 互斥锁：保护并发读写（FastAPI 多请求）
        self._counters: dict[str, int] = {}            # ID 发号器（按类型独立计数）
        self._db_fingerprint: str | None = None        # 主数据指纹（脏检查，避免每次 save 都回写 DB）
        self.data: dict[str, Any] = self._load()       # 加载世界状态（无文件时自动播种）
        # 非 default 项目：切到独立 DB 项目，避免与演示厂串数据
        if self.world_key in ("default", "__personal__"):
            self._sync_with_db()
        else:
            try:
                from server.db import sync as db_sync
                db_sync.set_active_project(self.world_key[:32], name=self.world_key)
                if db_sync.db_has_master():
                    db_sync.db_to_world(self.data)
                self._db_fingerprint = db_sync.master_fingerprint(self.data)
            except Exception:
                self._db_fingerprint = None
        self._reset_counters()                         # 依据现有数据校准发号器

    # ---------------- 加载与持久化 ----------------
    def _load(self) -> dict[str, Any]:
        """从磁盘加载世界状态；文件缺失/损坏时重新播种（P0 只读 + 自愈）。"""
        try:
            with open(self.path, "r", encoding="utf-8") as f:      # 尝试读取现有文件
                data = json.load(f)                                # 解析 JSON
        except (FileNotFoundError, json.JSONDecodeError):          # 缺失或损坏
            data = seed_world()                                    # 空世界（或 APS_SEED_DEMO 演示）
            self._write(data)
            return data
        # 旧 world.json 缺表时自愈（OR-05 等增量字段）
        from server.aps_domain.forecast import ensure_forecast_table
        from server.aps_domain.changeover import ensure_changeover_table
        from server.state.seed import purge_demo_residue
        ensure_forecast_table(data)
        ensure_changeover_table(data)
        # 项目世界 / 非演示模式：清掉残留的智能控制器等假主数据
        dirty = purge_demo_residue(data)
        # 已有 SQL/柔性现场数据，但工艺模型页经典表为空 → 投影过去
        if (data.get("flexMaterials") or data.get("flexRoutings")) and not (data.get("materials")):
            try:
                from server.importers.sql_pack import sync_flex_to_classic_master
                sync_flex_to_classic_master(data)
                dirty = True
            except Exception:
                pass
        # WZ*/毛坯供应类型自愈（旧世界未标委外时补齐）
        if data.get("flexRoutings") or data.get("routingSteps"):
            try:
                from server.aps_domain.sourcing import annotate_world_sourcing
                stats = annotate_world_sourcing(data)
                if any(stats.values()):
                    dirty = True
            except Exception:
                pass
        if dirty:
            self._write(data)
        return data

    def _sync_with_db(self) -> None:
        """主数据与 SQLite 对齐：DB 有数据 → 投影覆盖 world 主数据键；
        DB 为空而 world 有主数据 → 一次性迁入（保留现场已导入数据）。
        DB 故障不阻塞启动（world.json 仍可独立运行）。"""
        if os.environ.get("APS_DB_DISABLED"):                       # 显式关闭（隔离测试用）
            return
        try:
            from server.db import sync as db_sync
            if db_sync.db_has_master():                             # DB 是事实源 → 投影
                db_sync.db_to_world(self.data)
                self._write(self.data)                              # 投影结果落回 world.json
            elif self.data.get("materials") or self.data.get("flexOrders"):
                db_sync.world_to_db(self.data)                      # 首次迁移：world → DB
            self._db_fingerprint = db_sync.master_fingerprint(self.data)
        except Exception:                                           # DB 不可用时降级为纯 JSON
            self._db_fingerprint = None

    def _sync_master_to_db(self) -> None:
        """save 收尾：主数据变更时回写 SQLite（指纹脏检查）。"""
        if os.environ.get("APS_DB_DISABLED"):
            return
        try:
            from server.db import sync as db_sync
            fp = db_sync.master_fingerprint(self.data)
            if fp != self._db_fingerprint:
                db_sync.world_to_db(self.data)
                self._db_fingerprint = fp
        except Exception:
            pass                                                    # DB 故障不阻塞业务写盘

    def _write(self, data: dict[str, Any]) -> None:
        """原子写盘：先写临时文件再 os.replace（防止写一半崩溃损坏数据）。"""
        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)  # 确保目录存在
        fd, tmp = tempfile.mkstemp(dir=os.path.dirname(self.path) or ".", suffix=".tmp")  # 同目录临时文件（保证同盘 rename）
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:        # 打开临时文件
                json.dump(data, f, ensure_ascii=False, indent=1)   # 序列化（保留中文可读）
            os.replace(tmp, self.path)                             # 原子替换目标文件
        except BaseException:                                      # 任意失败
            if os.path.exists(tmp):                                # 清理残留临时文件
                os.unlink(tmp)
            raise                                                  # 上抛由调用方处理

    def save(self) -> None:
        """持久化当前世界状态（权力等级 P2 动作的收尾必须调用）。"""
        with self._lock:                                           # 串行化写盘
            self._write(self.data)                                 # 原子落盘
        self._sync_master_to_db()                                  # 主数据变更回写 SQLite

    def reset(self) -> None:
        """重置为种子数据（重放演示数据包；危险操作，调用方须走审计）。"""
        with self._lock:                                           # 串行化
            self.data = seed_world()                               # 重新播种（数据包优先）
            self._counters = {}                                    # 清空发号器
            self._write(self.data)                                 # 落盘
        self._reset_counters()                                     # 重新校准发号器
        self._sync_master_to_db()                                  # 重置结果同步 DB

    def restore(self, world: dict[str, Any]) -> None:
        """整体替换世界状态（Checkpoint 回滚专用，P2 动作放行后调用，§4.3）。"""
        import copy                                                # 局部导入（仅回滚用）
        with self._lock:                                           # 串行化
            self.data = copy.deepcopy(world)                       # 深拷贝隔离快照与运行态
            self._write(self.data)                                 # 落盘
        self._reset_counters()                                     # 按恢复后数据重新校准发号器（防重号）
        self._sync_master_to_db()                                  # 回滚后的主数据同步 DB

    # ---------------- ID 发号（与 data.js resetCounters/nextId 对齐） ----------------
    def _reset_counters(self) -> None:
        """扫描现有数据，把每类计数器校准到当前最大 ID（防重号）。"""
        scan_map = {                                               # 类型 → 数据表 的映射
            "salesOrder": "salesOrders", "productionOrder": "productionOrders",
            "workOrder": "workOrders", "scheduleVersion": "scheduleVersions",
            "conflict": "conflicts", "log": "logs", "audit": "auditEvents",
            "maintenance": "maintenance",              # 维保计划（MD-03 主数据维护可新增）
            "purchaseOrder": "purchaseOrders",         # 采购建议单（MRP 分解产物）
            "outsourceOrder": "outsourceOrders",       # 委外建议单（MRP 分解产物）
            "forecastOrder": "forecastOrders",         # OR-05 预测/长周期订单
            "changeoverMatrix": "changeoverMatrix",    # MD-06 换型矩阵
            "flexScheduleVersion": "flexScheduleVersions",  # 柔性排产版本
            "flexVirtualLine": "flexVirtualLines",     # 虚拟产线实例
            "flexWorkOrder": "flexWorkOrders",         # 柔性工单
            "flexConflict": "flexConflicts",           # 柔性冲突
            "material": "materials",                   # 工艺模型：物料
            "bom": "boms", "bomItem": "bomItems",      # BOM 头 / 明细
            "operation": "operations",                 # 工序库
            "routing": "routings", "routingStep": "routingSteps",  # 工艺路线 / 步骤
            "lineProduct": "lineProducts",             # 产线-产品绑定
        }
        for key, table in scan_map.items():                        # 逐类型扫描
            max_id = 0                                             # 当前最大 ID
            for item in self.data.get(table, []):                  # 遍历该表
                if isinstance(item.get("id"), int):                # 仅数值 ID 参与
                    max_id = max(max_id, item["id"])               # 更新最大值
            self._counters[key] = max_id                           # 校准计数器

    def next_id(self, kind: str) -> int:
        """取下一个 ID（进程内自增，类型隔离）。"""
        with self._lock:                                           # 保护计数器
            self._counters[kind] = self._counters.get(kind, 0) + 1  # 自增
            return self._counters[kind]                            # 返回新号


# ---------------- 按租户/项目隔离的世界缓存 ----------------
_stores: dict[tuple[str, str], WorldStore] = {}
_stores_lock = threading.Lock()


def _current_world_key() -> tuple[str, str]:
    from server.auth.context import get_identity

    identity = get_identity()
    try:
        from server.state.projects import get_project_store
        key = get_project_store().active_world_key()
    except Exception:
        key = "default"
    if key in ("default", "", "__personal__") and identity.user_id:
        key = f"personal-{identity.user_id}"
    return identity.tenant_uuid, _safe_scope(key, "default")


def _scoped_store(tenant_uuid: str, world_key: str) -> WorldStore:
    cache_key = (tenant_uuid, world_key)
    with _stores_lock:
        store = _stores.get(cache_key)
        if store is None:
            store = WorldStore(world_key=world_key, tenant_uuid=tenant_uuid)
            _stores[cache_key] = store
        return store



def peek_scoped_store(tenant_uuid: str, world_key: str) -> WorldStore | None:
    """只读窥探已加载的 scoped store（不新建、不落盘）；未加载返回 None。

    用于审批出卡时做输入快照指纹：生产路径世界已加载则捕获指纹；
    测试/未加载路径返回 None，漂移强制随之跳过（见 round-1 计划假设 1）。
    """
    cache_key = (tenant_uuid, world_key)
    with _stores_lock:
        return _stores.get(cache_key)


def get_store() -> WorldStore:
    """Return the world selected by the authenticated user's workspace."""
    tenant_uuid, key = _current_world_key()
    return _scoped_store(tenant_uuid, key)


def switch_store(world_key: str | None) -> WorldStore:
    """Return a scoped world after the caller has persisted its active project selection."""
    from server.auth.context import get_identity

    identity = get_identity()
    key = (world_key or "default").strip() or "default"
    if key in ("default", "__personal__") and identity.user_id:
        key = f"personal-{identity.user_id}"
    return _scoped_store(identity.tenant_uuid, _safe_scope(key, "default"))