906 lines
32 KiB
Python
906 lines
32 KiB
Python
#!/usr/bin/env python3
|
||
"""Conservatively enrich TSL API descriptions and search tags.
|
||
|
||
Markdown is the fact source. This tool never changes API headings, signatures,
|
||
parameter tables, return types, examples, or scope placement. It only updates
|
||
the first prose description line and the optional ``<!-- tags: ... -->`` line.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import html
|
||
import re
|
||
import sys
|
||
import unicodedata
|
||
from collections import defaultdict
|
||
from dataclasses import dataclass, field, replace
|
||
from pathlib import Path
|
||
|
||
from bs4 import BeautifulSoup
|
||
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
if str(SCRIPT_DIR) not in sys.path:
|
||
sys.path.insert(0, str(SCRIPT_DIR))
|
||
|
||
from api_markdown import DECLARATION_LINE_RE, iter_api_entries
|
||
|
||
|
||
TAGS_RE = re.compile(r"^<!--\s*tags:\s*(.*?)\s*-->$")
|
||
ACCESS_RE = re.compile(r"^访问[:：]\s*(.*?)\s*$", re.IGNORECASE)
|
||
H1_RE = re.compile(r"^#(?!#)\s+(.+?)\s*$")
|
||
TERMINAL_PUNCTUATION_RE = re.compile(r"[。．.!！?？；;，,：:]+$")
|
||
MULTISPACE_RE = re.compile(r"\s+")
|
||
FORMULA_RE = re.compile(
|
||
r"^([^。；;，,]{2,40}?)\s*(?<![<>=!])=(?!=)\s*(.+)$"
|
||
)
|
||
COMPARISON_RE = re.compile(r"<=|>=|==|!=|<|>")
|
||
RELATED_FUNCTION_RE = re.compile(
|
||
r"^([A-Za-z][A-Za-z0-9_]*)\s*相关函数$"
|
||
)
|
||
ACRONYM_RE = re.compile(r"(?<![A-Za-z0-9_])([A-Z][A-Z0-9_]{1,15})(?![A-Za-z0-9_])")
|
||
ACTION_CLAUSE_RE = re.compile(
|
||
r"^(?:用于|如果|由于|若(?!干)|支持|查询|输入|按照|按(?!钮)|与系统参数|"
|
||
r"对.+?(?:进行|使用)|从.+?(?:获取|读取|计算|返回)|"
|
||
r"由.+?(?:倒推|获取|计算|得到|生成))"
|
||
)
|
||
|
||
PROHIBITED_PROCESS_RE = re.compile(
|
||
r"Source\s*ID|来源[:：]|已验证|验证通过|not[_ ]?found|"
|
||
r"测试过程|探测结果|官网定义",
|
||
re.IGNORECASE,
|
||
)
|
||
GENERIC_SUMMARIES = {
|
||
"函数",
|
||
"方法",
|
||
"属性",
|
||
"常量",
|
||
"接口",
|
||
"对象",
|
||
"类",
|
||
"内置函数",
|
||
"内部函数",
|
||
"统计分布相关函数",
|
||
}
|
||
GENERIC_TAGS = {
|
||
"函数",
|
||
"方法",
|
||
"属性",
|
||
"常量",
|
||
"数据",
|
||
"tsl",
|
||
"builtin",
|
||
"dotnet",
|
||
"third",
|
||
"deprecated",
|
||
"function",
|
||
"method",
|
||
"property",
|
||
}
|
||
CLASS_DESCRIPTIONS = {
|
||
"imap": "用于连接 IMAP 服务器、管理邮箱并读取邮件的内置对象",
|
||
"mailmsg": "用于创建、解析、保存和发送邮件消息的内置对象",
|
||
"messagepart": "表示邮件正文、附件等消息分段的内置对象",
|
||
"pop3": "用于连接 POP3 服务器并读取或删除邮件的内置对象",
|
||
"smtp": "用于连接 SMTP 服务器并发送邮件的内置对象",
|
||
"tcipher": "提供对称加密和解密能力的内置对象",
|
||
"tcookie": "表示 HTTP Cookie 名称、值和属性的内置对象",
|
||
"tcookiecollection": "用于创建、查找和管理 HTTP Cookie 集合的内置对象",
|
||
"tfilestream": "以文件为存储后端并提供流式读写能力的内置对象",
|
||
"thandlestream": "封装系统句柄并提供流式读写能力的内置对象",
|
||
"thashedstringlist": "使用哈希索引加速查找的字符串列表内置对象",
|
||
"tinifile": "用于读取和写入 INI 配置文件的内置对象",
|
||
"tmeminifile": "在内存中读取、修改和保存 INI 配置的内置对象",
|
||
"tmemorystream": "以内存缓冲区保存和操作流数据的内置对象",
|
||
"tregistryinifile": "以系统注册表为存储后端的 INI 配置内置对象",
|
||
"trsa": "提供 RSA 非对称加密、解密和签名能力的内置对象",
|
||
"tsession": "表示服务端会话及其键值数据的内置对象",
|
||
"tsessionman": "用于创建、查找、维护和删除服务端会话的内置对象",
|
||
"tstream": "定义顺序读写、定位和复制操作的流基类",
|
||
"tstringlist": "提供字符串集合存储、查找、排序和名称值管理能力的内置对象",
|
||
"twebrequest": "封装 HTTP 请求头、请求参数、Cookie 和请求正文的内置对象",
|
||
"twebresponse": "用于设置 HTTP 状态、响应头、Cookie 和响应正文的内置对象",
|
||
}
|
||
GENERIC_DESCRIPTION_OVERRIDES = {
|
||
"unicodeesc2": "将字符串转换为 \\uxxxx 形式的 Unicode 编码串",
|
||
"sf_x2inv": "根据累计概率计算卡方分布的逆函数值",
|
||
"sf_tinv": "根据累计概率计算 t 分布的逆函数值",
|
||
"sf_sinx": "计算正弦积分",
|
||
"sf_normalinv2": "根据累计概率计算标准正态分布的逆函数值",
|
||
"sf_normalinv": "根据累计概率计算正态分布的逆函数值",
|
||
"sf_normal": "计算正态分布函数值",
|
||
"sf_gammainv": "根据累计概率计算标准 Gamma 分布的逆函数值",
|
||
"sf_gamma": "计算伽马函数值",
|
||
"sf_finv": "根据累计概率计算 F 分布的逆函数值",
|
||
"sf_ex": "计算指数积分",
|
||
"sf_error": "计算误差函数（概率函数）值",
|
||
"sf_cosx": "计算余弦积分",
|
||
"sf_betainv": "根据累计概率计算贝塔分布的逆函数值",
|
||
"sf_beta": "计算贝塔函数值",
|
||
"readfile": "读取本地文件中的数据",
|
||
"millisecondspan": "计算两个日期时间之间相差的毫秒数",
|
||
"importfile": "从文件中导入数据",
|
||
"getmsgdigest": "计算字符串或文件的摘要并返回摘要字符串",
|
||
"exportfile": "将数据导出到文件",
|
||
}
|
||
AWKWARD_DESCRIPTION_OVERRIDES = {
|
||
"encodegraph": "根据指定的类型、名称、数据和属性数组生成图形并返回",
|
||
"slndepreciation": (
|
||
"按直线折旧法计算每期折旧额，计算公式为"
|
||
"（资产原值 Cost - 资产残值 Salvage）/ 折旧周期数 Life"
|
||
),
|
||
"format": "根据 formatstr 指定的格式将参数连接成字符串并返回",
|
||
"ansiformat": "根据 formatstr 指定的格式将参数连接成字符串并返回",
|
||
"dupestring": "将字符串 AText 重复 ACount 次并返回结果",
|
||
"stockpjcj": "返回区间平均成交价，计算公式为区间成交金额 / 区间成交量",
|
||
}
|
||
LEADING_VERBS = (
|
||
"返回",
|
||
"获取",
|
||
"读取",
|
||
"写入",
|
||
"设置",
|
||
"创建",
|
||
"删除",
|
||
"移除",
|
||
"清除",
|
||
"添加",
|
||
"插入",
|
||
"查找",
|
||
"搜索",
|
||
"判断",
|
||
"检查",
|
||
"验证",
|
||
"计算",
|
||
"转换",
|
||
"解析",
|
||
"格式化",
|
||
"比较",
|
||
"执行",
|
||
"调用",
|
||
"连接",
|
||
"关闭",
|
||
"打开",
|
||
"发送",
|
||
"接收",
|
||
"保存",
|
||
"加载",
|
||
"列出",
|
||
"表示",
|
||
"提供",
|
||
"释放",
|
||
"交换",
|
||
"排序",
|
||
"过滤",
|
||
"统计",
|
||
"控制",
|
||
"指示",
|
||
"提取",
|
||
"根据",
|
||
"将",
|
||
"把",
|
||
"生成",
|
||
"构造",
|
||
"采用",
|
||
"利用",
|
||
"输出",
|
||
"得到",
|
||
"获得",
|
||
"取得",
|
||
"新建",
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SourceRecord:
|
||
corpus: str
|
||
page_id: str
|
||
title: str
|
||
page: Path
|
||
description: str = ""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class EntryContext:
|
||
name: str
|
||
signature: str
|
||
kind: str
|
||
scope: str
|
||
page: str
|
||
summary: str
|
||
access: str = ""
|
||
owner: str = ""
|
||
tags: tuple[str, ...] = ()
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PageContext:
|
||
title: str
|
||
path: str
|
||
|
||
|
||
@dataclass
|
||
class AuditChange:
|
||
page: str
|
||
name: str
|
||
kind: str
|
||
old_description: str
|
||
new_description: str
|
||
old_tags: tuple[str, ...]
|
||
new_tags: tuple[str, ...]
|
||
sources: tuple[str, ...]
|
||
|
||
|
||
@dataclass
|
||
class Audit:
|
||
entries: int = 0
|
||
source_matches: int = 0
|
||
ambiguous_sources: int = 0
|
||
changed_descriptions: int = 0
|
||
changed_tags: int = 0
|
||
changes: list[AuditChange] = field(default_factory=list)
|
||
|
||
def merge(self, other: "Audit") -> None:
|
||
self.entries += other.entries
|
||
self.source_matches += other.source_matches
|
||
self.ambiguous_sources += other.ambiguous_sources
|
||
self.changed_descriptions += other.changed_descriptions
|
||
self.changed_tags += other.changed_tags
|
||
self.changes.extend(other.changes)
|
||
|
||
|
||
def normalize_key(value: str) -> str:
|
||
return unicodedata.normalize("NFKC", value).casefold().strip()
|
||
|
||
|
||
def clean_text(value: str) -> str:
|
||
value = html.unescape(value).replace("\xa0", " ")
|
||
value = value.replace("“", '"').replace("”", '"')
|
||
value = value.replace("‘", "'").replace("’", "'")
|
||
return MULTISPACE_RE.sub(" ", value).strip()
|
||
|
||
|
||
def strip_terminal_punctuation(value: str) -> str:
|
||
return TERMINAL_PUNCTUATION_RE.sub("", clean_text(value)).rstrip()
|
||
|
||
|
||
def safe_source_description(value: str) -> str:
|
||
value = strip_terminal_punctuation(value)
|
||
if not value or PROHIBITED_PROCESS_RE.search(value):
|
||
return ""
|
||
return value
|
||
|
||
|
||
def _extract_source_description(page: Path) -> str:
|
||
try:
|
||
soup = BeautifulSoup(page.read_text(encoding="utf-8"), "html.parser")
|
||
except (OSError, UnicodeError):
|
||
return ""
|
||
root = soup.select_one("#help_content") or soup
|
||
for expected in ("简述", "说明"):
|
||
for label in root.select(".DescriteTitle"):
|
||
if clean_text(label.get_text(" ", strip=True)) != expected:
|
||
continue
|
||
marker = label.parent if label.parent is not None else label
|
||
sibling = marker.find_next_sibling()
|
||
if sibling is None:
|
||
sibling = label.find_next_sibling()
|
||
if sibling is not None:
|
||
return safe_source_description(sibling.get_text(" ", strip=True))
|
||
return ""
|
||
|
||
|
||
class SourceCatalog:
|
||
PREFERENCES = {
|
||
"builtin": ("tsl_base", "net_function", "knowledge_base"),
|
||
"third": ("net_function", "tsl_base", "knowledge_base"),
|
||
"deprecated": ("net_function", "tsl_base", "knowledge_base"),
|
||
"dotnet": ("net_function", "tsl_base", "knowledge_base"),
|
||
}
|
||
|
||
def __init__(self, records: dict[str, list[SourceRecord]] | None = None):
|
||
self.records = records or {}
|
||
self._description_cache: dict[Path, str] = {}
|
||
|
||
@classmethod
|
||
def empty(cls) -> "SourceCatalog":
|
||
return cls({})
|
||
|
||
@classmethod
|
||
def from_docs_root(cls, docs_root: Path) -> "SourceCatalog":
|
||
records: dict[str, list[SourceRecord]] = defaultdict(list)
|
||
for corpus in ("net_function", "tsl_base", "knowledge_base"):
|
||
root = docs_root / corpus
|
||
manifest = root / "manifest.tsv"
|
||
if not manifest.is_file():
|
||
continue
|
||
with manifest.open(encoding="utf-8", newline="") as handle:
|
||
for row in csv.DictReader(handle, delimiter="\t"):
|
||
if row.get("status") != "ok" or not row.get("title"):
|
||
continue
|
||
record = SourceRecord(
|
||
corpus=corpus,
|
||
page_id=row.get("id", ""),
|
||
title=row["title"],
|
||
page=root / row.get("path", ""),
|
||
)
|
||
records[normalize_key(record.title)].append(record)
|
||
return cls(dict(records))
|
||
|
||
def _load_description(self, record: SourceRecord) -> SourceRecord:
|
||
if record.description:
|
||
return record
|
||
if record.page not in self._description_cache:
|
||
self._description_cache[record.page] = _extract_source_description(
|
||
record.page
|
||
)
|
||
return replace(
|
||
record, description=self._description_cache[record.page]
|
||
)
|
||
|
||
def match(self, entry: EntryContext) -> list[SourceRecord]:
|
||
matches = self.records.get(normalize_key(entry.name), [])
|
||
preference = self.PREFERENCES.get(
|
||
entry.scope,
|
||
("net_function", "tsl_base", "knowledge_base"),
|
||
)
|
||
rank = {name: index for index, name in enumerate(preference)}
|
||
|
||
def page_order(record: SourceRecord):
|
||
try:
|
||
page_id = (0, int(record.page_id))
|
||
except ValueError:
|
||
page_id = (1, record.page_id)
|
||
return rank.get(record.corpus, 99), page_id
|
||
|
||
return sorted(matches, key=page_order)
|
||
|
||
def describe(self, records: list[SourceRecord]) -> list[SourceRecord]:
|
||
return [self._load_description(record) for record in records]
|
||
|
||
|
||
def _starts_with_verb(value: str) -> bool:
|
||
return value.startswith(LEADING_VERBS) or bool(
|
||
ACTION_CLAUSE_RE.match(value)
|
||
)
|
||
|
||
|
||
def _choose_source_description(
|
||
current: str, sources: list[SourceRecord]
|
||
) -> str:
|
||
candidates = [
|
||
safe_source_description(record.description) for record in sources
|
||
]
|
||
candidates = [value for value in candidates if value]
|
||
if not candidates:
|
||
return current
|
||
if len(sources) == 1 and current in GENERIC_SUMMARIES:
|
||
candidate = candidates[0]
|
||
if candidate not in GENERIC_SUMMARIES and len(candidate) > len(current):
|
||
return candidate
|
||
if PROHIBITED_PROCESS_RE.search(current):
|
||
return candidates[0]
|
||
return current
|
||
|
||
|
||
def _property_description(value: str, access: str) -> str:
|
||
if _starts_with_verb(value):
|
||
return value
|
||
if (
|
||
len(value) > 30
|
||
or any(mark in value for mark in "。；，")
|
||
or value.startswith(("功能同", "只读", "用于"))
|
||
):
|
||
return value
|
||
normalized = normalize_key(access).replace(" ", "")
|
||
if value.startswith("是否"):
|
||
return (
|
||
f"控制{value}"
|
||
if "write" in normalized
|
||
else f"指示{value}"
|
||
)
|
||
if "read" in normalized and "write" in normalized:
|
||
return f"获取或设置{value}"
|
||
if "write" in normalized:
|
||
return f"设置{value}"
|
||
return f"获取{value}"
|
||
|
||
|
||
def _name_intent(name: str) -> str:
|
||
key = normalize_key(name)
|
||
rules = (
|
||
(("is", "if", "has", "can", "check", "valid"), "判断"),
|
||
(("get", "find", "query", "search"), "获取"),
|
||
(("read", "load"), "读取"),
|
||
(("write", "save", "export"), "写入"),
|
||
(("set",), "设置"),
|
||
(("create", "make", "new"), "创建"),
|
||
(("delete", "remove", "drop"), "删除"),
|
||
(("clear", "reset"), "清除"),
|
||
(("add", "append", "insert"), "添加"),
|
||
(("parse", "decode"), "解析"),
|
||
(("encode", "convert", "to"), "转换"),
|
||
(("format",), "格式化"),
|
||
(("calc", "compute"), "计算"),
|
||
(("open",), "打开"),
|
||
(("close",), "关闭"),
|
||
(("send", "post"), "发送"),
|
||
(("connect", "login"), "连接"),
|
||
(("list",), "列出"),
|
||
)
|
||
for prefixes, intent in rules:
|
||
if key.startswith(prefixes):
|
||
return intent
|
||
return ""
|
||
|
||
|
||
def improve_description(
|
||
entry: EntryContext, sources: list[SourceRecord]
|
||
) -> str:
|
||
current = strip_terminal_punctuation(entry.summary)
|
||
current = _choose_source_description(current, sources)
|
||
current = strip_terminal_punctuation(current)
|
||
if not current:
|
||
return entry.summary
|
||
|
||
if entry.kind == "class":
|
||
replacement = CLASS_DESCRIPTIONS.get(normalize_key(entry.name))
|
||
if replacement and re.fullmatch(
|
||
rf"{re.escape(entry.name)}\s*内置对象", current, re.IGNORECASE
|
||
):
|
||
return replacement
|
||
|
||
if entry.kind == "property":
|
||
return _property_description(current, entry.access)
|
||
|
||
key = normalize_key(entry.name)
|
||
generic_override = GENERIC_DESCRIPTION_OVERRIDES.get(key)
|
||
if generic_override and current.endswith("相关函数"):
|
||
return generic_override
|
||
awkward_override = AWKWARD_DESCRIPTION_OVERRIDES.get(key)
|
||
if awkward_override and current.startswith(
|
||
("返回根据", "返回将", "返回:", "返回：")
|
||
):
|
||
return awkward_override
|
||
|
||
current = re.sub(
|
||
r"^(?:该|本)函数(?:主要(?:是)?)?\s*", "", current
|
||
)
|
||
current = re.sub(r"^功能[:：]\s*", "", current)
|
||
if current.startswith("是否"):
|
||
current = f"判断{current}"
|
||
|
||
for old, new in (
|
||
("读出", "读取"),
|
||
("取出", "获取"),
|
||
("得到", "获取"),
|
||
("获得", "获取"),
|
||
("取得", "获取"),
|
||
("新建", "创建"),
|
||
):
|
||
if current.startswith(old):
|
||
current = new + current[len(old):]
|
||
break
|
||
if current.startswith("读") and not current.startswith("读取"):
|
||
current = "读取" + current[1:]
|
||
if current.startswith("写") and not current.startswith("写入"):
|
||
current = "写入" + current[1:]
|
||
current = current.replace("读出", "读取")
|
||
current = current.replace("装载内容", "加载内容")
|
||
|
||
related_function = RELATED_FUNCTION_RE.fullmatch(current)
|
||
if related_function:
|
||
return f"返回 {related_function.group(1)} 对应的数据"
|
||
|
||
formula = FORMULA_RE.match(current)
|
||
page_key = normalize_key(entry.page)
|
||
if (
|
||
formula
|
||
and ("/financial/" in page_key or "/financial_report/" in page_key)
|
||
and not _starts_with_verb(current)
|
||
):
|
||
left, right = formula.groups()
|
||
return f"计算{left.strip()}，公式为{right.strip()}"
|
||
|
||
if _starts_with_verb(current):
|
||
return current
|
||
|
||
if "/financial/" in page_key or "/financial_report/" in page_key:
|
||
if (
|
||
re.match(r"^\d+(?:\s|[.．、)])", current)
|
||
or COMPARISON_RE.search(current)
|
||
):
|
||
return current
|
||
return f"返回{current}"
|
||
|
||
intent = _name_intent(entry.name)
|
||
if intent and len(current) <= 28 and not any(
|
||
verb in current[:16] for verb in LEADING_VERBS
|
||
):
|
||
return f"{intent}{current}"
|
||
return current
|
||
|
||
|
||
DOMAIN_RULES = (
|
||
(r"数组|列表|(?:^|[^a-z])(?:array|fmarray|list)(?:$|[^a-z])", ("数组", "列表", "array", "list")),
|
||
(r"矩阵|(?:^|[^a-z])matrix(?:$|[^a-z])|\bmt_", ("矩阵", "matrix")),
|
||
(r"字符串|文本|(?:^|[^a-z])(?:string|char|text)(?:$|[^a-z])", ("字符串", "文本", "string", "text")),
|
||
(r"日期|时间|(?:^|[^a-z])(?:datetime|date|time)(?:$|[^a-z])", ("日期时间", "日期", "时间", "datetime")),
|
||
(r"文件|目录|路径|(?:^|[^a-z])(?:file|folder|directory|path)(?:$|[^a-z])", ("文件", "目录", "路径", "file")),
|
||
(r"网络|(?:^|[^a-z])(?:http|https|cgi|cookie|url)(?:$|[^a-z])", ("网络", "HTTP", "请求", "network")),
|
||
(r"缓存|(?:^|[^a-z])cache(?:$|[^a-z])", ("缓存", "缓存管理", "cache")),
|
||
(r"数据库|(?:^|[^a-z])(?:sql|dbf|ini)(?:$|[^a-z])", ("数据库", "SQL", "database")),
|
||
(r"进程|线程|(?:^|[^a-z])(?:process|thread|pipe)(?:$|[^a-z])", ("进程", "线程", "process")),
|
||
(r"颜色|(?:^|[^a-z])(?:color|rgb|cmyk)(?:$|[^a-z])", ("颜色", "RGB", "color")),
|
||
(r"图形|图表|(?:^|[^a-z])(?:graph|chart)(?:$|[^a-z])", ("图形", "图表", "graph")),
|
||
(r"统计|概率|分布|(?:^|[^a-z])(?:statistics|cdf|pdf)(?:$|[^a-z])", ("统计", "概率", "statistics")),
|
||
(r"优化|线性规划|(?:^|[^a-z])optimization(?:$|[^a-z])", ("优化", "求解", "optimization")),
|
||
(r"数值|数学|(?:^|[^a-z])(?:numeric|math)(?:$|[^a-z])", ("数学", "数值", "numeric")),
|
||
(r"类型转换|(?:^|[^a-z])(?:convert|conversion)(?:$|[^a-z])", ("类型转换", "转换", "conversion")),
|
||
(r"对象|(?:^|[^a-z])(?:object|class)(?:$|[^a-z])", ("对象", "实例", "object")),
|
||
(r"(?:^|[^a-z])(?:com|ole)(?:$|[^a-z])|activex", ("COM", "OLE", "自动化")),
|
||
(r"(?:^|[^a-z])ftp(?:$|[^a-z])", ("FTP", "文件传输")),
|
||
(r"邮件|(?:^|[^a-z])(?:smtp|pop3|imap|mail)(?:$|[^a-z])", ("邮件", "SMTP", "POP3", "IMAP")),
|
||
(r"财务|金融|证券|股票|基金|债券|期货|期权", ("金融",)),
|
||
(r"报表|报告期|report", ("报表", "报告期")),
|
||
)
|
||
|
||
SPECIAL_TAG_RULES = (
|
||
(r"去重|删除重复|dropduplicate|dedup", ("去重", "删除重复", "deduplicate")),
|
||
(r"哈希索引|哈希表", ("哈希索引", "哈希表", "hash")),
|
||
(
|
||
r"摘要|(?:^|[^a-z0-9_])(?:crc32|md5|sha1|sha224|sha256|sha384|sha512|sm3|digest|hash)(?:$|[^a-z0-9_])",
|
||
("摘要", "哈希", "散列", "digest", "hash"),
|
||
),
|
||
(r"排序|sort", ("排序", "sort")),
|
||
(r"过滤|筛选|filter", ("过滤", "筛选", "filter")),
|
||
(r"查找|搜索|find|search|lookup", ("查找", "搜索", "lookup")),
|
||
)
|
||
|
||
INTENT_TAGS = {
|
||
"返回": ("返回", "获取"),
|
||
"获取": ("获取", "查询"),
|
||
"读取": ("读取", "获取"),
|
||
"写入": ("写入", "保存"),
|
||
"设置": ("设置", "修改"),
|
||
"创建": ("创建", "生成"),
|
||
"删除": ("删除", "移除"),
|
||
"清除": ("清除", "重置"),
|
||
"添加": ("添加", "追加"),
|
||
"查找": ("查找", "搜索"),
|
||
"判断": ("判断", "检查"),
|
||
"计算": ("计算", "求解"),
|
||
"转换": ("转换", "编码"),
|
||
"解析": ("解析", "parse"),
|
||
"格式化": ("格式化", "format"),
|
||
"打开": ("打开", "open"),
|
||
"关闭": ("关闭", "close"),
|
||
"发送": ("发送", "提交"),
|
||
"连接": ("连接", "登录"),
|
||
"列出": ("列出", "列表"),
|
||
"输出": ("输出", "打印"),
|
||
}
|
||
|
||
|
||
def _taxonomy_tags(title: str) -> list[str]:
|
||
result = []
|
||
for raw in re.split(r"\s*/\s*|\s+-\s+", title):
|
||
value = raw.strip(" #")
|
||
value = re.sub(r"\([^)]*\)|（[^）]*）", "", value).strip()
|
||
for suffix in ("相关函数", "及其实现", "相关"):
|
||
if value.endswith(suffix):
|
||
value = value[: -len(suffix)].rstrip()
|
||
if not value or normalize_key(value) in GENERIC_TAGS:
|
||
continue
|
||
if len(value) > 18:
|
||
continue
|
||
chinese = "".join(re.findall(r"[\u3400-\u9fff]+", value))
|
||
latin = re.sub(r"[\u3400-\u9fff]+", " ", value).strip()
|
||
if latin and normalize_key(latin) not in GENERIC_TAGS:
|
||
result.extend(
|
||
re.findall(r"[A-Za-z][A-Za-z0-9_+.-]*", latin)
|
||
)
|
||
if chinese:
|
||
result.append(chinese)
|
||
return result
|
||
|
||
|
||
def _description_intent(summary: str) -> str:
|
||
value = strip_terminal_punctuation(summary)
|
||
replacements = {
|
||
"得到": "获取",
|
||
"获得": "获取",
|
||
"取得": "获取",
|
||
"新建": "创建",
|
||
}
|
||
for source, target in replacements.items():
|
||
if value.startswith(source):
|
||
return target
|
||
for intent in INTENT_TAGS:
|
||
if value.startswith(intent):
|
||
return intent
|
||
for intent in ("读取", "写入", "保存", "计算", "判断", "转换", "解析"):
|
||
if intent in value[:16]:
|
||
return "写入" if intent == "保存" else intent
|
||
return ""
|
||
|
||
|
||
def derive_tags(entry: EntryContext, page: PageContext) -> list[str]:
|
||
tags: list[str] = []
|
||
seen: set[str] = set()
|
||
|
||
def add(value: str) -> None:
|
||
value = clean_text(value).strip()
|
||
key = normalize_key(value)
|
||
if (
|
||
not value
|
||
or key in seen
|
||
or key in GENERIC_TAGS
|
||
or len(tags) >= 12
|
||
):
|
||
return
|
||
seen.add(key)
|
||
tags.append(value)
|
||
|
||
for tag in entry.tags:
|
||
add(tag)
|
||
|
||
for tag in _taxonomy_tags(page.title):
|
||
add(tag)
|
||
|
||
haystack = " ".join(
|
||
(entry.name, entry.signature, entry.summary, page.title, page.path)
|
||
)
|
||
domain_haystack = " ".join((page.title, page.path))
|
||
for pattern, aliases in SPECIAL_TAG_RULES:
|
||
if re.search(pattern, haystack, re.IGNORECASE):
|
||
for alias in aliases:
|
||
add(alias)
|
||
|
||
summary = strip_terminal_punctuation(entry.summary)
|
||
intent = _description_intent(summary)
|
||
if not intent and not _starts_with_verb(summary):
|
||
is_market_price = (
|
||
normalize_key(entry.name) in {"open", "close"}
|
||
and "盘价" in summary
|
||
)
|
||
if not is_market_price:
|
||
intent = _name_intent(entry.name)
|
||
for alias in INTENT_TAGS.get(intent, ()):
|
||
add(alias)
|
||
for pattern, aliases in DOMAIN_RULES:
|
||
if re.search(pattern, domain_haystack, re.IGNORECASE):
|
||
for alias in aliases:
|
||
add(alias)
|
||
for acronym in ACRONYM_RE.findall(entry.summary):
|
||
add(acronym)
|
||
return tags
|
||
|
||
|
||
def _scope_for_page(page: str) -> str:
|
||
parts = Path(page).parts
|
||
return parts[0] if len(parts) > 1 else Path(page).stem
|
||
|
||
|
||
def _page_title(lines: list[str], page: str) -> PageContext:
|
||
for line in lines:
|
||
match = H1_RE.match(line)
|
||
if match:
|
||
return PageContext(match.group(1), page)
|
||
return PageContext(Path(page).stem, page)
|
||
|
||
|
||
def _entry_metadata(lines: list[str], start: int, end: int):
|
||
declaration_seen = False
|
||
description_index = None
|
||
tag_index = None
|
||
access = ""
|
||
for index in range(start + 1, end):
|
||
text = lines[index].strip()
|
||
if not text:
|
||
continue
|
||
if not declaration_seen:
|
||
if DECLARATION_LINE_RE.fullmatch(text):
|
||
declaration_seen = True
|
||
continue
|
||
tag_match = TAGS_RE.fullmatch(text)
|
||
if tag_match:
|
||
tag_index = index
|
||
continue
|
||
access_match = ACCESS_RE.fullmatch(text)
|
||
if access_match:
|
||
access = access_match.group(1)
|
||
continue
|
||
if description_index is None:
|
||
if text.startswith(("|", "#", "返回：", "类型：", "可见性：", "值：")):
|
||
continue
|
||
description_index = index
|
||
tags = ()
|
||
if tag_index is not None:
|
||
tags = tuple(TAGS_RE.fullmatch(lines[tag_index].strip()).group(1).split())
|
||
return description_index, tag_index, access, tags
|
||
|
||
|
||
def enrich_markdown(
|
||
text: str, page: str, source_catalog: SourceCatalog
|
||
) -> tuple[str, Audit]:
|
||
had_final_newline = text.endswith("\n")
|
||
lines = text.splitlines()
|
||
page_context = _page_title(lines, page)
|
||
entries = [entry for entry in iter_api_entries(lines) if entry.heading.valid]
|
||
audit = Audit(entries=len(entries))
|
||
root_owner = ""
|
||
owners: dict[int, str] = {}
|
||
for entry in entries:
|
||
if entry.heading.level == 2:
|
||
root_owner = entry.heading.name if entry.heading.kind in {"class", "unit"} else ""
|
||
owners[entry.start] = root_owner if entry.heading.level > 2 else ""
|
||
|
||
for api in reversed(entries):
|
||
description_index, tag_index, access, existing_tags = _entry_metadata(
|
||
lines, api.start, api.end
|
||
)
|
||
if description_index is None:
|
||
continue
|
||
old_description = lines[description_index].strip()
|
||
entry = EntryContext(
|
||
name=api.heading.name,
|
||
signature=api.heading.signature,
|
||
kind=api.heading.kind,
|
||
scope=_scope_for_page(page),
|
||
page=page,
|
||
summary=old_description,
|
||
access=access,
|
||
owner=owners.get(api.start, ""),
|
||
tags=existing_tags,
|
||
)
|
||
sources = source_catalog.match(entry)
|
||
if sources:
|
||
audit.source_matches += 1
|
||
if len(sources) > 1:
|
||
audit.ambiguous_sources += 1
|
||
evidence = sources
|
||
normalized_summary = strip_terminal_punctuation(old_description)
|
||
if (
|
||
normalized_summary in GENERIC_SUMMARIES
|
||
or PROHIBITED_PROCESS_RE.search(normalized_summary)
|
||
):
|
||
evidence = source_catalog.describe(sources)
|
||
new_description = improve_description(entry, evidence)
|
||
entry_for_tags = replace(entry, summary=new_description)
|
||
new_tags = tuple(derive_tags(entry_for_tags, page_context))
|
||
|
||
if new_description != old_description:
|
||
lines[description_index] = new_description
|
||
audit.changed_descriptions += 1
|
||
if new_tags != existing_tags:
|
||
rendered = f"<!-- tags: {' '.join(new_tags)} -->"
|
||
if tag_index is not None:
|
||
lines[tag_index] = rendered
|
||
else:
|
||
insert_at = description_index + 1
|
||
if insert_at < len(lines) and lines[insert_at].strip() == "":
|
||
insert_at += 1
|
||
lines[insert_at:insert_at] = [rendered, ""]
|
||
else:
|
||
lines[insert_at:insert_at] = ["", rendered, ""]
|
||
audit.changed_tags += 1
|
||
|
||
if new_description != old_description or new_tags != existing_tags:
|
||
audit.changes.append(
|
||
AuditChange(
|
||
page=page,
|
||
name=api.heading.name,
|
||
kind=api.heading.kind,
|
||
old_description=old_description,
|
||
new_description=new_description,
|
||
old_tags=existing_tags,
|
||
new_tags=new_tags,
|
||
sources=tuple(
|
||
f"{record.corpus}:{record.page_id}" for record in sources
|
||
),
|
||
)
|
||
)
|
||
|
||
result = "\n".join(lines)
|
||
if had_final_newline:
|
||
result += "\n"
|
||
return result, audit
|
||
|
||
|
||
def _write_report(path: Path, audit: Audit) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with path.open("w", encoding="utf-8", newline="") as handle:
|
||
writer = csv.writer(handle, delimiter="\t", lineterminator="\n")
|
||
writer.writerow(
|
||
(
|
||
"page",
|
||
"name",
|
||
"kind",
|
||
"old_description",
|
||
"new_description",
|
||
"old_tags",
|
||
"new_tags",
|
||
"sources",
|
||
)
|
||
)
|
||
for item in audit.changes:
|
||
writer.writerow(
|
||
(
|
||
item.page,
|
||
item.name,
|
||
item.kind,
|
||
item.old_description,
|
||
item.new_description,
|
||
" ".join(item.old_tags),
|
||
" ".join(item.new_tags),
|
||
" ".join(item.sources),
|
||
)
|
||
)
|
||
|
||
|
||
def _selected(page: str, scopes: set[str], prefixes: tuple[str, ...]) -> bool:
|
||
if scopes and _scope_for_page(page) not in scopes:
|
||
return False
|
||
return not prefixes or page.startswith(prefixes)
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
mode = parser.add_mutually_exclusive_group(required=True)
|
||
mode.add_argument("--audit", action="store_true", help="report changes only")
|
||
mode.add_argument("--apply", action="store_true", help="write proposed changes")
|
||
parser.add_argument("--root", type=Path, required=True, help="codegen Markdown root")
|
||
parser.add_argument(
|
||
"--docs-root", type=Path, required=True, help="cached tmp/docs root"
|
||
)
|
||
parser.add_argument("--report", type=Path, help="optional TSV audit report")
|
||
parser.add_argument("--scope", action="append", default=[])
|
||
parser.add_argument("--prefix", action="append", default=[])
|
||
args = parser.parse_args(argv)
|
||
|
||
catalog = SourceCatalog.from_docs_root(args.docs_root)
|
||
aggregate = Audit()
|
||
changed_files = 0
|
||
for md in sorted(args.root.rglob("*.md")):
|
||
page = md.relative_to(args.root).as_posix()
|
||
if not _selected(page, set(args.scope), tuple(args.prefix)):
|
||
continue
|
||
before_stat = md.stat()
|
||
original = md.read_text(encoding="utf-8")
|
||
enriched, audit = enrich_markdown(original, page, catalog)
|
||
aggregate.merge(audit)
|
||
if enriched == original:
|
||
continue
|
||
changed_files += 1
|
||
if args.apply:
|
||
after_stat = md.stat()
|
||
if (
|
||
after_stat.st_mtime_ns != before_stat.st_mtime_ns
|
||
or after_stat.st_size != before_stat.st_size
|
||
):
|
||
raise RuntimeError(f"target changed while processing: {md}")
|
||
md.write_text(enriched, encoding="utf-8", newline="\n")
|
||
|
||
if args.report:
|
||
_write_report(args.report, aggregate)
|
||
print(
|
||
"entries={entries} source_matches={source_matches} "
|
||
"ambiguous_sources={ambiguous_sources} changed_files={changed_files} "
|
||
"changed_descriptions={changed_descriptions} changed_tags={changed_tags}".format(
|
||
entries=aggregate.entries,
|
||
source_matches=aggregate.source_matches,
|
||
ambiguous_sources=aggregate.ambiguous_sources,
|
||
changed_files=changed_files,
|
||
changed_descriptions=aggregate.changed_descriptions,
|
||
changed_tags=aggregate.changed_tags,
|
||
)
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|