158 lines
6.4 KiB
Python
158 lines
6.4 KiB
Python
"""Plugin development checks and a separate market materials gate."""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from artemis.plugins.evidence import validate_tools
|
|
from artemis.plugins.market import (
|
|
REPORT_NAME,
|
|
content_fingerprint,
|
|
draft_submission,
|
|
package_directory,
|
|
package_files,
|
|
validate_submission,
|
|
)
|
|
from artemis.plugins.models import PluginManifest, PluginProfile
|
|
from artemis.plugins.session import PluginSession, resolve_server
|
|
from artemis.plugins.store import contained_path
|
|
|
|
|
|
async def check(
|
|
directory: Path,
|
|
live: bool,
|
|
values: dict[str, str],
|
|
*,
|
|
market: bool = False,
|
|
fingerprint: bool = False,
|
|
prepare_market: bool = False,
|
|
) -> dict:
|
|
if prepare_market and not directory.is_dir():
|
|
raise ValueError("请在插件目录生成市场草稿，再打包 ZIP")
|
|
with package_directory(directory) as (root, archive_sha256):
|
|
manifest = PluginManifest.model_validate_json((root / "plugin.json").read_text("utf-8"))
|
|
manifest.check_compatible()
|
|
if manifest.skill:
|
|
skill = contained_path(root, manifest.skill)
|
|
if (
|
|
not skill.is_file()
|
|
or skill.stat().st_size > 256 * 1024
|
|
or not skill.read_text("utf-8").strip()
|
|
):
|
|
raise ValueError("缺少有效的 Skill 或超过大小限制")
|
|
if manifest.schema_version not in {2, 3}:
|
|
raise ValueError("新市场插件必须适配 schema_version=2 或 3")
|
|
if any(name not in manifest.config or manifest.config[name].secret for name in values):
|
|
raise ValueError("--values 只能包含清单中声明的普通配置，不得包含密钥")
|
|
files = package_files(root) if market or fingerprint or prepare_market else None
|
|
if prepare_market:
|
|
assert files is not None
|
|
draft = draft_submission(manifest, files)
|
|
with (root / REPORT_NAME).open("x", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(draft, ensure_ascii=False, indent=2) + "\n")
|
|
return {
|
|
"status": "draft",
|
|
"plugin_id": manifest.id,
|
|
"version": manifest.version,
|
|
"content_sha256": draft["content_sha256"],
|
|
}
|
|
submission = None
|
|
if market:
|
|
assert files is not None
|
|
submission = validate_submission(root, manifest, files)
|
|
handshake = bool((live or market) and manifest.server)
|
|
if handshake:
|
|
server = resolve_server(manifest, PluginProfile(values=values), root)
|
|
server.env["PYTHONDONTWRITEBYTECODE"] = "1"
|
|
session = await PluginSession(server, root).start()
|
|
try:
|
|
validate_tools(manifest, session.tools)
|
|
finally:
|
|
await session.aclose()
|
|
if files is not None and package_files(root) != files:
|
|
raise ValueError("MCP 握手期间包内容发生变化，检查记录失效")
|
|
if market:
|
|
assert files is not None
|
|
if validate_submission(root, manifest, files) != submission:
|
|
raise ValueError("MCP 握手期间市场记录发生变化，检查记录失效")
|
|
result = {
|
|
"status": "passed",
|
|
"plugin_id": manifest.id,
|
|
"version": manifest.version,
|
|
"scope": "market_materials" if market else "development",
|
|
"mcp_handshake": handshake,
|
|
}
|
|
if files is not None:
|
|
result["content_sha256"] = content_fingerprint(files)
|
|
if archive_sha256 is not None:
|
|
result["archive_sha256"] = archive_sha256
|
|
return result
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("directory", type=Path, help="插件目录或 ZIP；不会安装到现有插件库")
|
|
parser.add_argument(
|
|
"--live", action="store_true", help="仅启动/关闭 MCP 并核对声明，不调用工具"
|
|
)
|
|
parser.add_argument("--values", default="{}", help="普通配置 JSON；不得包含密钥")
|
|
modes = parser.add_mutually_exclusive_group()
|
|
modes.add_argument(
|
|
"--market", action="store_true", help="核对市场分发、内容绑定和验收材料，含 MCP 握手"
|
|
)
|
|
modes.add_argument(
|
|
"--prepare-market", action="store_true", help="创建未验证的 MARKET.json 草稿，不覆盖旧文件"
|
|
)
|
|
modes.add_argument(
|
|
"--fingerprint", action="store_true", help="输出包内容摘要；不宣称市场验收通过"
|
|
)
|
|
parser.add_argument("--json-output", type=Path, help="将检查结果写到包外 JSON 文件")
|
|
args = parser.parse_args()
|
|
result = {}
|
|
can_write = True
|
|
try:
|
|
if args.json_output and (
|
|
args.json_output.resolve() == args.directory.resolve()
|
|
or (
|
|
args.directory.is_dir()
|
|
and args.json_output.resolve().is_relative_to(args.directory.resolve())
|
|
)
|
|
):
|
|
can_write = False
|
|
raise ValueError("检查输出必须在插件包之外，不能覆盖输入包")
|
|
values = json.loads(args.values)
|
|
if not isinstance(values, dict) or any(not isinstance(v, str) for v in values.values()):
|
|
raise ValueError("--values 必须是字符串值的 JSON 对象")
|
|
result = asyncio.run(
|
|
check(
|
|
args.directory.resolve(),
|
|
args.live,
|
|
values,
|
|
market=args.market,
|
|
fingerprint=args.fingerprint,
|
|
prepare_market=args.prepare_market,
|
|
)
|
|
)
|
|
except (ValueError, OSError, RuntimeError) as exc:
|
|
message = (
|
|
"输入格式或清单/市场材料字段不合法"
|
|
if isinstance(exc, (json.JSONDecodeError, ValidationError))
|
|
else str(exc)
|
|
)
|
|
result = {"status": "failed", "error": message}
|
|
if args.json_output and can_write:
|
|
args.json_output.write_text(
|
|
json.dumps(result, ensure_ascii=False, indent=2) + "\n", "utf-8"
|
|
)
|
|
print(json.dumps(result, ensure_ascii=False))
|
|
print("检查范围不包括源码正确性与真实业务通过；市场材料仍需来源审查和真实任务验收。")
|
|
if result["status"] == "failed":
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|