sap-cli-skill/assets/sapcli/manifest.py

"""清单文件 (manifest.json) 管理模块。

负责清单的读取、写入、查询与更新。
"""
from __future__ import annotations

import json
import logging
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

logger = logging.getLogger("sapcli.manifest")

MANIFEST_FILENAME = "manifest.json"
CURRENT_VERSION = 1


@dataclass
class ManifestEntry:
    """单条清单记录。"""

    name: str
    type: str
    file: str
    system_status: str = "not_exists"       # "active" / "inactive" / "not_exists"
    corr_nr: str | None = None
    depends_on: list[str] = field(default_factory=list)
    last_sync: str | None = None            # ISO timestamp
    last_sync_result: str = "pending"       # "success" / "failed" / "pending" / "skipped"

    def to_dict(self) -> dict[str, Any]:
        return {
            "type": self.type,
            "file": self.file,
            "system_status": self.system_status,
            "corr_nr": self.corr_nr,
            "depends_on": self.depends_on,
            "last_sync": self.last_sync,
            "last_sync_result": self.last_sync_result,
        }

    @classmethod
    def from_dict(cls, name: str, data: dict[str, Any]) -> ManifestEntry:
        return cls(
            name=name,
            type=data.get("type", ""),
            file=data.get("file", ""),
            system_status=data.get("system_status", "not_exists"),
            corr_nr=data.get("corr_nr"),
            depends_on=data.get("depends_on", []),
            last_sync=data.get("last_sync"),
            last_sync_result=data.get("last_sync_result", "pending"),
        )


@dataclass
class Manifest:
    """清单容器。"""

    version: int = CURRENT_VERSION
    last_init: str | None = None
    last_refresh: str | None = None
    objects: dict[str, ManifestEntry] = field(default_factory=dict)
    _project_path: str = field(default="", repr=False)

    # ── 持久化 ──

    @classmethod
    def load(cls, project_path: str) -> Manifest:
        """从项目目录下的 manifest.json 加载清单。"""
        filepath = os.path.join(project_path, MANIFEST_FILENAME)
        if not os.path.isfile(filepath):
            raise FileNotFoundError(
                f"清单文件不存在: {filepath}\n  请先执行 init 命令初始化项目"
            )

        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)

        version = data.get("version", 1)
        if version != CURRENT_VERSION:
            logger.warning(
                "清单版本不匹配: 期望 %d, 实际 %d，建议重新 init",
                CURRENT_VERSION,
                version,
            )

        objects: dict[str, ManifestEntry] = {}
        for name, obj_data in data.get("objects", {}).items():
            objects[name] = ManifestEntry.from_dict(name, obj_data)

        manifest = cls(
            version=version,
            last_init=data.get("last_init"),
            last_refresh=data.get("last_refresh"),
            objects=objects,
            _project_path=project_path,
        )
        logger.info("清单已加载: %d 个对象 (%s)", len(objects), filepath)
        return manifest

    def save(self) -> None:
        """将清单写回 manifest.json。"""
        filepath = os.path.join(self._project_path, MANIFEST_FILENAME)
        data = {
            "version": self.version,
            "last_init": self.last_init,
            "last_refresh": self.last_refresh,
            "objects": {
                name: entry.to_dict()
                for name, entry in self.objects.items()
            },
        }
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        logger.info("清单已保存: %s", filepath)

    # ── CRUD ──

    def upsert(self, entry: ManifestEntry) -> None:
        """新增或更新条目。"""
        self.objects[entry.name] = entry

    def remove(self, name: str) -> bool:
        """移除条目。返回是否实际移除了。"""
        if name in self.objects:
            del self.objects[name]
            return True
        return False

    def get(self, name: str) -> ManifestEntry | None:
        """按名称获取条目。"""
        return self.objects.get(name)

    # ── 查询 ──

    def pending_objects(self) -> list[ManifestEntry]:
        """返回需要处理的对象列表（非 active+success 的对象）。"""
        results: list[ManifestEntry] = []
        for entry in self.objects.values():
            if entry.system_status == "active" and entry.last_sync_result == "success":
                continue
            results.append(entry)
        return results

    def active_up_to_date_objects(self) -> list[ManifestEntry]:
        """返回已同步且最新激活的对象列表。"""
        results: list[ManifestEntry] = []
        for entry in self.objects.values():
            if entry.system_status == "active" and entry.last_sync_result == "success":
                results.append(entry)
        return results

    # ── 工具 ──

    def file_path(self, entry: ManifestEntry) -> str:
        """返回条目对应文件的绝对路径。"""
        # entry.file 使用正斜杠，需要适配 os.path.join
        rel = entry.file.replace("/", os.sep)
        return os.path.join(self._project_path, rel)


def init_manifest(project_path: str) -> Manifest:
    """创建一个新的空清单（不保存）。"""
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    return Manifest(
        version=CURRENT_VERSION,
        last_init=now,
        last_refresh=None,
        objects={},
        _project_path=project_path,
    )