273 lines
9.4 KiB
Python
273 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PLUGIN = ROOT / "plugins" / "1c"
|
|
DEFAULT_REPORT = ROOT / "reports" / "1c-plugin-health.json"
|
|
|
|
|
|
def run(command: list[str], *, allow_fail: bool = False) -> dict:
|
|
result = subprocess.run(
|
|
command,
|
|
cwd=ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
status = "ok" if result.returncode == 0 else "failed"
|
|
if allow_fail and result.returncode != 0:
|
|
status = "blocked"
|
|
return {
|
|
"command": command,
|
|
"status": status,
|
|
"returncode": result.returncode,
|
|
"stdout": result.stdout.strip(),
|
|
"stderr": result.stderr.strip(),
|
|
}
|
|
|
|
|
|
def prepare_rag_smoke_index(temp_path: Path) -> list[dict]:
|
|
rag_source = temp_path / "metadata.health.generated.md"
|
|
corpus = temp_path / "rag_corpus.jsonl"
|
|
manifest = temp_path / "rag_manifest.json"
|
|
index = temp_path / "rag_index.json"
|
|
steps = [
|
|
[
|
|
sys.executable,
|
|
"scripts/convert_1c_metadata_to_rag.py",
|
|
"--input",
|
|
"plugins/1c/metadata/examples/metadata.example.json",
|
|
"--output",
|
|
str(rag_source),
|
|
],
|
|
[
|
|
sys.executable,
|
|
"scripts/convert_1c_bsl_modules_to_rag.py",
|
|
"--input",
|
|
"plugins/1c/metadata/examples/bsl-modules.example.json",
|
|
"--output",
|
|
str(temp_path / "bsl.health.generated.md"),
|
|
],
|
|
[
|
|
sys.executable,
|
|
"scripts/validate_1c_rag_sources.py",
|
|
"--source-dir",
|
|
str(temp_path),
|
|
],
|
|
[
|
|
sys.executable,
|
|
"scripts/prepare_1c_rag_corpus.py",
|
|
"--source-dir",
|
|
str(temp_path),
|
|
"--output",
|
|
str(corpus),
|
|
"--manifest",
|
|
str(manifest),
|
|
],
|
|
[
|
|
sys.executable,
|
|
"scripts/build_1c_rag_index.py",
|
|
"--corpus",
|
|
str(corpus),
|
|
"--output",
|
|
str(index),
|
|
],
|
|
]
|
|
return [run(step, allow_fail=True) for step in steps]
|
|
|
|
|
|
def read_yaml(path: Path) -> dict:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
data = yaml.safe_load(handle)
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} must contain a YAML mapping")
|
|
return data
|
|
|
|
|
|
def check_required_files() -> dict:
|
|
required = [
|
|
PLUGIN / "plugin.yaml",
|
|
PLUGIN / "prompts" / "system.md",
|
|
PLUGIN / "prompts" / "rag-answer.md",
|
|
PLUGIN / "rag" / "profiles.yaml",
|
|
PLUGIN / "rag" / "quality-smoke.json",
|
|
PLUGIN / "rag" / "profile-routing-smoke.json",
|
|
PLUGIN / "tools" / "tool-contract.yaml",
|
|
PLUGIN / "connector" / "contracts" / "openapi.yaml",
|
|
PLUGIN / "connector" / "policies" / "read-only-query.yaml",
|
|
PLUGIN / "connector" / "policies" / "change-workflow.yaml",
|
|
PLUGIN / "connector" / "policies" / "config-layer-write-policy.yaml",
|
|
PLUGIN / "connector" / "Dockerfile",
|
|
PLUGIN / "connector" / "docker-compose.yml",
|
|
PLUGIN / "connector" / ".env.example",
|
|
PLUGIN / "connector" / "pyproject.toml",
|
|
PLUGIN / "connector" / "service.yaml",
|
|
PLUGIN / "metadata" / "schema.json",
|
|
PLUGIN / "metadata" / "moxel-schema-registry.json",
|
|
PLUGIN / "metadata" / "examples" / "metadata.example.json",
|
|
PLUGIN / "metadata" / "examples" / "metadata-v2.example.json",
|
|
PLUGIN / "metadata" / "examples" / "bsl-modules.example.json",
|
|
PLUGIN / "schemas" / "metadata-snapshot-v2.schema.json",
|
|
PLUGIN / "schemas" / "bsl-module-snapshot.schema.json",
|
|
PLUGIN / "schemas" / "moxel-schema-registry.schema.json",
|
|
PLUGIN / "training" / "examples" / "instruction.examples.jsonl",
|
|
PLUGIN / "training" / "configs" / "qwen3-coder-30b-a3b-lora.yaml",
|
|
PLUGIN / "evals" / "smoke.yaml",
|
|
]
|
|
missing = [str(path.relative_to(ROOT)) for path in required if not path.exists()]
|
|
return {
|
|
"status": "ok" if not missing else "failed",
|
|
"missing": missing,
|
|
}
|
|
|
|
|
|
def summarize_manifest() -> dict:
|
|
manifest = read_yaml(PLUGIN / "plugin.yaml")
|
|
return {
|
|
"id": manifest.get("id"),
|
|
"version": manifest.get("version"),
|
|
"status": manifest.get("status"),
|
|
"tasks": manifest.get("tasks") or [],
|
|
"entrypoints": sorted((manifest.get("entrypoints") or {}).keys()),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Run 1C plugin health checks.")
|
|
parser.add_argument("--report", type=Path, default=DEFAULT_REPORT)
|
|
parser.add_argument("--print", action="store_true")
|
|
parser.add_argument(
|
|
"--no-report",
|
|
action="store_true",
|
|
help="Do not write a health report file; useful for clean local checks.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
with tempfile.TemporaryDirectory(prefix="llm-1c-health-") as temp_dir:
|
|
temp_index = Path(temp_dir) / "rag_index.json"
|
|
commands = {
|
|
"metadata_example": [
|
|
sys.executable,
|
|
"scripts/validate_1c_metadata_snapshot.py",
|
|
"plugins/1c/metadata/examples/metadata.example.json",
|
|
],
|
|
"metadata_v2_example": [
|
|
sys.executable,
|
|
"scripts/validate_1c_metadata_snapshot.py",
|
|
"plugins/1c/metadata/examples/metadata-v2.example.json",
|
|
],
|
|
"bsl_modules_example": [
|
|
sys.executable,
|
|
"scripts/validate_1c_bsl_modules.py",
|
|
"plugins/1c/metadata/examples/bsl-modules.example.json",
|
|
],
|
|
"readonly_query_allowed": [
|
|
sys.executable,
|
|
"scripts/validate_1c_readonly_query.py",
|
|
"--query",
|
|
"ВЫБРАТЬ Первые 10 Ссылка ИЗ Справочник.Номенклатура",
|
|
],
|
|
"training_examples": [
|
|
sys.executable,
|
|
"scripts/validate_1c_training_data.py",
|
|
"plugins/1c/training/examples/instruction.examples.jsonl",
|
|
],
|
|
"evals": [sys.executable, "scripts/validate_evals.py", "plugins/1c/evals/smoke.yaml"],
|
|
"connector_standalone_check": [sys.executable, "scripts/check_1c_connector_standalone.py"],
|
|
"write_plan_contract": [sys.executable, "scripts/check_1c_write_plan_contract.py"],
|
|
"bsl_symbol_check": [sys.executable, "scripts/check_1c_bsl_symbol_resolver.py"],
|
|
"code_symbol_contract": [sys.executable, "scripts/check_1c_code_symbol_contract.py"],
|
|
"module_origin_contract": [sys.executable, "scripts/check_1c_module_origin_contract.py"],
|
|
"extension_action_contract": [sys.executable, "scripts/check_1c_extension_action_contract.py"],
|
|
"moxel_schema_registry_contract": [
|
|
sys.executable,
|
|
"scripts/check_1c_moxel_schema_registry.py",
|
|
"--registry",
|
|
"plugins/1c/metadata/moxel-schema-registry.json",
|
|
"--output",
|
|
str(Path(temp_dir) / "moxel-schema-registry-check.json"),
|
|
],
|
|
"moxel_status": [
|
|
sys.executable,
|
|
"scripts/status_1c_moxel.py",
|
|
"--output-json",
|
|
str(Path(temp_dir) / "moxel-status.json"),
|
|
"--output-markdown",
|
|
str(Path(temp_dir) / "moxel-status.md"),
|
|
],
|
|
"rag_prompt_guardrails": [
|
|
sys.executable,
|
|
"scripts/check_1c_rag_prompt.py",
|
|
"--index",
|
|
str(temp_index),
|
|
],
|
|
"rag_quality": [
|
|
sys.executable,
|
|
"scripts/check_1c_rag_quality.py",
|
|
"--index",
|
|
str(temp_index),
|
|
],
|
|
"rag_profile_routing": [
|
|
sys.executable,
|
|
"scripts/check_1c_rag_profiles.py",
|
|
],
|
|
"training_preflight": [sys.executable, "scripts/preflight_1c_training.py"],
|
|
}
|
|
|
|
checks = {
|
|
"manifest": summarize_manifest(),
|
|
"required_files": check_required_files(),
|
|
"rag_smoke_prepare": prepare_rag_smoke_index(Path(temp_dir)),
|
|
"commands": {},
|
|
}
|
|
|
|
for name, command in commands.items():
|
|
checks["commands"][name] = run(
|
|
command,
|
|
allow_fail=name in {"training_preflight"},
|
|
)
|
|
|
|
failed = []
|
|
for name, result in checks["commands"].items():
|
|
if result["status"] == "failed":
|
|
failed.append(name)
|
|
if checks["required_files"]["status"] == "failed":
|
|
failed.append("required_files")
|
|
|
|
report = {
|
|
"plugin": "1c",
|
|
"status": "failed" if failed else "ok",
|
|
"failed_checks": failed,
|
|
"checks": checks,
|
|
}
|
|
|
|
if not args.no_report:
|
|
args.report.parent.mkdir(parents=True, exist_ok=True)
|
|
args.report.write_text(
|
|
json.dumps(report, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
if args.print:
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
else:
|
|
if not args.no_report:
|
|
print(f"Wrote 1C plugin health report to {args.report}")
|
|
print(f"Status: {report['status']}")
|
|
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|