games-development-ai/tier2/gen-worker/scripts/game_artifact_sync.py

"""游戏包对象存储同步 CLI。

本地合成夹具示例（只构建 manifest，绝不上传）：
  PYTHONPATH=tier2/gen-worker python3 tier2/gen-worker/scripts/game_artifact_sync.py \
    --root /tmp/synthetic-game-artifact-fixture \
    --tenant-id fixture --game-id 9000001 --version-id 9000002 --package-type phaser \
    --runtime-manifest game-package.json --source-provider tier2_source_project \
    --source-game-id synthetic-fixture --source-revision-id fixture-r1 \
    --source-manifest-ref tier2-source-project:synthetic-fixture:fixture-r1 --source-hash <sha256>

默认只构建本地 manifest。波次 1 上传还必须显式传 --external-single-writer，表示调用方已在本工具外
持有该 tenant/game/version 的单写锁；上传结果只到 verified_pending，不代表数据库 committed。
"""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path

from worker.game_artifact_store import (
    ArtifactError,
    build_manifest,
    download_manifest,
    fetch_manifest,
    upload_manifest,
)


def _config(bucket: str) -> dict:
    """读取专用 artifact-writer 配置；禁止回落复用现有 MinIO root 凭据。"""
    try:
        from service import infra_config
        config = {
            "endpoint": os.getenv("GAME_ARTIFACT_ENDPOINT", infra_config.get("game_artifact_minio", "endpoint", "")),
            "access_key": os.getenv("GAME_ARTIFACT_ACCESS_KEY", infra_config.get("game_artifact_minio", "access_key", "")),
            "secret_key": os.getenv("GAME_ARTIFACT_SECRET_KEY", infra_config.get("game_artifact_minio", "secret_key", "")),
            "secure": os.getenv("GAME_ARTIFACT_SECURE", str(infra_config.get("game_artifact_minio", "secure", True))).lower() in {"1", "true", "yes"},
            "artifact_bucket": os.getenv("GAME_ARTIFACT_BUCKET", infra_config.get("game_artifact_minio", "artifact_bucket", bucket)),
            "evidence_bucket": os.getenv("GAME_EVIDENCE_BUCKET", infra_config.get("game_artifact_minio", "evidence_bucket", "game-evidence")),
        }
        if not config["endpoint"] or not config["access_key"] or not config["secret_key"]:
            raise ArtifactError("专用 game_artifact_minio 配置未就位，拒绝复用现有 MinIO root 凭据")
        return config
    except Exception as exc:  # noqa: BLE001 - CLI 在缺 tier2 配置时给出稳定错误
        raise ArtifactError(f"读取 MinIO 配置失败:{exc}") from exc


def main() -> int:
    parser = argparse.ArgumentParser(description="构建、上传或下载不可变游戏内容版本")
    parser.add_argument("--root", type=Path, help="具体游戏目录（只作为本次上传输入）")
    parser.add_argument("--tenant-id")
    parser.add_argument("--game-id")
    parser.add_argument("--version-id")
    parser.add_argument("--package-type", choices=["phaser", "littlejs", "canvas"])
    parser.add_argument("--runtime-manifest", help="游戏目录内现有 GamePackage/1 JSON 相对路径")
    parser.add_argument("--source-provider", choices=["tier2_source_project", "game_source_archive"])
    parser.add_argument("--source-game-id")
    parser.add_argument("--source-revision-id")
    parser.add_argument("--source-manifest-ref")
    parser.add_argument("--source-hash")
    parser.add_argument("--bucket", default="game-artifacts")
    parser.add_argument("--manifest-out", type=Path)
    parser.add_argument("--upload", action="store_true", help="显式访问 MinIO 并上传")
    parser.add_argument(
        "--external-single-writer", action="store_true",
        help="确认调用方已持有外部单写锁；波次 1 不提供对象存储条件写",
    )
    parser.add_argument("--download-key", help="受信 artifact-manifest 对象 key；不接受任意 URL")
    parser.add_argument("--expected-manifest-hash", help="数据库登记的 artifact manifest SHA-256")
    parser.add_argument("--target", type=Path, help="下载物化目录")
    args = parser.parse_args()
    try:
        if args.download_key:
            if args.upload or args.external_single_writer or not all((args.target, args.expected_manifest_hash,
                                       args.tenant_id, args.game_id, args.version_id)):
                raise ArtifactError("下载模式必须提供 --target、期望 hash 和 tenant/game/version，且不能带上传参数")
            manifest = fetch_manifest(
                _config(args.bucket), key=args.download_key,
                expected_manifest_hash=args.expected_manifest_hash,
                expected_tenant_id=args.tenant_id,
                expected_game_id=args.game_id,
                expected_version_id=args.version_id,
            )
            result = download_manifest(
                manifest, _config(args.bucket), args.target,
                expected_manifest_hash=args.expected_manifest_hash,
            )
            print(json.dumps({
                "gameId": manifest["gameId"],
                "versionId": manifest["versionId"],
                "manifestHash": manifest["manifestHash"],
                "target": str(result),
            }, ensure_ascii=False))
            return 0
        required = (
            args.root, args.tenant_id, args.game_id, args.version_id, args.package_type,
            args.runtime_manifest, args.source_provider, args.source_game_id,
            args.source_revision_id, args.source_manifest_ref, args.source_hash,
        )
        if not all(required):
            raise ArtifactError("构建/上传模式缺少生产版本、GamePackage 或 sourceRevision 必填参数")
        manifest = build_manifest(args.root, args.tenant_id, args.game_id, args.version_id,
                                  package_type=args.package_type,
                                  runtime_manifest_path=args.runtime_manifest,
                                  source_revision={
                                      "provider": args.source_provider,
                                      "gameId": args.source_game_id,
                                      "revisionId": args.source_revision_id,
                                      "manifestRef": args.source_manifest_ref,
                                      "sourceHash": args.source_hash,
                                  })
        payload = json.dumps(manifest, ensure_ascii=False, indent=2) + "\n"
        if args.manifest_out:
            args.manifest_out.parent.mkdir(parents=True, exist_ok=True)
            args.manifest_out.write_text(payload, encoding="utf-8")
        print(json.dumps({
            "tenantId": manifest["tenantId"],
            "gameId": manifest["gameId"],
            "versionId": manifest["versionId"],
            "objects": manifest["totals"]["objects"],
            "bytes": manifest["totals"]["bytes"],
            "sourceHash": manifest["sourceRevision"]["sourceHash"],
            "assetHash": manifest["assetHash"],
            "bundleHash": manifest["bundleHash"],
            "manifestHash": manifest["manifestHash"],
            "upload": args.upload,
        }, ensure_ascii=False))
        if args.upload:
            if not args.external_single_writer:
                raise ArtifactError("波次 1 上传必须显式确认 --external-single-writer")
            result = upload_manifest(
                args.root, manifest, _config(args.bucket), external_single_writer=True,
            )
            print(json.dumps(result, ensure_ascii=False))
        return 0
    except ArtifactError as exc:
        print(f"GAME-ARTIFACT-ERROR:{exc}")
        return 2


if __name__ == "__main__":
    raise SystemExit(main())