3a27e7ec2b
- approve only appListSecurityEvents and appGetSecurityEvent at operation level - keep appAcknowledgeSecurityEvent needs_verification - add HTTP+DB completed approval IT and update coverage gates/docs
737 lines
30 KiB
Python
Executable File
737 lines
30 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Generate the P1R API coverage matrix.
|
||
|
||
The script is intentionally conservative: if it cannot prove an operation has a
|
||
dedicated business implementation, it marks the operation as incomplete or
|
||
blocked. This prevents catch-all contract handlers from being counted as done.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import hashlib
|
||
import re
|
||
from dataclasses import asdict, dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
try:
|
||
import yaml
|
||
except ImportError as exc:
|
||
raise SystemExit("PyYAML is required. Install it with: python3 -m pip install PyYAML") from exc
|
||
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||
CONTRACT_ROOT = REPO_ROOT / "docs" / "api-contracts"
|
||
MUSE_CLOUD_ROOT = REPO_ROOT / "muse-cloud"
|
||
REPORT_ROOT = REPO_ROOT / "docs" / "superpowers" / "reports"
|
||
JSON_REPORT = REPORT_ROOT / "p1r-api-coverage.json"
|
||
MARKDOWN_REPORT = REPORT_ROOT / "p1r-api-coverage.md"
|
||
HTTP_METHODS = {"get", "post", "put", "patch", "delete"}
|
||
NON_REAL_IMPLEMENTATION_STATUSES = {"catch_all", "generic_persistence", "sse_placeholder", "missing"}
|
||
NON_REAL_COMPLETION_STATUSES = {"incomplete", "blocked"}
|
||
JAVA_HTTP_METHODS = {method.upper() for method in HTTP_METHODS}
|
||
MAPPING_METHODS = {
|
||
"Get": "GET",
|
||
"Post": "POST",
|
||
"Put": "PUT",
|
||
"Patch": "PATCH",
|
||
"Delete": "DELETE",
|
||
}
|
||
SSE_PLACEHOLDER_READY_MARKERS = {
|
||
"muse ai task stream ready",
|
||
"muse event stream ready",
|
||
}
|
||
APPROVED_COMPLETED_DOMAINS = {"ai", "knowledge"}
|
||
APPROVED_COMPLETED_OPERATIONS = {
|
||
"events:streamEvents",
|
||
# P1R Meta completed approval 使用 operation-level 清单，避免整域 Meta 被误升 completed。
|
||
"meta:listMetaSchemas",
|
||
"meta:getMetaSchema",
|
||
"meta:getMetaSchemaVersion",
|
||
"meta:saveMetaSchemaDraft",
|
||
"meta:validateMetaSchemaDraft",
|
||
"meta:previewMetaSchemaDraftImpact",
|
||
"meta:publishMetaSchemaDraft",
|
||
"meta:activateMetaSchemaVersion",
|
||
"meta:rollbackMetaSchemaVersion",
|
||
"meta:deprecateMetaSchemaVersion",
|
||
"meta:setMetaSchemaGrayRules",
|
||
"meta:listProtectionNodes",
|
||
"meta:getProtectionNode",
|
||
"meta:listFunctionChains",
|
||
"meta:previewFunctionChainImpact",
|
||
"meta:activateFunctionChainVersion",
|
||
# P1R Account completed approval 第一批仍使用 operation-level 清单，避免整域 Account 被误升 completed。
|
||
"account:getCurrentUser",
|
||
"account:getProfile",
|
||
"account:updateProfile",
|
||
"account:adminListAccountUsers",
|
||
"account:adminGetUserEntitlements",
|
||
"account:getAppEntitlements",
|
||
"account:adminGetBalanceSnapshots",
|
||
"account:getAppBalanceSnapshots",
|
||
"account:adminCreateQuotaAdjustment",
|
||
"account:adminListQuotaAdjustments",
|
||
# P1R Account Security Events read completed approval 只推进两个只读接口。
|
||
"account:appListSecurityEvents",
|
||
"account:appGetSecurityEvent",
|
||
# P1R Market completed approval 第一批仍使用 operation-level 清单，避免整域 Market 被误升 completed。
|
||
"market:getMarketplaceAsset",
|
||
"market:listMarketplaceCategories",
|
||
"market:favoriteAsset",
|
||
"market:unfavoriteAsset",
|
||
# P1R Content completed approval 仍使用 operation-level 清单，避免整域 Content 被误升 completed。
|
||
"content:listWorks",
|
||
"content:getWork",
|
||
"content:listChapters",
|
||
"content:getChapter",
|
||
"content:listBlocks",
|
||
"content:getBlock",
|
||
"content:saveBlock",
|
||
"content:getBlockSourceAttribution",
|
||
"content:getPlanning",
|
||
"content:savePlanningItem",
|
||
"content:adminListWorks",
|
||
"content:adminGetWork",
|
||
"content:adminListChapters",
|
||
"content:adminRiskAction",
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Operation:
|
||
domain: str
|
||
side: str
|
||
method: str
|
||
path: str
|
||
operationId: str
|
||
isWrite: bool
|
||
requiresCommandId: bool
|
||
externalDependencies: list[str]
|
||
targetStage: str
|
||
implementationStatus: str
|
||
completionStatus: str
|
||
controllerFiles: list[str]
|
||
serviceFiles: list[str]
|
||
notes: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RouteMapping:
|
||
methods: tuple[str, ...]
|
||
routes: tuple[str, ...]
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Generate P1R API coverage matrix")
|
||
parser.add_argument("--check", action="store_true", help="validate generated report after writing it")
|
||
args = parser.parse_args()
|
||
|
||
operations = collect_operations()
|
||
REPORT_ROOT.mkdir(parents=True, exist_ok=True)
|
||
report = build_report(operations)
|
||
JSON_REPORT.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
MARKDOWN_REPORT.write_text(render_markdown(report), encoding="utf-8")
|
||
|
||
if args.check:
|
||
validate_report(report)
|
||
print(f"Generated {JSON_REPORT.relative_to(REPO_ROOT)}")
|
||
print(f"Generated {MARKDOWN_REPORT.relative_to(REPO_ROOT)}")
|
||
return 0
|
||
|
||
|
||
def collect_operations() -> list[Operation]:
|
||
source_index = build_source_index()
|
||
operations: list[Operation] = []
|
||
for contract_path in sorted(CONTRACT_ROOT.glob("*/openapi.yaml")):
|
||
domain = contract_path.parent.name
|
||
with contract_path.open("r", encoding="utf-8") as handle:
|
||
document = yaml.safe_load(handle)
|
||
for path, path_item in sorted((document.get("paths") or {}).items()):
|
||
if not isinstance(path_item, dict):
|
||
continue
|
||
for method, operation_doc in sorted(path_item.items()):
|
||
if method.lower() not in HTTP_METHODS or not isinstance(operation_doc, dict):
|
||
continue
|
||
operation_id = operation_doc.get("operationId")
|
||
if not operation_id:
|
||
raise SystemExit(f"Missing operationId in {contract_path}:{path}:{method}")
|
||
full_path = normalize_path(path)
|
||
side = detect_side(full_path)
|
||
is_write = method.upper() not in {"GET", "HEAD", "OPTIONS"}
|
||
requires_command_id = operation_requires_command_id(operation_doc, document)
|
||
implementation = classify_implementation(full_path, side, method.upper(), operation_id, domain, source_index)
|
||
operations.append(
|
||
Operation(
|
||
domain=domain,
|
||
side=side,
|
||
method=method.upper(),
|
||
path=full_path,
|
||
operationId=operation_id,
|
||
isWrite=is_write,
|
||
requiresCommandId=requires_command_id,
|
||
externalDependencies=detect_external_dependencies(domain, full_path, operation_id),
|
||
targetStage=target_stage(domain),
|
||
implementationStatus=implementation["implementationStatus"],
|
||
completionStatus=implementation["completionStatus"],
|
||
controllerFiles=implementation["controllerFiles"],
|
||
serviceFiles=implementation["serviceFiles"],
|
||
notes=implementation["notes"],
|
||
)
|
||
)
|
||
return operations
|
||
|
||
|
||
def build_source_index() -> dict[str, Any]:
|
||
java_files = sorted(MUSE_CLOUD_ROOT.glob("**/src/main/java/**/*.java"))
|
||
contract_support_files: list[str] = []
|
||
generic_persistence_files: list[str] = []
|
||
sse_placeholder_mappings: dict[str, list[RouteMapping]] = {}
|
||
dedicated_controllers: list[str] = []
|
||
service_files: list[str] = []
|
||
all_text: dict[str, str] = {}
|
||
|
||
for java_file in java_files:
|
||
text = java_file.read_text(encoding="utf-8", errors="ignore")
|
||
rel = str(java_file.relative_to(REPO_ROOT))
|
||
all_text[rel] = text
|
||
if "MuseApiContractSupport.handle" in text:
|
||
contract_support_files.append(rel)
|
||
if "MuseContractPersistenceService" in text or "contractPersistenceService.handle" in text:
|
||
generic_persistence_files.append(rel)
|
||
placeholder_mappings = sse_placeholder_route_mappings(text)
|
||
if placeholder_mappings:
|
||
sse_placeholder_mappings[rel] = placeholder_mappings
|
||
if "@RestController" in text and "MuseApiContractSupport.handle" not in text and "contractPersistenceService.handle" not in text:
|
||
dedicated_controllers.append(rel)
|
||
if "/application/" in rel or "/service/" in rel:
|
||
service_files.append(rel)
|
||
|
||
return {
|
||
"allText": all_text,
|
||
"contractSupportFiles": contract_support_files,
|
||
"genericPersistenceFiles": generic_persistence_files,
|
||
"ssePlaceholderMappings": sse_placeholder_mappings,
|
||
"dedicatedControllers": dedicated_controllers,
|
||
"serviceFiles": service_files,
|
||
}
|
||
|
||
|
||
def classify_implementation(path: str, side: str, method: str, operation_id: str, domain: str, source_index: dict[str, Any]) -> dict[str, Any]:
|
||
matching_dedicated = files_containing_route_or_operation(
|
||
source_index["dedicatedControllers"],
|
||
source_index["allText"],
|
||
path,
|
||
side,
|
||
method,
|
||
operation_id,
|
||
allow_operation_id_fallback=False,
|
||
)
|
||
matching_services = files_matching_domain_services(source_index["serviceFiles"], domain)
|
||
matching_catch_all = files_containing_route_or_operation(
|
||
source_index["contractSupportFiles"],
|
||
source_index["allText"],
|
||
path,
|
||
side,
|
||
method,
|
||
operation_id,
|
||
allow_operation_id_fallback=False,
|
||
)
|
||
matching_generic = files_containing_route_or_operation(
|
||
source_index["genericPersistenceFiles"],
|
||
source_index["allText"],
|
||
path,
|
||
side,
|
||
method,
|
||
operation_id,
|
||
allow_operation_id_fallback=True,
|
||
)
|
||
matching_sse = files_containing_sse_placeholder_route(source_index["ssePlaceholderMappings"], path, side, method)
|
||
|
||
if matching_sse:
|
||
return status(
|
||
"sse_placeholder",
|
||
"incomplete",
|
||
matching_sse,
|
||
matching_services,
|
||
"SSE endpoint returns placeholder events and needs P1R real task stream implementation.",
|
||
)
|
||
if matching_dedicated:
|
||
dedicated_completion, dedicated_notes = dedicated_status_for(domain, operation_id)
|
||
return status(
|
||
"dedicated",
|
||
dedicated_completion,
|
||
matching_dedicated,
|
||
matching_services,
|
||
dedicated_notes,
|
||
)
|
||
if matching_catch_all:
|
||
return status(
|
||
"catch_all",
|
||
"incomplete",
|
||
matching_catch_all,
|
||
matching_services,
|
||
"Handled by MuseApiContractSupport catch-all, which is not P1R completion.",
|
||
)
|
||
if matching_generic:
|
||
return status(
|
||
"generic_persistence",
|
||
"incomplete",
|
||
matching_generic,
|
||
matching_services,
|
||
"Handled by MuseContractPersistenceService generic persistence, which is not P1R completion.",
|
||
)
|
||
return status("missing", "incomplete", [], matching_services, "No backend route was identified by the conservative scanner.")
|
||
|
||
|
||
def status(implementation: str, completion: str, controllers: list[str], services: list[str], notes: str) -> dict[str, Any]:
|
||
return {
|
||
"implementationStatus": implementation,
|
||
"completionStatus": completion,
|
||
"controllerFiles": controllers,
|
||
"serviceFiles": services,
|
||
"notes": notes,
|
||
}
|
||
|
||
|
||
def operation_key(domain: str, operation_id: str) -> str:
|
||
return f"{domain}:{operation_id}"
|
||
|
||
|
||
def dedicated_status_for(domain: str, operation_id: str) -> tuple[str, str]:
|
||
# AI/Knowledge 是历史 domain-level 用户批准；Events/Meta 使用 operation-level approval，
|
||
# 避免未来新增 operation 自动继承 completed。
|
||
if domain in APPROVED_COMPLETED_DOMAINS or operation_key(domain, operation_id) in APPROVED_COMPLETED_OPERATIONS:
|
||
return (
|
||
"completed",
|
||
"Dedicated entry exists; user-approved P1R acceptance evidence allows this operation to advance to completed coverage.",
|
||
)
|
||
return (
|
||
"needs_verification",
|
||
"Dedicated entry exists; later P1R stage must verify DTO, state machine, persistence, and real external closure.",
|
||
)
|
||
|
||
|
||
def files_containing_route_or_operation(
|
||
files: list[str],
|
||
text_by_file: dict[str, str],
|
||
path: str,
|
||
side: str,
|
||
method: str,
|
||
operation_id: str,
|
||
allow_operation_id_fallback: bool,
|
||
) -> list[str]:
|
||
route_variants = route_variants_for(path)
|
||
matches: list[str] = []
|
||
for file_path in files:
|
||
if not file_matches_side(file_path, side):
|
||
continue
|
||
text = text_by_file[file_path]
|
||
route_mappings = extract_route_mappings(text)
|
||
route_matched = any(route_mapping_matches(route_mapping, method, route_variants) for route_mapping in route_mappings)
|
||
operation_id_matched = allow_operation_id_fallback and operation_id_occurs(text, operation_id)
|
||
if route_matched or operation_id_matched:
|
||
matches.append(file_path)
|
||
return matches
|
||
|
||
|
||
def files_containing_sse_placeholder_route(route_by_file: dict[str, list[RouteMapping]], path: str, side: str, method: str) -> list[str]:
|
||
route_variants = route_variants_for(path)
|
||
matches: list[str] = []
|
||
for file_path, route_mappings in route_by_file.items():
|
||
if not file_matches_side(file_path, side):
|
||
continue
|
||
if any(route_mapping_matches(route_mapping, method, route_variants) for route_mapping in route_mappings):
|
||
matches.append(file_path)
|
||
return matches
|
||
|
||
|
||
def file_matches_side(file_path: str, side: str) -> bool:
|
||
if "/controller/admin/" in file_path:
|
||
return side == "admin"
|
||
if "/controller/app/" in file_path:
|
||
return side == "app"
|
||
return True
|
||
|
||
|
||
def route_variants_for(path: str) -> list[str]:
|
||
variants = {
|
||
path,
|
||
path.removeprefix("/app-api"),
|
||
path.removeprefix("/admin-api"),
|
||
}
|
||
for variant in list(variants):
|
||
if variant.startswith("/muse/"):
|
||
variants.add(variant.removeprefix("/muse"))
|
||
normalized_variants = {variant if variant.startswith("/") else "/" + variant for variant in variants}
|
||
return sorted(normalized_variants, key=len, reverse=True)
|
||
|
||
|
||
def extract_route_literals(text: str) -> list[str]:
|
||
return re.findall(r'"(/[^"]*)"', text)
|
||
|
||
|
||
def extract_route_mappings(text: str) -> list[RouteMapping]:
|
||
class_routes = extract_class_request_mapping_routes(text)
|
||
mappings: list[RouteMapping] = []
|
||
for match in re.finditer(r"@(Get|Post|Put|Patch|Delete|Request)Mapping(?:\s*\((.*?)\))?", text, flags=re.DOTALL):
|
||
if is_before_primary_type_declaration(text, match.start()):
|
||
continue
|
||
method_mappings = route_mappings_from_annotation(match.group(1), match.group(2) or "", allow_empty_route=True)
|
||
mappings.extend(combine_class_and_method_routes(class_routes, method_mappings))
|
||
return mappings
|
||
|
||
|
||
def extract_class_request_mapping_routes(text: str) -> tuple[str, ...]:
|
||
primary_type_index = primary_type_declaration_index(text)
|
||
if primary_type_index == -1:
|
||
return ("",)
|
||
|
||
routes: set[str] = set()
|
||
for match in re.finditer(r"@RequestMapping(?:\s*\((.*?)\))?", text[:primary_type_index], flags=re.DOTALL):
|
||
routes.update(extract_route_literals(match.group(1) or ""))
|
||
return tuple(sorted(routes)) if routes else ("",)
|
||
|
||
|
||
def is_before_primary_type_declaration(text: str, index: int) -> bool:
|
||
primary_type_index = primary_type_declaration_index(text)
|
||
return primary_type_index != -1 and index < primary_type_index
|
||
|
||
|
||
def primary_type_declaration_index(text: str) -> int:
|
||
match = re.search(r"\b(?:class|interface|record)\s+[A-Za-z_][A-Za-z0-9_]*", text)
|
||
return match.start() if match else -1
|
||
|
||
|
||
def route_mappings_from_annotation(annotation_kind: str, annotation_body: str, allow_empty_route: bool = False) -> list[RouteMapping]:
|
||
routes = tuple(sorted(set(extract_route_literals(annotation_body))))
|
||
if not routes:
|
||
if not allow_empty_route:
|
||
return []
|
||
routes = ("",)
|
||
|
||
if annotation_kind in MAPPING_METHODS:
|
||
return [RouteMapping(methods=(MAPPING_METHODS[annotation_kind],), routes=routes)]
|
||
|
||
request_methods = tuple(sorted(set(re.findall(r"RequestMethod\.([A-Z]+)", annotation_body)) & JAVA_HTTP_METHODS))
|
||
if not request_methods:
|
||
return []
|
||
return [RouteMapping(methods=request_methods, routes=routes)]
|
||
|
||
|
||
def combine_class_and_method_routes(class_routes: tuple[str, ...], method_mappings: list[RouteMapping]) -> list[RouteMapping]:
|
||
combined_mappings: list[RouteMapping] = []
|
||
for method_mapping in method_mappings:
|
||
routes = tuple(
|
||
sorted(
|
||
{
|
||
join_route_paths(class_route, method_route)
|
||
for class_route in class_routes
|
||
for method_route in method_mapping.routes
|
||
}
|
||
)
|
||
)
|
||
combined_mappings.append(RouteMapping(methods=method_mapping.methods, routes=routes))
|
||
return combined_mappings
|
||
|
||
|
||
def join_route_paths(class_route: str, method_route: str) -> str:
|
||
# Spring MVC 会把类级 base path 与方法级相对 path 合并；空方法路径表示使用类级路径本身。
|
||
parts = [part.strip("/") for part in (class_route, method_route) if part and part.strip("/")]
|
||
return "/" + "/".join(parts) if parts else "/"
|
||
|
||
|
||
def route_mapping_matches(route_mapping: RouteMapping, method: str, route_variants: list[str]) -> bool:
|
||
if method not in route_mapping.methods:
|
||
return False
|
||
return any(
|
||
route_literal_matches(route_literal, route_variant)
|
||
for route_literal in route_mapping.routes
|
||
for route_variant in route_variants
|
||
)
|
||
|
||
|
||
def route_literal_matches(route_literal: str, route_variant: str) -> bool:
|
||
literal = normalize_route_template(route_literal)
|
||
variant = normalize_route_template(route_variant)
|
||
if literal == variant:
|
||
return True
|
||
if literal.endswith("/**"):
|
||
base = literal.removesuffix("/**")
|
||
return variant == base or variant.startswith(base + "/")
|
||
return False
|
||
|
||
|
||
def normalize_route_template(route: str) -> str:
|
||
# Java Controller 与 OpenAPI 都用 {name} 表示路径变量；变量名不同也应视为同一路由形状。
|
||
return re.sub(r"\{[^}/]+\}", "{}", route)
|
||
|
||
|
||
def operation_id_occurs(text: str, operation_id: str) -> bool:
|
||
# 只允许标识符边界匹配，避免 createWork 命中 createWorkflow 这类前缀误判。
|
||
return re.search(rf"(?<![A-Za-z0-9_]){re.escape(operation_id)}(?![A-Za-z0-9_])", text) is not None
|
||
|
||
|
||
def files_matching_domain_services(files: list[str], domain: str) -> list[str]:
|
||
domain_path = "/module/" + ("member" if domain == "account" else domain) + "/"
|
||
return [file_path for file_path in files if domain_path in file_path]
|
||
|
||
|
||
def normalize_path(path: str) -> str:
|
||
if path.startswith("/app-api/") or path.startswith("/admin-api/"):
|
||
return path
|
||
return path
|
||
|
||
|
||
def detect_side(path: str) -> str:
|
||
if path.startswith("/admin-api/"):
|
||
return "admin"
|
||
if path.startswith("/app-api/"):
|
||
return "app"
|
||
return "unknown"
|
||
|
||
|
||
def operation_requires_command_id(operation_doc: dict[str, Any], document: dict[str, Any]) -> bool:
|
||
if operation_has_required_command_header(operation_doc, document):
|
||
return True
|
||
|
||
# 只把 schema.required 中的 commandId 视为必填；properties 或 description 里的 commandId 不代表请求必须携带。
|
||
for content_doc in (operation_doc.get("requestBody") or {}).get("content", {}).values():
|
||
if not isinstance(content_doc, dict):
|
||
continue
|
||
schema_doc = content_doc.get("schema")
|
||
if isinstance(schema_doc, dict) and local_schema_requires_command_id(schema_doc, document, set()):
|
||
return True
|
||
return False
|
||
|
||
|
||
def operation_has_required_command_header(operation_doc: dict[str, Any], document: dict[str, Any]) -> bool:
|
||
for parameter_doc in operation_doc.get("parameters") or []:
|
||
if not isinstance(parameter_doc, dict):
|
||
continue
|
||
resolved = resolve_parameter(parameter_doc, document)
|
||
if not isinstance(resolved, dict):
|
||
continue
|
||
if resolved.get("in") == "header" and resolved.get("name") == "X-Command-Id" and resolved.get("required") is True:
|
||
return True
|
||
return False
|
||
|
||
|
||
def resolve_parameter(parameter_doc: dict[str, Any], document: dict[str, Any]) -> Any:
|
||
ref = parameter_doc.get("$ref")
|
||
if isinstance(ref, str) and ref.startswith("#/"):
|
||
return resolve_local_ref(document, ref)
|
||
return parameter_doc
|
||
|
||
|
||
def local_schema_requires_command_id(schema_doc: dict[str, Any], document: dict[str, Any], seen_refs: set[str]) -> bool:
|
||
required_fields = schema_doc.get("required") or []
|
||
if "commandId" in required_fields:
|
||
return True
|
||
|
||
ref = schema_doc.get("$ref")
|
||
if isinstance(ref, str) and ref.startswith("#/") and ref not in seen_refs:
|
||
seen_refs.add(ref)
|
||
resolved = resolve_local_ref(document, ref)
|
||
if isinstance(resolved, dict) and local_schema_requires_command_id(resolved, document, seen_refs):
|
||
return True
|
||
|
||
# 只递归 OpenAPI 组合 schema；不递归 properties，避免把可选 commandId 误判成必填。
|
||
for key in ("allOf", "anyOf", "oneOf"):
|
||
for nested_schema in schema_doc.get(key) or []:
|
||
if isinstance(nested_schema, dict) and local_schema_requires_command_id(nested_schema, document, seen_refs):
|
||
return True
|
||
return False
|
||
|
||
|
||
def resolve_local_ref(document: dict[str, Any], ref: str) -> Any:
|
||
current: Any = document
|
||
for part in ref.removeprefix("#/").split("/"):
|
||
if not isinstance(current, dict):
|
||
return None
|
||
current = current.get(part)
|
||
return current
|
||
|
||
|
||
def is_sse_placeholder_controller(text: str) -> bool:
|
||
if "SseEmitter" not in text:
|
||
return False
|
||
if any(marker in text for marker in SSE_PLACEHOLDER_READY_MARKERS):
|
||
return True
|
||
if ".name(\"done\")" in text:
|
||
return True
|
||
return "stream ready" in text and "emitter.send(SseEmitter.event()" in text and "emitter.complete();" in text
|
||
|
||
|
||
def sse_placeholder_route_mappings(text: str) -> list[RouteMapping]:
|
||
if not is_sse_placeholder_controller(text):
|
||
return []
|
||
|
||
mappings: set[RouteMapping] = set()
|
||
markers = list(SSE_PLACEHOLDER_READY_MARKERS) + [".name(\"done\")", "stream ready"]
|
||
for marker in markers:
|
||
search_from = 0
|
||
while True:
|
||
marker_index = text.find(marker, search_from)
|
||
if marker_index == -1:
|
||
break
|
||
annotation_index = nearest_mapping_annotation_before(text, marker_index)
|
||
if annotation_index != -1:
|
||
for match in re.finditer(
|
||
r"@(Get|Post|Put|Patch|Delete|Request)Mapping(?:\s*\((.*?)\))?",
|
||
text[annotation_index:marker_index],
|
||
flags=re.DOTALL,
|
||
):
|
||
mappings.update(route_mappings_from_annotation(match.group(1), match.group(2) or ""))
|
||
search_from = marker_index + len(marker)
|
||
return sorted(mappings, key=lambda mapping: (mapping.methods, mapping.routes))
|
||
|
||
|
||
def nearest_mapping_annotation_before(text: str, marker_index: int) -> int:
|
||
mapping_annotations = ["@GetMapping", "@PostMapping", "@PutMapping", "@PatchMapping", "@DeleteMapping", "@RequestMapping"]
|
||
return max(text.rfind(annotation, 0, marker_index) for annotation in mapping_annotations)
|
||
|
||
|
||
def detect_external_dependencies(domain: str, path: str, operation_id: str) -> list[str]:
|
||
haystack = f"{domain} {path} {operation_id}".lower()
|
||
dependencies: set[str] = set()
|
||
if any(word in haystack for word in ["ai", "task", "suggestion", "agent", "job", "source-event"]):
|
||
dependencies.update(["New-API", "SSE"])
|
||
if any(word in haystack for word in ["knowledge", "kb", "document", "entity", "relation", "graph", "reindex"]):
|
||
dependencies.add("RAGFlow")
|
||
if any(word in haystack for word in ["import", "export", "download", "file", "document"]):
|
||
dependencies.add("FileService")
|
||
if any(word in haystack for word in ["marketplace", "purchase", "install", "handoff", "license"]):
|
||
dependencies.update(["Account", "MarketAuthorization"])
|
||
if any(word in haystack for word in ["account", "usage", "quota", "balance", "correlation", "new-api-binding"]):
|
||
dependencies.update(["New-API", "AccountLedger"])
|
||
return sorted(dependencies)
|
||
|
||
|
||
def target_stage(domain: str) -> str:
|
||
return {
|
||
"content": "P1R-1 Content Real API",
|
||
"meta": "P1R-2 Meta Real API",
|
||
"account": "P1R-3 Account Real API",
|
||
"ai": "P1R-4 AI Real API",
|
||
"knowledge": "P1R-5 Knowledge Real API",
|
||
"market": "P1R-6 Market Real API",
|
||
"events": "P1R-7 End-to-End Acceptance",
|
||
}.get(domain, "P1R-0 Baseline Gate")
|
||
|
||
|
||
def build_report(operations: list[Operation]) -> dict[str, Any]:
|
||
summary = {
|
||
"totalOperations": len(operations),
|
||
"completedOperations": sum(1 for operation in operations if operation.completionStatus == "completed"),
|
||
"incompleteOperations": sum(1 for operation in operations if operation.completionStatus == "incomplete"),
|
||
"needsVerificationOperations": sum(1 for operation in operations if operation.completionStatus == "needs_verification"),
|
||
"catchAllOperations": sum(1 for operation in operations if operation.implementationStatus == "catch_all"),
|
||
"genericPersistenceOperations": sum(1 for operation in operations if operation.implementationStatus == "generic_persistence"),
|
||
"ssePlaceholderOperations": sum(1 for operation in operations if operation.implementationStatus == "sse_placeholder"),
|
||
"missingOperations": sum(1 for operation in operations if operation.implementationStatus == "missing"),
|
||
"blockedOperations": sum(1 for operation in operations if operation.completionStatus == "blocked"),
|
||
}
|
||
return {
|
||
"generatedAt": stable_generated_at(operations),
|
||
"sourceContracts": [str(path.relative_to(REPO_ROOT)) for path in sorted(CONTRACT_ROOT.glob("*/openapi.yaml"))],
|
||
"summary": summary,
|
||
"operations": [asdict(operation) for operation in operations],
|
||
}
|
||
|
||
|
||
def stable_generated_at(operations: list[Operation]) -> str:
|
||
# 让提交的基线报告可重复生成，避免每次 --check 只因为当前时间产生无意义 diff。
|
||
payload = json.dumps([asdict(operation) for operation in operations], ensure_ascii=False, sort_keys=True)
|
||
digest_seconds = int(hashlib.sha256(payload.encode("utf-8")).hexdigest()[:8], 16) % 86400
|
||
hours, remainder = divmod(digest_seconds, 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
return f"2026-05-25T{hours:02d}:{minutes:02d}:{seconds:02d}+00:00"
|
||
|
||
|
||
def render_markdown(report: dict[str, Any]) -> str:
|
||
lines = [
|
||
"# P1R API Coverage Report",
|
||
"",
|
||
f"- Generated at: `{report['generatedAt']}`",
|
||
f"- Total operations: `{report['summary']['totalOperations']}`",
|
||
f"- Completed: `{report['summary']['completedOperations']}`",
|
||
f"- Needs verification: `{report['summary']['needsVerificationOperations']}`",
|
||
f"- Incomplete: `{report['summary']['incompleteOperations']}`",
|
||
f"- Catch-all: `{report['summary']['catchAllOperations']}`",
|
||
f"- Generic persistence: `{report['summary']['genericPersistenceOperations']}`",
|
||
f"- SSE placeholder: `{report['summary']['ssePlaceholderOperations']}`",
|
||
f"- Missing: `{report['summary']['missingOperations']}`",
|
||
"",
|
||
"## Operations",
|
||
"",
|
||
"| Domain | Side | Method | Path | Operation | Implementation | Completion | Target Stage |",
|
||
"|--------|------|--------|------|-----------|----------------|------------|--------------|",
|
||
]
|
||
for operation in report["operations"]:
|
||
lines.append(
|
||
"| {domain} | {side} | {method} | `{path}` | `{operationId}` | {implementationStatus} | {completionStatus} | {targetStage} |".format(
|
||
**operation
|
||
)
|
||
)
|
||
lines.extend(["", "## Incomplete Or Blocked", ""])
|
||
for operation in report["operations"]:
|
||
if operation["completionStatus"] in {"incomplete", "blocked"}:
|
||
lines.append(
|
||
f"- `{operation['operationId']}` `{operation['method']} {operation['path']}`: "
|
||
f"{operation['implementationStatus']} - {operation['notes']}"
|
||
)
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def validate_report(report: dict[str, Any]) -> None:
|
||
required_fields = {
|
||
"domain",
|
||
"side",
|
||
"method",
|
||
"path",
|
||
"operationId",
|
||
"isWrite",
|
||
"requiresCommandId",
|
||
"externalDependencies",
|
||
"targetStage",
|
||
"implementationStatus",
|
||
"completionStatus",
|
||
"controllerFiles",
|
||
"serviceFiles",
|
||
"notes",
|
||
}
|
||
if report["summary"]["totalOperations"] <= 0:
|
||
raise SystemExit("No operations found")
|
||
for operation in report["operations"]:
|
||
missing = required_fields - operation.keys()
|
||
if missing:
|
||
raise SystemExit(f"Operation {operation.get('operationId')} missing fields: {sorted(missing)}")
|
||
if (
|
||
operation["implementationStatus"] in NON_REAL_IMPLEMENTATION_STATUSES
|
||
and operation["completionStatus"] not in NON_REAL_COMPLETION_STATUSES
|
||
):
|
||
raise SystemExit(
|
||
f"{operation['operationId']} non-real implementation must be incomplete or blocked"
|
||
)
|
||
if operation["completionStatus"] == "completed" and not approved_completed_operation(operation):
|
||
raise SystemExit(f"{operation['operationId']} is incorrectly marked completed")
|
||
if "message" in json.dumps(operation, ensure_ascii=False):
|
||
raise SystemExit(f"{operation['operationId']} contains forbidden response evidence field: message")
|
||
|
||
|
||
def approved_completed_operation(operation: dict[str, Any]) -> bool:
|
||
# completed 只能来自已批准推进的 dedicated 域或单项 operation，避免非目标 domain 被误计完成。
|
||
if operation["implementationStatus"] != "dedicated":
|
||
return False
|
||
if operation["domain"] in APPROVED_COMPLETED_DOMAINS:
|
||
return True
|
||
return operation_key(operation["domain"], operation["operationId"]) in APPROVED_COMPLETED_OPERATIONS
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|