#!/usr/bin/env python3 """按显式项目上下文确定性选择 active ACK 知识。 匹配采用大小写敏感 glob；entry 中每个非空 scope 维度都必须被查询上下文命中。 scope.all=true 的条目始终命中并优先占用 --limit，数量超过预算时显式失败。其余 结果按作用域具体程度及稳定引用排序。本脚本只输出数据，绝不执行 knowledge.yaml 中的任何文本。 """ from __future__ import annotations import argparse import fnmatch import json import sys from functools import lru_cache from pathlib import Path from typing import Any from validate_knowledge import ( # type: ignore SCOPE_FIELDS, infer_project_root, load_yaml, stable_ref, validate_builtin_structure, validate_semantics, ) DEFAULT_LIMIT = 10 MAX_LIMIT = 100 def _path_glob_matches(value: str, pattern: str) -> bool: """路径 glob：* 只匹配单段，只有完整的 ** 段可以跨越 /。""" value_parts = tuple(value.split("/")) pattern_parts = tuple(pattern.split("/")) @lru_cache(maxsize=None) def match(pattern_index: int, value_index: int) -> bool: if pattern_index == len(pattern_parts): return value_index == len(value_parts) pattern_part = pattern_parts[pattern_index] if pattern_part == "**": return match(pattern_index + 1, value_index) or ( value_index < len(value_parts) and match(pattern_index, value_index + 1) ) return ( value_index < len(value_parts) and fnmatch.fnmatchcase(value_parts[value_index], pattern_part) and match(pattern_index + 1, value_index + 1) ) return match(0, 0) def scope_matches( scope: dict[str, Any], context: dict[str, list[str]] ) -> bool: if scope.get("all") is True: return True constrained = False for field in SCOPE_FIELDS: patterns = scope.get(field) if not isinstance(patterns, list) or not patterns: continue constrained = True values = context.get(field, []) matcher = _path_glob_matches if field == "paths" else fnmatch.fnmatchcase if not values or not any( matcher(value, pattern) for pattern in patterns if isinstance(pattern, str) for value in values ): return False return constrained def _pattern_specificity(pattern: str) -> tuple[int, int, int, int, int]: wildcard_count = sum(pattern.count(char) for char in ("*", "?", "[")) double_star_count = sum(1 for part in pattern.split("/") if part == "**") literal_count = sum(char not in "*?[]!" for char in pattern) exact = int(wildcard_count == 0) depth = len(pattern.split("/")) return (exact, literal_count, -double_star_count, -wildcard_count, depth) def _specificity(entry: dict[str, Any]) -> tuple[int, int, int, int, int, int, int]: scope = entry.get("scope") if not isinstance(scope, dict) or scope.get("all") is True: return (0, 0, 0, 0, 0, 0, 0) populated = 0 exact = 0 literal = 0 double_star = 0 wildcard = 0 depth = 0 extra_or_patterns = 0 for field in SCOPE_FIELDS: values = scope.get(field) if isinstance(values, list) and values: populated += 1 scores = [ _pattern_specificity(value) for value in values if isinstance(value, str) ] if scores: dimension_score = min(scores) exact += dimension_score[0] literal += dimension_score[1] double_star += dimension_score[2] wildcard += dimension_score[3] depth += dimension_score[4] extra_or_patterns += len(scores) - 1 return ( populated, exact, literal, double_star, wildcard, depth, -extra_or_patterns, ) def select_entries( data: dict[str, Any], context: dict[str, list[str]], *, limit: int = DEFAULT_LIMIT, ) -> list[dict[str, Any]]: if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= MAX_LIMIT: raise ValueError(f"limit 必须在 1..{MAX_LIMIT} 之间") entries = data.get("entries") if not isinstance(entries, list): return [] matched = [ entry for entry in entries if isinstance(entry, dict) and entry.get("status") == "active" and isinstance(entry.get("scope"), dict) and scope_matches(entry["scope"], context) and stable_ref(entry) is not None ] global_entries = [ entry for entry in matched if isinstance(entry.get("scope"), dict) and entry["scope"].get("all") is True ] scoped_entries = [entry for entry in matched if entry not in global_entries] global_entries.sort(key=lambda entry: stable_ref(entry) or "") if len(global_entries) > limit: raise ValueError( f"命中的全项目知识有 {len(global_entries)} 条，超过 --limit={limit}；" "提高 limit 后重试，不能静默丢弃全项目护栏" ) scoped_entries.sort( key=lambda entry: ( *(-part for part in _specificity(entry)), stable_ref(entry) or "", ) ) return [ *global_entries, *scoped_entries[: limit - len(global_entries)], ] def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="选择当前任务适用的 active ACK 知识") parser.add_argument( "knowledge", nargs="?", default="knowledge.yaml", help="知识库路径" ) parser.add_argument("--component", action="append", default=[]) parser.add_argument("--path", action="append", default=[]) parser.add_argument("--dependency", action="append", default=[]) parser.add_argument("--version", action="append", default=[]) parser.add_argument("--tag", action="append", default=[]) parser.add_argument("--symbol", action="append", default=[]) parser.add_argument("--error-signature", action="append", default=[]) parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT) parser.add_argument( "--project-root", help="可选项目根目录，用于 verificationRegistry symlink containment 校验", ) parser.add_argument( "--format", choices=("json", "refs"), default="json", dest="output_format" ) return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) knowledge_path = Path(args.knowledge) if not knowledge_path.is_file(): sys.stderr.write(f"找不到知识库文件: {knowledge_path}\n") return 2 if not 1 <= args.limit <= MAX_LIMIT: sys.stderr.write(f"--limit 必须在 1..{MAX_LIMIT} 之间\n") return 2 if args.project_root: project_root = Path(args.project_root).expanduser() if not project_root.is_dir(): sys.stderr.write(f"项目根目录不存在: {project_root}\n") return 2 project_root = project_root.resolve() else: project_root = infer_project_root(knowledge_path) data = load_yaml(knowledge_path, "知识库") errors = validate_builtin_structure(data) errors.extend(validate_semantics(data, project_root=project_root)) registry = data.get("verificationRegistry") if project_root is None and isinstance(registry, dict) and registry: errors.append( "verificationRegistry 非空但无法确定项目根目录；请传入 --project-root" ) errors = list(dict.fromkeys(errors)) if errors: sys.stderr.write(f"知识库无效，拒绝选择，共 {len(errors)} 项：\n") for error in errors: sys.stderr.write(f" - {error}\n") return 1 context = { "components": args.component, "paths": args.path, "dependencies": args.dependency, "versions": args.version, "tags": args.tag, "symbols": args.symbol, "errorSignatures": args.error_signature, } try: selected = select_entries(data, context, limit=args.limit) except ValueError as exc: sys.stderr.write(f"知识选择失败: {exc}\n") return 1 refs = [stable_ref(entry) for entry in selected] if args.output_format == "refs": if refs: sys.stdout.write("\n".join(ref for ref in refs if ref) + "\n") return 0 payload = { "count": len(selected), "limit": args.limit, "refs": refs, "entries": [ { "ref": stable_ref(entry), **entry, "verificationTarget": ( data.get("verificationRegistry", {}).get( entry.get("verification", {}).get("ref") ) if isinstance(data.get("verificationRegistry"), dict) and isinstance(entry.get("verification"), dict) else None ), } for entry in selected ], } sys.stdout.write(json.dumps(payload, ensure_ascii=False, indent=2) + "\n") return 0 if __name__ == "__main__": raise SystemExit(main())