#!/usr/bin/env python3 import argparse import difflib import re import sys import unicodedata from dataclasses import dataclass from pathlib import Path from urllib.parse import unquote SKILL_ROOT = Path(__file__).resolve().parents[1] DEFAULT_REFERENCES_DIR = SKILL_ROOT / "references" HEADING_RE = re.compile(r"^(#{1,6})(?!#)\s+(.+?)\s*$") FENCE_RE = re.compile(r"^```([^`]*)$") MARKDOWN_LINK_RE = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)") ASCII_TOKEN_RE = re.compile(r"[a-z_][a-z0-9_.$:+-]*", re.IGNORECASE) CHINESE_RUN_RE = re.compile(r"[\u3400-\u9fff]+") INLINE_CODE_RE = re.compile(r"`([^`\n]+)`") FENCED_CODE_RE = re.compile(r"```[^\n]*\n(.*?)```", re.DOTALL) IDENTITY_PREFIX = "代码块身份：" BLOCK_DESCRIPTION_PREFIX = "代码块说明：" ALLOWED_IDENTITIES = { "可直接照写示例", "反例 / 不可照写", "输出片段", "配置片段 / 概念骨架", "仅服务端可执行示例", } ROUTER_PHRASES = ("路由中心", "选择一个主专题", "候选页继续判断") EXCLUDED_REFERENCE_FILES = {"index.md"} DUTY_HEADING = "本篇职责" SUSPICIOUS_FENCE_RE = re.compile(r"^(?:\s+`{3}|`{4,})") WRITE_PRELUDE_ANCHORS = ( ("02_core_model.md", "文件模型核心规则"), ("01_quickstart.md", "语言核心事实速查"), ) HEADING_TOKEN_SCORE = 12 HEADING_EXACT_SCORE = 20 TERM_TOKEN_SCORE = 10 TERM_EXACT_SCORE = 16 PAGE_TITLE_TOKEN_SCORE = 5 PAGE_TITLE_EXACT_SCORE = 8 BODY_TOKEN_SCORE = 3 BODY_EXACT_SCORE = 4 DIRECT_EXAMPLE_BOOST = 8 PITFALL_PAGE_BOOST = 18 COUNTEREXAMPLE_BOOST = 8 EXACT_ERROR_BOOST = 14 MIXED_QUERY_MIN_SCORE = 10 @dataclass(frozen=True) class Section: id: str page: Path page_title: str heading_path: tuple[str, ...] body: str identities: tuple[str, ...] searchable_text: str @dataclass(frozen=True) class ValidationProblem: page: Path line: int message: str @dataclass(frozen=True) class QueryMatch: section: Section score: int priority: tuple[int, int, int, int, int] = (0, 0, 0, 0, 0) @dataclass(frozen=True) class ScoreBreakdown: heading_path: int exact_term: int page_title: int body: int mode_boost: int diagnose_priority: bool = False @property def lexical_total(self) -> int: return self.heading_path + self.exact_term + self.page_title + self.body @property def total(self) -> int: return self.lexical_total + self.mode_boost @property def priority(self) -> tuple[int, int, int, int, int]: if self.diagnose_priority: return ( self.mode_boost, self.heading_path, self.exact_term, self.page_title, self.body, ) return ( self.heading_path, self.exact_term, self.page_title, self.mode_boost, self.body, ) @dataclass class QueryResult: query: str mode: str matches: list[QueryMatch] prelude: list[Section] def normalize(value: str) -> str: return unicodedata.normalize("NFKC", value).casefold() def _slug(value: str) -> str: normalized = normalize(value) slug = re.sub(r"[^\w]+", "-", normalized, flags=re.UNICODE).strip("-_") return slug or "section" def section_id(relative_page: Path | str, heading_path: tuple[str, ...]) -> str: page = Path(relative_page) parts = [_slug(page.with_suffix("").as_posix()), *(_slug(item) for item in heading_path)] return "--".join(parts) def _reference_pages(references_dir: Path) -> list[Path]: return [ page for page in sorted(references_dir.glob("*.md"), key=lambda item: item.name) if page.name not in EXCLUDED_REFERENCE_FILES ] def _heading_records(lines: list[str]) -> tuple[str, list[tuple[int, int, str]]]: page_title = "" records: list[tuple[int, int, str]] = [] in_fence = False for index, line in enumerate(lines): if FENCE_RE.match(line.rstrip("\r\n")): in_fence = not in_fence continue if in_fence: continue match = HEADING_RE.match(line.rstrip("\r\n")) if not match: continue level = len(match.group(1)) title = match.group(2) if level == 1 and not page_title: page_title = title elif level in (2, 3): records.append((index, level, title)) return page_title, records def _associated_identity(lines: list[str], opening_fence: int) -> str | None: previous = opening_fence - 1 while previous >= 0 and not lines[previous].strip(): previous -= 1 while previous >= 0 and lines[previous].strip().startswith(BLOCK_DESCRIPTION_PREFIX): previous -= 1 while previous >= 0 and not lines[previous].strip(): previous -= 1 if previous < 0: return None metadata = lines[previous].strip() if not metadata.startswith(IDENTITY_PREFIX): return None identity = metadata[len(IDENTITY_PREFIX) :].strip() earlier = previous - 1 while earlier >= 0 and not lines[earlier].strip(): earlier -= 1 if earlier >= 0 and lines[earlier].strip().startswith(IDENTITY_PREFIX): return None return identity if identity in ALLOWED_IDENTITIES else None def _identities(body: str) -> tuple[str, ...]: identities: list[str] = [] lines = body.splitlines() in_fence = False for index, line in enumerate(lines): if not FENCE_RE.match(line): continue if not in_fence: identity = _associated_identity(lines, index) if identity is not None: identities.append(identity) in_fence = not in_fence return tuple(identities) def load_sections(references_dir: Path = DEFAULT_REFERENCES_DIR) -> list[Section]: sections: list[Section] = [] id_counts: dict[str, int] = {} for page in _reference_pages(Path(references_dir)): text = page.read_text(encoding="utf-8") lines = text.splitlines(keepends=True) page_title, headings = _heading_records(lines) h2_title = "" for position, (start, level, title) in enumerate(headings): if level == 2: h2_title = title heading_path = (title,) else: heading_path = (h2_title, title) if h2_title else (title,) end = len(lines) for next_start, next_level, _ in headings[position + 1 :]: if next_level <= level: end = next_start break body = "".join(lines[start:end]) base_id = section_id(page.relative_to(references_dir), heading_path) id_counts[base_id] = id_counts.get(base_id, 0) + 1 unique_id = base_id if id_counts[base_id] > 1: unique_id = f"{base_id}-{id_counts[base_id]}" searchable_text = normalize( "\n".join((page.stem, page_title, *heading_path, body)) ) sections.append( Section( id=unique_id, page=page, page_title=page_title, heading_path=heading_path, body=body, identities=_identities(body), searchable_text=searchable_text, ) ) return sections def _identity_problems(page: Path, lines: list[str]) -> list[ValidationProblem]: problems: list[ValidationProblem] = [] for index, line in enumerate(lines, start=1): stripped = line.strip() if stripped.startswith(IDENTITY_PREFIX): identity = stripped[len(IDENTITY_PREFIX) :].strip() if identity not in ALLOWED_IDENTITIES: problems.append(ValidationProblem(page, index, f"未知身份：{identity}")) in_fence = False for index, line in enumerate(lines): if FENCE_RE.match(line): if in_fence: in_fence = False else: if _associated_identity(lines, index) is None: problems.append( ValidationProblem( page, index + 1, "每个代码围栏必须关联恰好一个代码块身份" ) ) in_fence = True continue if not in_fence and SUSPICIOUS_FENCE_RE.match(line): problems.append( ValidationProblem( page, index + 1, "不支持的代码围栏形态（缩进围栏或四个及以上反引号）" ) ) if in_fence: problems.append(ValidationProblem(page, len(lines), "代码围栏未闭合")) return problems def _local_link_problems( page: Path, text: str, references_dir: Path ) -> list[ValidationProblem]: problems: list[ValidationProblem] = [] # Blank out fenced/inline code without changing line offsets, so reported # line numbers line up with the original file. without_fences = re.sub( r"```.*?```", lambda match: re.sub(r"[^\n]", " ", match.group(0)), text, flags=re.DOTALL, ) searchable_markdown = re.sub( r"`[^`\n]*`", lambda match: " " * len(match.group(0)), without_fences ) for match in MARKDOWN_LINK_RE.finditer(searchable_markdown): target = match.group(1).strip().split(maxsplit=1)[0].strip("<>") if target.startswith(("#", "http://", "https://", "mailto:")): continue target_path = unquote(target.split("#", 1)[0].replace("\\", "/")) resolved = (page.parent / target_path).resolve() try: resolved.relative_to(references_dir.resolve()) except ValueError: exists = False else: exists = resolved.is_file() if not exists: line = searchable_markdown.count("\n", 0, match.start()) + 1 problems.append(ValidationProblem(page, line, f"本地链接不存在：{target}")) return problems def validate_references( references_dir: Path = DEFAULT_REFERENCES_DIR, ) -> list[ValidationProblem]: references_dir = Path(references_dir) problems: list[ValidationProblem] = [] index_page = references_dir / "index.md" if index_page.exists(): problems.append(ValidationProblem(index_page, 1, "references 中不得保留 index.md")) for page in sorted(references_dir.glob("*.md"), key=lambda item: item.name): text = page.read_text(encoding="utf-8") lines = text.splitlines() problems.extend(_identity_problems(page, lines)) problems.extend(_local_link_problems(page, text, references_dir)) for phrase in ROUTER_PHRASES: for index, line in enumerate(lines, start=1): if phrase in line: problems.append( ValidationProblem(page, index, f"包含人工路由协议：{phrase}") ) sections = load_sections(references_dir) ids: dict[str, Section] = {} for section in sections: if section.id in ids: problems.append(ValidationProblem(section.page, 1, f"重复 section ID：{section.id}")) ids[section.id] = section for page_name, heading in WRITE_PRELUDE_ANCHORS: page_sections = [ section for section in sections if section.page.name == page_name ] # Only enforce the anchor when the page is present, so validating a # synthetic references dir (tests) does not demand the bundled pages. if page_sections and not any( heading in section.heading_path for section in page_sections ): problems.append( ValidationProblem( references_dir / page_name, 1, f"write 模式前置章节缺失：{page_name} 的「{heading}」", ) ) # 概念地图逐页从「本篇职责」段生成；有该段的页必须产出非空摘要， # 否则某页职责段被清空/写坏时地图会静默缺页。 mapped_pages = {page_name for page_name, _, _ in build_concept_map(references_dir)} for section in sections: if section.heading_path != (DUTY_HEADING,): continue if section.page.name not in mapped_pages: problems.append( ValidationProblem( section.page, 1, f"概念地图摘要为空：{section.page.name} 的「{DUTY_HEADING}」", ) ) return problems def query_tokens(text: str) -> set[str]: normalized = normalize(text) tokens = set(ASCII_TOKEN_RE.findall(normalized)) for run in CHINESE_RUN_RE.findall(normalized): tokens.add(run) tokens.update(run[index : index + 2] for index in range(len(run) - 1)) return {token for token in tokens if token.strip()} def _has_chinese_context(section: Section, query: str) -> bool: runs = CHINESE_RUN_RE.findall(normalize(query)) for run in runs: tokens = ( [run] if len(run) < 2 else [run[index : index + 2] for index in range(len(run) - 1)] ) matched = sum(token in section.searchable_text for token in tokens) if matched >= (len(tokens) + 1) // 2: return True return not runs def _code_text(body: str) -> str: inline = INLINE_CODE_RE.findall(body) fenced = FENCED_CODE_RE.findall(body) return normalize("\n".join((*inline, *fenced))) def _score_section(section: Section, query: str, mode: str) -> ScoreBreakdown: normalized_query = normalize(query).strip() tokens = query_tokens(query) heading_text = normalize("\n".join(section.heading_path)) page_title_text = normalize(section.page_title) body_text = normalize(section.body) term_text = _code_text(section.body) heading_score = 0 term_score = 0 page_title_score = 0 body_score = 0 for token in tokens: if token in heading_text: heading_score += HEADING_TOKEN_SCORE if token in term_text: term_score += TERM_TOKEN_SCORE if token in page_title_text: page_title_score += PAGE_TITLE_TOKEN_SCORE if token in body_text: body_score += BODY_TOKEN_SCORE if normalized_query and normalized_query in heading_text: heading_score += HEADING_EXACT_SCORE if normalized_query and normalized_query in term_text: term_score += TERM_EXACT_SCORE if normalized_query and normalized_query in page_title_text: page_title_score += PAGE_TITLE_EXACT_SCORE if normalized_query and normalized_query in body_text: body_score += BODY_EXACT_SCORE mode_boost = 0 if mode == "write" and "可直接照写示例" in section.identities: mode_boost += DIRECT_EXAMPLE_BOOST if mode == "diagnose": if section.page.name == "11_pitfalls.md": mode_boost += PITFALL_PAGE_BOOST if "反例 / 不可照写" in section.identities: mode_boost += COUNTEREXAMPLE_BOOST if normalized_query and normalized_query in section.searchable_text: mode_boost += EXACT_ERROR_BOOST return ScoreBreakdown( heading_path=heading_score, exact_term=term_score, page_title=page_title_score, body=body_score, mode_boost=mode_boost, diagnose_priority=mode == "diagnose", ) def _write_prelude(sections: list[Section]) -> list[Section]: prelude: list[Section] = [] for page_name, heading in WRITE_PRELUDE_ANCHORS: match = next( ( section for section in sections if section.page.name == page_name and heading in section.heading_path ), None, ) if match is not None: prelude.append(match) return prelude STRING_LITERAL_RE = re.compile(r'"[^"\n]*"|\'[^\'\n]*\'') def _bare_tokens(tokens: set[str]) -> set[str]: # ASCII_TOKEN_RE 的连续字符类会把尾缀符号吞进 token（如编译开关 # `{$varByRef-}` 产出 `varbyref-`）；按剥掉尾部符号的裸形态比较。 return {token.rstrip(".$:+-") for token in tokens} - {""} def _ascii_anchor(section: Section, ascii_tokens: set[str]) -> bool: # 只有足够长的标识符按标识符位置命中（标题、行内代码或剥离字符串 # 字面量后的围栏代码）才豁免中文门控；短 token 与字符串样例数据 # （如 "XYZ"、"xyz*"）不算点名，避免样例值驱动召回。 anchor_tokens = { token for token in _bare_tokens(ascii_tokens) if len(token) >= 4 } if not anchor_tokens: return False heading_tokens = set( ASCII_TOKEN_RE.findall(normalize("\n".join(section.heading_path))) ) code_text = "\n".join( (*INLINE_CODE_RE.findall(section.body), *FENCED_CODE_RE.findall(section.body)) ) code_tokens = set( ASCII_TOKEN_RE.findall(normalize(STRING_LITERAL_RE.sub(" ", code_text))) ) return bool(anchor_tokens & _bare_tokens(heading_tokens | code_tokens)) def _related_sections(a: Section, b: Section) -> bool: if a.page != b.page or len(a.heading_path) == len(b.heading_path): return False shorter, longer = sorted((a.heading_path, b.heading_path), key=len) return longer[: len(shorter)] == shorter def query_sections( query: str, mode: str, limit: int = 5, references_dir: Path = DEFAULT_REFERENCES_DIR, ) -> QueryResult: if mode not in {"write", "diagnose", "explain"}: raise ValueError(f"unsupported mode: {mode}") if not 1 <= limit <= 10: raise ValueError("limit must be between 1 and 10") all_sections = load_sections(references_dir) sections = all_sections ascii_tokens = { token for token in query_tokens(query) if ASCII_TOKEN_RE.fullmatch(token) } if ascii_tokens: sections = [ section for section in sections if any(token in section.searchable_text for token in ascii_tokens) ] has_chinese = bool(CHINESE_RUN_RE.search(normalize(query))) minimum_score = MIXED_QUERY_MIN_SCORE if ascii_tokens and has_chinese else 1 ranked: list[QueryMatch] = [] for section in sections: # 中文上下文门控只裁剪正文级 ASCII 噪声；标识符在标题或代码里 # 精确命中的 section 不因中文措辞不同而被丢弃。 if ( ascii_tokens and has_chinese and not _ascii_anchor(section, ascii_tokens) and not _has_chinese_context(section, query) ): continue score = _score_section(section, query, mode) if score.lexical_total < minimum_score: continue ranked.append(QueryMatch(section, score.total, score.priority)) ranked.sort( key=lambda match: ( *(-value for value in match.priority), match.section.page.as_posix(), match.section.id, ) ) # H2 聚合 section 的正文逐字包含其 H3 子节；父子同时入选时只保留 # 排名更高的一个，避免同一内容重复返回。 matches: list[QueryMatch] = [] for match in ranked: if any( _related_sections(match.section, kept.section) for kept in matches ): continue matches.append(match) if len(matches) == limit: break prelude = _write_prelude(all_sections) if mode == "write" else [] match_ids = {match.section.id for match in matches} prelude = [section for section in prelude if section.id not in match_ids] return QueryResult(query=query, mode=mode, matches=matches, prelude=prelude) def render_result(result: QueryResult) -> str: lines = [ "# TSL Syntax Lookup", "", f"Mode: `{result.mode}`", f"Query: {result.query}", ] if result.prelude: lines.extend(["", "## Required context"]) for section in result.prelude: lines.extend( [ "", f"Section ID: `{section.id}`", f"Source: `{section.page.as_posix()}`", "", section.body.rstrip(), ] ) for index, match in enumerate(result.matches, start=1): lines.extend( [ "", f"## Match {index}", "", f"Score: {match.score}", f"Section ID: `{match.section.id}`", f"Source: `{match.section.page.as_posix()}`", "", match.section.body.rstrip(), ] ) lines.extend(["", "## Narrower section IDs"]) lines.extend(f"- `{match.section.id}`" for match in result.matches) return "\n".join(lines).rstrip() + "\n" def _duty_summary(body: str) -> str: lines = body.splitlines() paragraph: list[str] = [] for line in lines[1:]: stripped = line.strip() if not stripped: if paragraph: break continue paragraph.append(stripped) summary = " ".join(paragraph) # 职责段以冒号引出列表时（如 09），首段本身空洞，把随后的 # 列表项折叠进摘要才能保留实际覆盖面。 if summary.endswith(("：", ":")): items: list[str] = [] seen_paragraph = False for line in lines[1:]: stripped = line.strip() if not stripped: continue if not seen_paragraph: if stripped == summary or stripped in summary: seen_paragraph = True continue if stripped.startswith(("-", "*", "·")): items.append(stripped.lstrip("-*· ").strip()) elif items: break if items: summary = summary + "；".join(items) + "。" return summary def build_concept_map(references_dir: Path = DEFAULT_REFERENCES_DIR) -> list[tuple[str, str, str]]: entries: list[tuple[str, str, str]] = [] seen_pages: set[str] = set() for section in load_sections(references_dir): page_name = section.page.name if page_name in seen_pages: continue if section.heading_path != (DUTY_HEADING,): continue summary = _duty_summary(section.body) if not summary: continue seen_pages.add(page_name) entries.append((page_name, section.page_title, summary)) entries.sort(key=lambda item: item[0]) return entries def render_concept_map(entries: list[tuple[str, str, str]]) -> str: lines = [ "# TSL 概念地图", "", "把自然语言需求映射到该查哪个 TSL 概念，随后仍用 `--query` 取精确语法。", "本清单不含语法细节，也不替代 lookup；用命中专题里的关键语法词组成查询。", ] for page_name, page_title, summary in entries: lines.extend(["", f"## {page_title}", "", summary]) return "\n".join(lines).rstrip() + "\n" def _configure_utf8() -> None: for stream in (sys.stdout, sys.stderr): reconfigure = getattr(stream, "reconfigure", None) if reconfigure is not None: reconfigure(encoding="utf-8") def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Search TSL syntax reference sections") action = parser.add_mutually_exclusive_group(required=True) action.add_argument("--query") action.add_argument("--section") action.add_argument("--map", dest="show_map", action="store_true") action.add_argument("--check", action="store_true") parser.add_argument("--mode", choices=("write", "diagnose", "explain")) parser.add_argument("--limit", type=int, default=5) parser.add_argument("--references-dir", type=Path, default=DEFAULT_REFERENCES_DIR) return parser def _nearest_section_ids( requested: str, sections: list[Section], limit: int = 5 ) -> list[str]: ranked = sorted( sections, key=lambda section: ( -difflib.SequenceMatcher(None, requested, section.id).ratio(), section.id, ), ) return [section.id for section in ranked[:limit]] def main(argv: list[str] | None = None) -> int: _configure_utf8() parser = _parser() args = parser.parse_args(argv) if not 1 <= args.limit <= 10: parser.error("--limit must be between 1 and 10") if args.query is not None and args.mode is None: parser.error("--mode is required with --query") if args.mode is not None and args.query is None: parser.error("--mode only applies to --query") if args.show_map: print(render_concept_map(build_concept_map(args.references_dir)), end="") return 0 if args.check: problems = validate_references(args.references_dir) for problem in problems: print(f"{problem.page}:{problem.line}: {problem.message}", file=sys.stderr) return 1 if problems else 0 if args.section is not None: sections = load_sections(args.references_dir) section = next( (item for item in sections if item.id == args.section), None, ) if section is None: print(f"section not found: {args.section}", file=sys.stderr) print("Nearest section IDs:", file=sys.stderr) for candidate in _nearest_section_ids(args.section, sections): print(f"- {candidate}", file=sys.stderr) return 2 result = QueryResult("", "section", [QueryMatch(section, 0)], []) else: result = query_sections(args.query, args.mode, args.limit, args.references_dir) if not result.matches: print("no matching sections", file=sys.stderr) return 2 print(render_result(result), end="") return 0 if __name__ == "__main__": raise SystemExit(main())