276 lines
9.2 KiB
Python
Executable File
276 lines
9.2 KiB
Python
Executable File
#!/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())
|