1021 lines
35 KiB
Python
1021 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
import argparse
|
||
import difflib
|
||
import json
|
||
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 = 30
|
||
COUNTEREXAMPLE_BOOST = 8
|
||
EXACT_ERROR_BOOST = 14
|
||
MIXED_QUERY_MIN_SCORE = 10
|
||
PAGE_INTENT_SCORE = 80
|
||
|
||
CHINESE_STOP_TOKENS = {
|
||
"一个",
|
||
"为什",
|
||
"什么",
|
||
"怎么",
|
||
"怎样",
|
||
"是否",
|
||
"能不",
|
||
"不能",
|
||
"帮我",
|
||
"想要",
|
||
"然后",
|
||
"里面",
|
||
}
|
||
|
||
ASCII_FILTER_STOP_TOKENS = {
|
||
"debug",
|
||
"please",
|
||
"program",
|
||
"tinysoft",
|
||
"tsl",
|
||
"tsf",
|
||
}
|
||
|
||
QUERY_SYNONYMS = {
|
||
"打印": ("输出", "writeLn"),
|
||
"打出来": ("输出", "writeLn"),
|
||
"左连接": ("左联接", "left join", "TS-SQL"),
|
||
"左外连接": ("左联接", "left join", "TS-SQL"),
|
||
"left outer join": ("left join", "左联接", "TS-SQL"),
|
||
"列表": ("数组",),
|
||
"复用文件": ("tsf", "unit"),
|
||
"多个文件": ("unit", "uses", "作用域"),
|
||
"跳出去": ("break", "控制流"),
|
||
"程序慢": ("性能分析", "计时", "profiler"),
|
||
"瓶颈": ("性能分析", "profiler"),
|
||
"debug": ("调试", "性能分析"),
|
||
"program": ("脚本",),
|
||
"tinysoft": ("天软", "TSL"),
|
||
"字符串转整数": ("类型转换", "strToInt"),
|
||
"高性能矩阵": ("FMArray",),
|
||
}
|
||
|
||
PAGE_INTENT_ALIASES = {
|
||
"01_quickstart.md": ("最简单能跑", "天软脚本", "tinysoft"),
|
||
"02_core_model.md": ("脚本和可复用", "可复用函数文件"),
|
||
"03_values_and_literals.md": ("字符串和数组下标",),
|
||
"04_variables_and_constants.md": ("常量怎么声明", "变量能不能直接赋值"),
|
||
"05_functions_and_calls.md": ("默认参数", "函数怎么带"),
|
||
"06_expressions_and_operators.md": ("赋值和相等比较",),
|
||
"07_control_flow.md": ("跳出去", "循环里满足条件"),
|
||
"08_objects_and_classes.md": ("定义类", "创建对象"),
|
||
"09_units_and_scope.md": ("多个文件", "复用一组函数"),
|
||
"10_runtime_context_and_with.md": ("临时切换系统参数",),
|
||
"11_pitfalls.md": ("声明函数后面写代码", "语法报错"),
|
||
"12_matrix_and_collections.md": ("某行存在", "二维数组怎么判断"),
|
||
"13_resultset_and_filters.md": ("保留匹配行", "按某一列"),
|
||
"14_ts_sql.md": ("左连接", "左外连接", "左联接", "数据库", "分组排序", "聚合排序"),
|
||
"15_debug_and_profiler.md": ("程序慢", "计时找瓶颈", "性能瓶颈", "性能问题", "debug"),
|
||
"16_lexical_structure_and_compile_options.md": ("变量名区分大小写", "注释怎么写"),
|
||
"17_types_and_conversions.md": ("字符串转整数", "类型转换"),
|
||
"18_external_calls_and_threads.md": ("调用 dll", "开线程"),
|
||
"19_namespace_libpath_and_unit_runtime.md": ("找不到 tsf", "搜索路径"),
|
||
"20_object_runtime_and_introspection.md": ("查看对象属于哪个类", "运行时对象"),
|
||
"21_builtin_runtime_objects.md": ("内存流",),
|
||
"22_matrix_deep_dive.md": ("矩阵求逆", "矩阵转置", "求逆和转置"),
|
||
"23_fmarray.md": ("高性能矩阵", "fmarray"),
|
||
"24_object_overloads_and_iteration.md": ("自定义对象支持下标", "for in"),
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Section:
|
||
id: str
|
||
page: Path
|
||
page_title: str
|
||
heading_path: tuple[str, ...]
|
||
body: str
|
||
local_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, ...] = ()
|
||
reasons: tuple[str, ...] = ()
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ScoreBreakdown:
|
||
intent: int
|
||
heading_path: int
|
||
exact_term: int
|
||
page_title: int
|
||
body: int
|
||
mode_boost: int
|
||
synonym_hits: int
|
||
diagnose_priority: bool = False
|
||
|
||
@property
|
||
def lexical_total(self) -> int:
|
||
return (
|
||
self.intent
|
||
+ 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, ...]:
|
||
if self.diagnose_priority:
|
||
return (
|
||
self.intent,
|
||
self.mode_boost,
|
||
self.heading_path,
|
||
self.exact_term,
|
||
self.page_title,
|
||
self.body,
|
||
)
|
||
return (
|
||
self.intent,
|
||
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]
|
||
limit: int
|
||
|
||
|
||
def normalize(value: str) -> str:
|
||
return unicodedata.normalize("NFKC", value).casefold()
|
||
|
||
|
||
SYMBOL_SLUG_REPLACEMENTS = (
|
||
("**", " double-star "),
|
||
("[]", " index "),
|
||
("::", " double-colon "),
|
||
(":.", " colon-dot "),
|
||
("*", " star "),
|
||
)
|
||
|
||
|
||
def _slug(value: str) -> str:
|
||
normalized = normalize(value)
|
||
for symbol, replacement in SYMBOL_SLUG_REPLACEMENTS:
|
||
normalized = normalized.replace(symbol, replacement)
|
||
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, 4):
|
||
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] = []
|
||
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)
|
||
heading_stack: dict[int, str] = {}
|
||
for position, (start, level, title) in enumerate(headings):
|
||
for stacked_level in tuple(heading_stack):
|
||
if stacked_level >= level:
|
||
del heading_stack[stacked_level]
|
||
heading_stack[level] = title
|
||
heading_path = tuple(
|
||
heading_stack[item]
|
||
for item in range(2, level + 1)
|
||
if item in heading_stack
|
||
)
|
||
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])
|
||
local_end = (
|
||
headings[position + 1][0]
|
||
if position + 1 < len(headings)
|
||
else len(lines)
|
||
)
|
||
local_body = "".join(lines[start:local_end])
|
||
base_id = section_id(page.relative_to(references_dir), heading_path)
|
||
searchable_text = normalize(
|
||
"\n".join((page.stem, page_title, *heading_path, local_body))
|
||
)
|
||
sections.append(
|
||
Section(
|
||
id=base_id,
|
||
page=page,
|
||
page_title=page_title,
|
||
heading_path=heading_path,
|
||
body=body,
|
||
local_body=local_body,
|
||
identities=_identities(local_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))
|
||
page_title, headings = _heading_records(text.splitlines(keepends=True))
|
||
h1_count = 0
|
||
in_fence = False
|
||
previous_level = 1
|
||
for index, line in enumerate(lines):
|
||
if FENCE_RE.match(line):
|
||
in_fence = not in_fence
|
||
continue
|
||
if in_fence:
|
||
continue
|
||
match = HEADING_RE.match(line)
|
||
if match is None:
|
||
continue
|
||
level = len(match.group(1))
|
||
if level == 1:
|
||
h1_count += 1
|
||
previous_level = 1
|
||
continue
|
||
if level not in (2, 3, 4):
|
||
continue
|
||
if level > previous_level + 1:
|
||
problems.append(
|
||
ValidationProblem(page, index + 1, "H2/H3/H4 标题层级跳跃")
|
||
)
|
||
previous_level = level
|
||
if h1_count != 1 or not page_title:
|
||
problems.append(ValidationProblem(page, 1, "每页必须有且仅有一个非空 H1"))
|
||
duty_sections = []
|
||
for position, (start, level, title) in enumerate(headings):
|
||
if level != 2 or title != DUTY_HEADING:
|
||
continue
|
||
end = len(lines)
|
||
for next_start, next_level, _ in headings[position + 1 :]:
|
||
if next_level <= level:
|
||
end = next_start
|
||
break
|
||
body = "\n".join(lines[start + 1 : end]).strip()
|
||
duty_sections.append(body)
|
||
if len(duty_sections) != 1 or not duty_sections[0]:
|
||
problems.append(
|
||
ValidationProblem(
|
||
page, 1, "每页必须有且仅有一个非空「本篇职责」"
|
||
)
|
||
)
|
||
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)}
|
||
reference_pages = {page.name for page in _reference_pages(references_dir)}
|
||
if mapped_pages != reference_pages:
|
||
missing = ", ".join(sorted(reference_pages - mapped_pages)) or "none"
|
||
problems.append(
|
||
ValidationProblem(
|
||
references_dir,
|
||
1,
|
||
f"概念地图页数与参考页不一致；缺失：{missing}",
|
||
)
|
||
)
|
||
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 _base_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() and token not in CHINESE_STOP_TOKENS
|
||
}
|
||
|
||
|
||
def _ascii_token_sequence(text: str) -> list[str]:
|
||
return [
|
||
token.rstrip(".$:+-")
|
||
for token in ASCII_TOKEN_RE.findall(normalize(text))
|
||
if token.rstrip(".$:+-")
|
||
]
|
||
|
||
|
||
def _query_contains_phrase(query: str, phrase: str) -> bool:
|
||
normalized_phrase = normalize(phrase)
|
||
if CHINESE_RUN_RE.search(normalized_phrase):
|
||
return normalized_phrase in normalize(query)
|
||
phrase_tokens = _ascii_token_sequence(phrase)
|
||
query_tokens_in_order = _ascii_token_sequence(query)
|
||
if not phrase_tokens:
|
||
return False
|
||
width = len(phrase_tokens)
|
||
return any(
|
||
query_tokens_in_order[index : index + width] == phrase_tokens
|
||
for index in range(len(query_tokens_in_order) - width + 1)
|
||
)
|
||
|
||
|
||
def _synonym_tokens(text: str) -> set[str]:
|
||
tokens: set[str] = set()
|
||
for phrase, synonyms in QUERY_SYNONYMS.items():
|
||
if not _query_contains_phrase(text, phrase):
|
||
continue
|
||
for synonym in synonyms:
|
||
tokens.update(_base_query_tokens(synonym))
|
||
return tokens
|
||
|
||
|
||
def query_tokens(text: str) -> set[str]:
|
||
return _base_query_tokens(text) | _synonym_tokens(text)
|
||
|
||
|
||
def _text_contains_token(text: str, token: str) -> bool:
|
||
normalized_text = normalize(text)
|
||
if ASCII_TOKEN_RE.fullmatch(token):
|
||
text_tokens = _bare_tokens(set(ASCII_TOKEN_RE.findall(normalized_text)))
|
||
return token.rstrip(".$:+-") in text_tokens
|
||
return token in normalized_text
|
||
|
||
|
||
def _text_contains_exact_query(text: str, query: str) -> bool:
|
||
normalized_query = normalize(query).strip()
|
||
if ASCII_TOKEN_RE.fullmatch(normalized_query):
|
||
return _text_contains_token(text, normalized_query)
|
||
return bool(normalized_query and normalized_query in normalize(text))
|
||
|
||
|
||
def _intent_score(section: Section, query: str) -> int:
|
||
aliases = PAGE_INTENT_ALIASES.get(section.page.name, ())
|
||
return PAGE_INTENT_SCORE * sum(
|
||
_query_contains_phrase(query, alias) for alias in aliases
|
||
)
|
||
|
||
|
||
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.local_body)
|
||
term_text = _code_text(section.local_body)
|
||
expanded_only_tokens = _synonym_tokens(query) - _base_query_tokens(query)
|
||
synonym_hits = sum(
|
||
any(
|
||
_text_contains_token(text, token)
|
||
for text in (heading_text, term_text, page_title_text, body_text)
|
||
)
|
||
for token in expanded_only_tokens
|
||
)
|
||
heading_score = 0
|
||
term_score = 0
|
||
page_title_score = 0
|
||
body_score = 0
|
||
for token in tokens:
|
||
if _text_contains_token(heading_text, token):
|
||
heading_score += HEADING_TOKEN_SCORE
|
||
if _text_contains_token(term_text, token):
|
||
term_score += TERM_TOKEN_SCORE
|
||
if _text_contains_token(page_title_text, token):
|
||
page_title_score += PAGE_TITLE_TOKEN_SCORE
|
||
if _text_contains_token(body_text, token):
|
||
body_score += BODY_TOKEN_SCORE
|
||
if _text_contains_exact_query(heading_text, query):
|
||
heading_score += HEADING_EXACT_SCORE
|
||
if _text_contains_exact_query(term_text, query):
|
||
term_score += TERM_EXACT_SCORE
|
||
if _text_contains_exact_query(page_title_text, query):
|
||
page_title_score += PAGE_TITLE_EXACT_SCORE
|
||
if _text_contains_exact_query(body_text, query):
|
||
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(
|
||
intent=_intent_score(section, query),
|
||
heading_path=heading_score,
|
||
exact_term=term_score,
|
||
page_title=page_title_score,
|
||
body=body_score,
|
||
mode_boost=mode_boost,
|
||
synonym_hits=synonym_hits,
|
||
diagnose_priority=mode == "diagnose",
|
||
)
|
||
|
||
|
||
def _score_reasons(score: ScoreBreakdown) -> tuple[str, ...]:
|
||
components = (
|
||
("intent", score.intent),
|
||
("heading", score.heading_path),
|
||
("identifier", score.exact_term),
|
||
("page_title", score.page_title),
|
||
("body", score.body),
|
||
("mode", score.mode_boost),
|
||
)
|
||
reasons = tuple(f"{name}={value}" for name, value in components if value)
|
||
if score.synonym_hits:
|
||
reasons += (f"synonym={score.synonym_hits}",)
|
||
return reasons
|
||
|
||
|
||
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.local_body),
|
||
*FENCED_CODE_RE.findall(section.local_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
|
||
prelude = _write_prelude(all_sections) if mode == "write" else []
|
||
required_ids = {section.id for section in prelude}
|
||
raw_ascii_tokens = {
|
||
token
|
||
for token in _base_query_tokens(query)
|
||
if ASCII_TOKEN_RE.fullmatch(token)
|
||
and token not in ASCII_FILTER_STOP_TOKENS
|
||
}
|
||
ascii_tokens = {
|
||
token
|
||
for token in raw_ascii_tokens
|
||
if any(_ascii_anchor(section, {token}) for section in all_sections)
|
||
}
|
||
if ascii_tokens:
|
||
sections = [
|
||
section
|
||
for section in sections
|
||
if any(
|
||
_text_contains_token(section.searchable_text, token)
|
||
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,
|
||
_score_reasons(score),
|
||
)
|
||
)
|
||
ranked.sort(
|
||
key=lambda match: (
|
||
-match.score,
|
||
*(-value for value in match.priority),
|
||
match.section.page.as_posix(),
|
||
match.section.id,
|
||
)
|
||
)
|
||
# H2 聚合 section 的正文逐字包含其 H3 子节；父子同时入选时只保留
|
||
# 排名更高的一个，避免同一内容重复返回。
|
||
matches: list[QueryMatch] = []
|
||
page_counts: dict[Path, int] = {}
|
||
for match in ranked:
|
||
if match.section.id in required_ids:
|
||
continue
|
||
if any(
|
||
_related_sections(match.section, kept.section) for kept in matches
|
||
):
|
||
continue
|
||
if page_counts.get(match.section.page, 0) >= 2:
|
||
continue
|
||
matches.append(match)
|
||
page_counts[match.section.page] = page_counts.get(match.section.page, 0) + 1
|
||
if len(matches) == limit:
|
||
break
|
||
return QueryResult(
|
||
query=query,
|
||
mode=mode,
|
||
matches=matches,
|
||
prelude=prelude,
|
||
limit=limit,
|
||
)
|
||
|
||
|
||
def _logical_source(section: Section) -> str:
|
||
return f"references/{section.page.name}"
|
||
|
||
|
||
def _safe_json_string(value: str) -> str:
|
||
encoded = json.dumps(value, ensure_ascii=False)
|
||
for separator in ("\u0085", "\u2028", "\u2029"):
|
||
encoded = encoded.replace(separator, f"\\u{ord(separator):04x}")
|
||
return encoded
|
||
|
||
|
||
def _plain_text_summary(body: str, limit: int = 180) -> str:
|
||
without_fences = FENCED_CODE_RE.sub(" ", body)
|
||
without_links = re.sub(
|
||
r"!?\[([^\]]*)\]\([^)]+\)", lambda match: match.group(1), without_fences
|
||
)
|
||
without_inline_code = INLINE_CODE_RE.sub(lambda match: match.group(1), without_links)
|
||
content_lines = []
|
||
for line in without_inline_code.splitlines():
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith(("#", IDENTITY_PREFIX, BLOCK_DESCRIPTION_PREFIX)):
|
||
continue
|
||
content_lines.append(stripped.lstrip("-* "))
|
||
summary = re.sub(r"\s+", " ", " ".join(content_lines)).strip()
|
||
if len(summary) <= limit:
|
||
return summary
|
||
return summary[: limit - 1].rstrip() + "…"
|
||
|
||
|
||
def render_candidates(result: QueryResult) -> str:
|
||
lines = [
|
||
"# TSL Syntax Candidates",
|
||
"",
|
||
f"Mode: `{result.mode}`",
|
||
f"Query: {_safe_json_string(result.query)}",
|
||
]
|
||
candidates = [
|
||
(section, 0, True, ("required=1",)) for section in result.prelude
|
||
] + [
|
||
(match.section, match.score, False, match.reasons) for match in result.matches
|
||
]
|
||
for index, (section, score, required, reasons) in enumerate(
|
||
candidates[: result.limit], start=1
|
||
):
|
||
lines.extend(
|
||
[
|
||
"",
|
||
f"## Candidate {index}",
|
||
"",
|
||
f"Score: {score}",
|
||
f"Required: {'yes' if required else 'no'}",
|
||
f"Section ID: `{section.id}`",
|
||
f"Source: `{_logical_source(section)}`",
|
||
f"Heading: `{' > '.join(section.heading_path)}`",
|
||
f"Why: `{', '.join(reasons) or 'lexical=1'}`",
|
||
f"Summary: {_plain_text_summary(section.local_body)}",
|
||
]
|
||
)
|
||
return "\n".join(lines).rstrip() + "\n"
|
||
|
||
|
||
def render_section(section: Section) -> str:
|
||
lines = [
|
||
"# TSL Syntax Section",
|
||
"",
|
||
f"Section ID: `{section.id}`",
|
||
f"Source: `{_logical_source(section)}`",
|
||
"",
|
||
section.body.rstrip(),
|
||
]
|
||
return "\n".join(lines).rstrip() + "\n"
|
||
|
||
|
||
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 = _plain_text_summary(section.local_body, limit=600)
|
||
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 获取候选、用 --section 取回事实正文。",
|
||
"本清单不含可照写语法，也不替代精确章节取回。",
|
||
]
|
||
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
|
||
print(render_section(section), end="")
|
||
return 0
|
||
else:
|
||
result = query_sections(args.query, args.mode, args.limit, args.references_dir)
|
||
if not result.matches:
|
||
print(render_candidates(result), end="")
|
||
print("no matching sections", file=sys.stderr)
|
||
return 2
|
||
print(render_candidates(result), end="")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|