90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""Operation Catalog 服务。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from typing import Any, Optional
|
|
|
|
from ..config.operation_catalog import OPERATION_CATALOG, OperationCatalogEntry
|
|
|
|
|
|
REQUIRED_OPERATION_CODES = {
|
|
"project.create",
|
|
"project.update",
|
|
"project.archive",
|
|
"project.reopen",
|
|
"task.create",
|
|
"task.update",
|
|
"task.assign",
|
|
"task.prioritize",
|
|
"task.status_transition",
|
|
"task.delete",
|
|
"personnel.role_update",
|
|
"personnel.workload_update",
|
|
"personnel.availability_update",
|
|
"knowledge.doc_upsert",
|
|
"knowledge.chunk_upsert",
|
|
"knowledge.link_upsert",
|
|
"knowledge.reindex",
|
|
"notification.create",
|
|
"notification.ack",
|
|
"notification.close",
|
|
"file.write_whitelist",
|
|
}
|
|
|
|
|
|
class OperationCatalogService:
|
|
def __init__(self) -> None:
|
|
self._catalog: tuple[OperationCatalogEntry, ...] = OPERATION_CATALOG
|
|
|
|
def list_operations(self, *, domain: Optional[str] = None) -> list[dict[str, Any]]:
|
|
rows = self._catalog
|
|
if domain:
|
|
domain_key = str(domain).strip().lower()
|
|
rows = tuple(
|
|
row for row in rows
|
|
if str(row.domain).strip().lower() == domain_key
|
|
)
|
|
return [row.model_dump() for row in rows]
|
|
|
|
def get_operation(self, operation_code: str) -> Optional[dict[str, Any]]:
|
|
target = str(operation_code or "").strip().lower()
|
|
for row in self._catalog:
|
|
if str(row.operation_code).strip().lower() == target:
|
|
return row.model_dump()
|
|
return None
|
|
|
|
def lint_catalog(self) -> dict[str, Any]:
|
|
codes = [row.operation_code for row in self._catalog]
|
|
domains = [str(row.domain) for row in self._catalog]
|
|
code_counter = Counter(codes)
|
|
duplicate_codes = sorted([code for code, count in code_counter.items() if count > 1])
|
|
missing_required = sorted(REQUIRED_OPERATION_CODES - set(codes))
|
|
|
|
invalid_entries: list[str] = []
|
|
for row in self._catalog:
|
|
if not row.operation_code:
|
|
invalid_entries.append("empty_operation_code")
|
|
if not row.input_schema:
|
|
invalid_entries.append(f"{row.operation_code}:missing_input_schema")
|
|
if not row.rbac_policy:
|
|
invalid_entries.append(f"{row.operation_code}:missing_rbac_policy")
|
|
if not row.idempotency_key.fields:
|
|
invalid_entries.append(f"{row.operation_code}:missing_idempotency_fields")
|
|
if not row.audit_fields:
|
|
invalid_entries.append(f"{row.operation_code}:missing_audit_fields")
|
|
|
|
ok = not duplicate_codes and not missing_required and not invalid_entries
|
|
return {
|
|
"ok": ok,
|
|
"operation_count": len(self._catalog),
|
|
"domain_count": len(set(domains)),
|
|
"domains": sorted(set(domains)),
|
|
"duplicate_codes": duplicate_codes,
|
|
"missing_required_codes": missing_required,
|
|
"invalid_entries": invalid_entries,
|
|
}
|
|
|
|
|
|
operation_catalog_service = OperationCatalogService()
|