📦 deps(tsl): sync tsl-playbook from ecedf69b

Source-Commit: ecedf69b9c
This commit is contained in:
ci[bot]
2026-08-24 09:29:51 +08:00
parent 2f0d80da65
commit 2b74e1466d
664 changed files with 28934 additions and 37020 deletions
+117
View File
@@ -0,0 +1,117 @@
# TSL API 数据维护
本文档面向本仓库维护者,不是安装后 API skill 的运行说明。命令均从仓库根目录执行。
## 维护 API 页面和索引
1. 按 [`STANDARD.md`](STANDARD.md) 修改或生成
`skills/tsl-api-reference/references/codegen/` 下的页面。
2. 对修改过的页面运行 lint;确认页面中的每个 `page#anchor` 都能定位。
3. 重建并检查统一索引:
```bash
python tools/tsl-codegen/scripts/build_index.py \
--skill-dir skills/tsl-api-reference
python tools/tsl-codegen/scripts/build_index.py \
--skill-dir skills/tsl-api-reference --check
```
4. 用 `lookup.py --name` 和 `class_lookup.py --class` 做一次精确查询,确认名称、归属、
成员和来源没有漂移。
## 维护 class:先判断是否真的需要 Framework Profile
`function_index.tsv` 中 `kind=class` 的每个 class 都会自动获得成员视图。普通 class 没有
Framework Profile 是正常且完整的状态,不要为了让它出现在列表里创建空 profile:
```bash
python skills/tsl-api-reference/scripts/class_lookup.py \
--class TStringList --include-members --format json
```
看到 `profile_status=not_profiled` 时,继续维护 class 页面、成员正文和索引即可。配置、
生命周期、回调或执行阶段不能从普通 class 成员列表臆造出来。
只有在参考页面提供了可核验的生命周期协议,并且 class 确实需要配置字段、回调、状态、
执行入口或结果查询时,才在
`skills/tsl-api-reference/data/framework_index.json` 增加 `kind=framework` 的条目。
Profile 必须为每个引用提供对应的 `page#anchor` 证据,并通过:
```bash
python skills/tsl-api-reference/scripts/class_lookup.py --check
```
如果证据不足,保留 `profile_status=not_profiled`,不要用猜测补齐 framework 生命周期。
Profile 的 `contract_status` 不是 `resolved` 时,也不能生成依赖未证实回调字段的代码。
### 用户或临时索引中的 class
维护自定义 `function_index.tsv` 时,可直接查询普通 class;只传 `--tsv` 不加载任何
curated Framework Profile,这不是错误:
```bash
python <skill-dir>/scripts/class_lookup.py \
--tsv <index-root>/data/function_index.tsv \
--class MyClass --include-members --format json
```
若确实维护了自定义 profile,显式传入 `--profiles <path>`,再运行 `--check` 验证引用。
## 维护数据字典快照
字典生成器属于本工具包,位于 `tools/tsl-codegen/scripts/build_dictionary.py`,不随
`tsl-api-reference` skill 安装。它依赖 `beautifulsoup4`,并要求导出的原始 HTML 目录:
```bash
python -m pip install beautifulsoup4
python tools/tsl-codegen/scripts/build_dictionary.py \
--source-root <exported-html-dir> \
--output-root skills/tsl-api-reference/references/data_dictionary \
--index skills/tsl-api-reference/data/dictionary_index.tsv \
--lexicon skills/tsl-api-reference/data/dictionary_lexicon.json
```
生成完成后,用 `dictionary_lookup.py --help` 中的真实示例做查询检查。
## 维护参数取值域
`skills/tsl-api-reference/data/value_domains.json` 是 API 参数取值域的检索数据,供
`lookup.py` 和 `dictionary_lookup.py` 共同读取。它不改变 `function_index.tsv` 的 API 元数据
职责,也不向 `dictionary_index.tsv` 增加 table/source/field 之外的 kind。
以下内容进入参数取值域:
- 由外部目录维护、需要跨 API 查询的版本化代码,如分类属性代码;
- 多个 API 参数共享的封闭枚举;只属于单个 API 的枚举仍留在该 API 页面;
- 已有正式文档但不构成封闭枚举的特殊字面值,使用 `catalog`
- 运行时才完整的系统或用户目录,如市场板块名称;
- 多个 API 共享的特殊选择器或值模式,如分类节点的 `.N` 后缀。
单个 API 自带的封闭枚举仍写在 codegen 录入数据的 `params[].values` 中,由生成器渲染为
参数取值段,不在两个位置重复维护。
维护规则:
1. FAQ 或帮助页逐条人工核对,不用脚本从语料批量筛选后直接入库。
2. 每个值记录原样值、名称、来源和适用的 API 参数;版本化目录按需记录父级、层级、
有效期及关联表。
3. `runtime_catalog` 必须为 `complete: false`,并提供系统/用户运行时解析器;静态值只能
作为已核实样例,不能宣称是完整目录。
4. 分类名称映射为当前板块名但本地证据不足时,关系标记为 `runtime_required`,由
`getBkList2` 或 `getUserBkList2` 核验后再用于代码。
5. 修改后运行参数域 CLI 回归和安装测试:
```bash
python -m unittest tools.tsl-codegen.tests.test_value_domains
python -m unittest test.test_playbook
```
## 完成检查
```bash
python skills/tsl-api-reference/scripts/class_lookup.py --check
python tools/tsl-codegen/scripts/build_index.py \
--skill-dir skills/tsl-api-reference --check
python -m unittest tools.tsl-codegen.tests.test_value_domains
python test/integration/check_doc_links.py
```
+12 -2
View File
@@ -18,7 +18,8 @@ tools/tsl-codegen/
│ ├─ generate.py yaml/json → markdown
│ ├─ lint.py markdown 格式校验
│ ├─ api_markdown.py lint/index 共用标题模型
─ build_index.py 重建 13 列 function_index.tsv
─ build_index.py 重建 13 列 function_index.tsv
│ └─ build_dictionary.py HTML 数据字典 → markdown/TSV
└─ tests/ 工具测试
```
@@ -193,6 +194,15 @@ tmp/my-api.json
一个录入文件对应一个 markdown 叶子页。录入文件不放入 skill;是否长期保留由
维护者自行决定
#### Class 与 Framework Profile
生成 class 页面并更新 `function_index.tsv` 后,class 已可通过自动成员视图查询,不要求
额外的 Framework Profile。`class_lookup.py` 返回 `profile_status=not_profiled` 表示普通
class,属于正常且完整的结果;不要为它创建空 profile。
本工具不会从 class 名称或成员列表推断生命周期、配置、回调和执行阶段。只有另有可核验
协议证据的框架类才需要 curated Framework Profile。
### 4. 选择 Skill 中的目标位置
TSL API skill 的相关目录如下:
@@ -391,4 +401,4 @@ python skills/tsl-api-reference/scripts/lookup.py --name DemoUnit.Document.Save
- `skills/tsl-api-reference/data/function_index.tsv`
TSL API skill 只需要 markdown 和 TSV。录入用的 yaml/json 可以由维护者在自己的
版本库中管理
版本库中管理
@@ -0,0 +1,837 @@
#!/usr/bin/env python3
"""Build the repository's searchable Tinysoft data-dictionary snapshot."""
from __future__ import annotations
import argparse
import csv
import io
import json
import os
import re
import tempfile
import unicodedata
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Iterable, Sequence
from bs4 import BeautifulSoup, NavigableString, Tag
INDEX_COLUMNS = (
"kind",
"scope",
"table_id",
"table_name",
"table_alias",
"field_id",
"field_name",
"field_alias",
"data_type",
"unit",
"description",
"extract_method",
"access_code",
"api_name",
"page",
"tags",
)
SECTION_LABELS = {
"更新日志",
"数据说明",
"数据来源",
"数据更新情况",
"基本概况",
"字段说明",
"特别说明",
"数据范例",
"访问代码",
"访问方法",
"取数示例",
"参考",
}
SCOPE_NAMES = {
"股票": "stock",
"基金": "fund",
"基金扩展": "fund",
"债券": "bond",
"期货": "futures",
"期权": "options",
"新三板": "neeq",
"指数": "index",
"回购": "repo",
"现货": "spot",
"宏观": "macro",
"中证": "csindex",
"天软": "tinysoft",
"行情": "market",
"证券": "security",
}
FIELD_NAME_HEADERS = ("字段名", "函数名", "名称", "含义", "字段")
UNIT_HEADERS = ("单位", "单位【注1】", "单位【注4】")
DESCRIPTION_HEADERS = ("详细说明", "说明", "备注")
INVALID_SLUG_CHARS_RE = re.compile(r"[\\/:*?\"<>|]+")
SPACE_RE = re.compile(r"\s+")
class DictionaryParseError(ValueError):
"""A dictionary page advertises a schema that cannot be parsed."""
class NotDictionaryPage(DictionaryParseError):
"""The source page is outside the data-dictionary branch."""
@dataclass(frozen=True)
class DictionaryField:
field_id: str
field_name: str
data_type: str
display_name: str
unit: str
description: str
api_name: str
@dataclass(frozen=True)
class DictionaryTable:
name: str
table_id: str
extract_method: str
access_code: str
update_info: tuple[str, ...]
aliases: tuple[str, ...] = ()
@dataclass(frozen=True)
class DictionaryExample:
description: str
code: str
output: str | None
@dataclass(frozen=True)
class DictionaryPage:
kind: str
scope: str
title: str
source_path: tuple[str, ...]
source_file: str
table: DictionaryTable
fields: tuple[DictionaryField, ...]
notes: tuple[str, ...]
examples: tuple[DictionaryExample, ...]
output_name: str = ""
@dataclass(frozen=True)
class DictionaryDocument:
source_file: str
source_path: tuple[str, ...]
records: tuple[DictionaryPage, ...]
@dataclass(frozen=True)
class BuildReport:
scanned_page_count: int
dictionary_page_count: int
navigation_page_count: int
page_count: int
table_count: int
source_count: int
field_count: int
ignored_pages: tuple[str, ...]
malformed_dictionary_pages: tuple[str, ...]
def _clean_text(value: str) -> str:
value = value.replace("\xa0", " ")
return SPACE_RE.sub(" ", value).strip()
def _cell_text(cell: Tag) -> str:
return _clean_text(cell.get_text(" ", strip=True))
def _table_rows(table: Tag) -> tuple[list[str], list[list[str]]]:
rows = []
for row in table.find_all("tr"):
cells = [_cell_text(cell) for cell in row.find_all(["th", "td"], recursive=False)]
if cells:
rows.append(cells)
if not rows:
return [], []
return rows[0], rows[1:]
def _page_title(content: Tag) -> str:
heading = content.find(["h1", "h2", "h3"])
if heading is None:
return ""
return _clean_text(heading.get_text(" ", strip=True).replace("复制链接", ""))
def _breadcrumb(raw: str) -> tuple[str, ...]:
prefix = raw.split('<div id="help_content">', 1)[0]
soup = BeautifulSoup(prefix, "html.parser")
labels = [_clean_text(link.get_text(" ", strip=True)) for link in soup.find_all("a")]
starts = [index for index, label in enumerate(labels) if label == "知识库"]
if starts:
labels = labels[starts[-1] :]
return tuple(label for label in labels if label)
def _is_dictionary_path(source_path: Sequence[str], title: str) -> bool:
return "天软数据字典" in source_path or title == "天软数据字典"
def _scope(source_path: Sequence[str]) -> str:
if "天软数据字典" not in source_path:
return ""
start = source_path.index("天软数据字典") + 1
for label in source_path[start:]:
if label in SCOPE_NAMES:
return SCOPE_NAMES[label]
return ""
def _domain_path(source_path: Sequence[str]) -> tuple[str, ...]:
if "天软数据字典" not in source_path:
return ()
start = source_path.index("天软数据字典") + 1
return tuple(source_path[start:])
def _section_marker(content: Tag, label: str) -> Tag | None:
for tag in content.find_all(["span", "strong"]):
if _clean_text(tag.get_text(" ", strip=True)).rstrip(":") == label:
return tag
return None
def _is_section_marker(node: Tag) -> bool:
if node.name not in {"span", "strong"}:
return False
return _clean_text(node.get_text(" ", strip=True)).rstrip(":") in SECTION_LABELS
def _inside(node: NavigableString, tag_name: str, class_name: str | None = None) -> bool:
parent = node.parent
while isinstance(parent, Tag):
if parent.name == tag_name and (class_name is None or class_name in parent.get("class", [])):
return True
parent = parent.parent
return False
def _section_lines(content: Tag, label: str) -> tuple[str, ...]:
marker = _section_marker(content, label)
if marker is None:
return ()
lines = []
for node in marker.next_elements:
if isinstance(node, Tag) and node is not marker and _is_section_marker(node):
break
if not isinstance(node, NavigableString) or node.parent is marker:
continue
if _inside(node, "table") or _inside(node, "div", "text-container"):
continue
text = _clean_text(str(node))
if text and (not lines or lines[-1] != text):
lines.append(text)
return tuple(lines)
def _nearest_section(table: Tag) -> str:
for tag in table.find_all_previous(["span", "strong"]):
label = _clean_text(tag.get_text(" ", strip=True)).rstrip(":")
if label in SECTION_LABELS:
return label
return ""
def _find_basic_table(content: Tag) -> Tag | None:
for table in content.find_all("table"):
header, _ = _table_rows(table)
if "表ID" in header and "表名" in header and "提取方式" in header:
return table
return None
def _field_table_header(header: Sequence[str]) -> bool:
return "类型" in header and any(name in header for name in FIELD_NAME_HEADERS)
def _find_field_tables(content: Tag) -> tuple[Tag, ...]:
found = []
for table in content.find_all("table"):
header, _ = _table_rows(table)
if _field_table_header(header) and _nearest_section(table) == "字段说明":
found.append(table)
return tuple(found)
def _column_index(header: Sequence[str], names: Sequence[str]) -> int | None:
for name in names:
if name in header:
return header.index(name)
return None
def _value(row: Sequence[str], index: int | None) -> str:
if index is None or index >= len(row):
return ""
return row[index]
def _parse_fields(table: Tag) -> tuple[DictionaryField, ...]:
header, rows = _table_rows(table)
id_index = _column_index(header, ("ID",))
identifier_index = _column_index(header, ("字段名", "函数名", "字段"))
display_index = _column_index(header, ("中文名", "名称", "含义"))
type_index = _column_index(header, ("类型",))
unit_index = _column_index(header, UNIT_HEADERS)
description_indices = [header.index(name) for name in DESCRIPTION_HEADERS if name in header]
api_index = _column_index(header, ("对应函数名称",))
fields = []
for row in rows:
field_id = _value(row, id_index)
identifier = _value(row, identifier_index)
display_name = _value(row, display_index)
api_name = _value(row, api_index)
if "函数名" in header and identifier:
api_name = api_name or identifier
field_name = identifier or display_name or field_id
display_name = display_name or identifier or field_id
descriptions = []
for index in description_indices:
value = _value(row, index)
if value and value not in descriptions:
descriptions.append(value)
if not field_name:
continue
fields.append(
DictionaryField(
field_id=field_id,
field_name=field_name,
data_type=_value(row, type_index),
display_name=display_name,
unit=_value(row, unit_index),
description=" ".join(descriptions),
api_name=api_name,
)
)
return tuple(fields)
def _extract_examples(content: Tag) -> tuple[DictionaryExample, ...]:
marker = _section_marker(content, "取数示例")
if marker is None:
return ()
examples = []
for container in marker.find_all_next("div", class_="text-container"):
previous = container.find_previous(["span", "strong"])
if previous is not marker and _is_section_marker(previous):
break
raw_lines = [_clean_text(line) for line in container.get_text("\n").splitlines()]
chunks: list[list[str]] = []
current: list[str] = []
for line in (line for line in raw_lines if line):
empty_output = re.fullmatch(r"//\s*(?:返回|输出)\s*[:]?", line, re.IGNORECASE)
if empty_output:
if current:
chunks.append(current)
current = []
continue
output_match = re.match(r"//\s*(?:返回|输出)\s*[:]\s*(.+)", line, re.IGNORECASE)
if not output_match and line.startswith("//") and current and any(
re.match(r"(?i)^\s*return\b", existing) for existing in current
):
chunks.append(current)
current = []
if output_match:
line = f"// 输出:{output_match.group(1).strip()}"
elif line.startswith("//"):
line = f"// {line[2:].strip()}"
current.append(line)
if current:
chunks.append(current)
for lines in chunks:
description = lines[0][2:].strip() if lines[0].startswith("//") else ""
output = None
for line in lines:
match = re.match(r"//\s*输出:(.+)", line)
if match:
output = match.group(1).strip()
examples.append(DictionaryExample(description, "\n".join(lines), output))
return tuple(examples)
def _basic_metadata(table: Tag) -> tuple[str, str, str]:
header, rows = _table_rows(table)
if len(rows) != 1:
raise DictionaryParseError("basic table must contain exactly one data row")
row = rows[0]
return (
_value(row, header.index("表ID")),
_value(row, header.index("表名")),
_value(row, header.index("提取方式")),
)
def _resolve_extract_method(content: Tag, extract_method: str) -> str:
if not re.fullmatch(r"【注\d+】", extract_method):
return extract_method
page_text = _clean_text(content.get_text(" ", strip=True))
match = re.search(r"(?:取数接口|接口)\s*([A-Za-z][A-Za-z0-9_]*)", page_text)
return match.group(1) if match else extract_method
def _normal_aliases(table_name: str, title: str) -> tuple[str, ...]:
candidates = [title, table_name.replace(".", ""), table_name]
if "." in table_name:
candidates.append(table_name.rsplit(".", 1)[-1])
return tuple(dict.fromkeys(value for value in candidates if value and value != table_name))
def _field_available(field: DictionaryField, source_name: str) -> bool:
description = unicodedata.normalize("NFKC", field.description).casefold()
source = source_name.casefold()
pattern = rf"{re.escape(source)}\s*[:]\s*(?:没有|无)(?:该)?字段"
return re.search(pattern, description) is None
def _split_market_sources(page: DictionaryPage) -> tuple[DictionaryPage, ...]:
shared = page.table
trade_table = replace(
shared,
name="TradeTable",
extract_method="TradeTable",
aliases=("交易明细", "交易明细表", "tradetable"),
)
market_table = replace(
shared,
name="MarketTable",
extract_method="MarketTable",
aliases=("分时", "分时表", "markettable"),
)
trade_examples = tuple(example for example in page.examples if "tradetable" in example.code.casefold())
market_examples = tuple(example for example in page.examples if "markettable" in example.code.casefold())
return (
replace(
page,
kind="source",
title="交易明细",
table=trade_table,
fields=tuple(field for field in page.fields if _field_available(field, "tradetable")),
examples=trade_examples,
),
replace(
page,
kind="source",
title="分时",
table=market_table,
fields=tuple(field for field in page.fields if _field_available(field, "markettable")),
examples=market_examples,
),
)
def _special_field_records(
content: Tag,
title: str,
source_path: tuple[str, ...],
source_file: str,
fields_by_table: Sequence[tuple[Tag, tuple[DictionaryField, ...]]],
) -> tuple[DictionaryPage, ...]:
access_lines = _section_lines(content, "访问方法")
methods = []
for line in access_lines:
if "" in line or ":" in line:
_, method = re.split(r"[:]", line, maxsplit=1)
methods.append(_clean_text(method))
all_examples = _extract_examples(content)
records = []
for index, (table, fields) in enumerate(fields_by_table):
label = ""
for sibling in table.previous_siblings:
if isinstance(sibling, NavigableString):
label = _clean_text(str(sibling)).rstrip(":")
if label:
break
record_title = f"{title}.{label}" if label else title
extract_method = methods[index] if index < len(methods) else ""
matching_examples = tuple(
example
for example in all_examples
if extract_method and extract_method.casefold() in example.code.casefold()
)
records.append(
DictionaryPage(
kind="source",
scope=_scope(source_path),
title=record_title,
source_path=_domain_path(source_path),
source_file=source_file,
table=DictionaryTable(
name=record_title,
table_id="",
extract_method=extract_method,
access_code="".join(access_lines),
update_info=_section_lines(content, "数据更新情况"),
aliases=tuple(filter(None, (title, label))),
),
fields=fields,
notes=_section_lines(content, "数据说明"),
examples=matching_examples or (all_examples if len(fields_by_table) == 1 else ()),
)
)
return tuple(records)
def parse_dictionary_document(path: Path) -> DictionaryDocument:
raw = path.read_text(encoding="utf-8", errors="replace")
soup = BeautifulSoup(raw, "html.parser")
content = soup.select_one("#help_content")
if content is None:
raise DictionaryParseError("missing #help_content")
title = _page_title(content)
source_path = _breadcrumb(raw)
if not _is_dictionary_path(source_path, title):
raise NotDictionaryPage(path.name)
basic_table = _find_basic_table(content)
field_tables = _find_field_tables(content)
if basic_table is None and not field_tables:
return DictionaryDocument(path.name, source_path, ())
if not field_tables:
raise DictionaryParseError("dictionary record has no recognized field table")
parsed_tables = tuple((table, _parse_fields(table)) for table in field_tables)
if any(not fields for _, fields in parsed_tables):
raise DictionaryParseError("field table contains no data rows")
if basic_table is None:
records = _special_field_records(content, title, source_path, path.name, parsed_tables)
return DictionaryDocument(path.name, source_path, records)
table_id, table_name, extract_method = _basic_metadata(basic_table)
extract_method = _resolve_extract_method(content, extract_method)
if not table_name:
raise DictionaryParseError("basic table has no table name")
fields = tuple(field for _, group in parsed_tables for field in group)
page = DictionaryPage(
kind="table",
scope=_scope(source_path),
title=title,
source_path=_domain_path(source_path),
source_file=path.name,
table=DictionaryTable(
name=table_name,
table_id=table_id,
extract_method=extract_method,
access_code="".join(_section_lines(content, "访问代码")),
update_info=_section_lines(content, "数据更新情况"),
aliases=_normal_aliases(table_name, title),
),
fields=fields,
notes=_section_lines(content, "数据说明"),
examples=_extract_examples(content),
)
method_key = re.sub(r"[^a-z]", "", extract_method.casefold())
if "markettable" in method_key and "tradetable" in method_key:
records = _split_market_sources(page)
else:
records = (page,)
return DictionaryDocument(path.name, source_path, records)
def _slug_part(value: str) -> str:
value = unicodedata.normalize("NFKC", value).strip().casefold()
value = INVALID_SLUG_CHARS_RE.sub("_", value)
value = re.sub(r"[.。·,,、]+", "_", value)
value = SPACE_RE.sub("_", value)
value = re.sub(r"_+", "_", value).strip("_")
return value
def dictionary_slug(page: DictionaryPage) -> str:
if page.output_name:
return page.output_name
parts = [*page.source_path, page.title]
if page.kind == "source" and page.table.name.casefold() not in page.title.casefold():
parts.append(page.table.name)
compact = []
for part in parts:
slug = _slug_part(part)
if slug and (not compact or compact[-1] != slug):
compact.append(slug)
if not compact:
raise DictionaryParseError(f"cannot create slug for {page.source_file}")
return "__".join(compact) + ".md"
def _resolve_slug_collisions(pages: Sequence[DictionaryPage]) -> tuple[DictionaryPage, ...]:
grouped: dict[str, list[DictionaryPage]] = {}
for page in pages:
grouped.setdefault(dictionary_slug(page), []).append(page)
resolved = []
used = set()
for page in pages:
base = dictionary_slug(page)
group = grouped[base]
if len(group) == 1:
output_name = base
else:
discriminator = _slug_part(page.table.table_id or page.table.extract_method)
if not discriminator or "" in discriminator:
sources = ", ".join(item.source_file for item in group)
raise DictionaryParseError(f"slug collision needs a semantic discriminator: {base}: {sources}")
output_name = f"{Path(base).stem}__{discriminator}.md"
if output_name in used:
sources = ", ".join(item.source_file for item in group)
raise DictionaryParseError(f"slug collision: {output_name}: {sources}")
used.add(output_name)
resolved.append(replace(page, output_name=output_name))
return tuple(resolved)
def _markdown_cell(value: str) -> str:
return value.replace("|", "\\|").replace("\n", "<br>")
def _render_note(note: str) -> str:
"""Keep source numbering as text instead of creating a nested list."""
return re.sub(r"^(\s*\d+)[.]", r"\1\\.", note)
def render_dictionary_page(page: DictionaryPage) -> str:
path = " / ".join((*page.source_path, page.title))
lines = [f"# 天软数据字典 / {path}", "", f"类型:{page.kind}", "", "## 表信息", ""]
name_label = "数据源名称" if page.kind == "source" else "表名"
metadata = [(name_label, page.table.name)]
if page.table.table_id:
metadata.append(("表 ID", page.table.table_id))
if page.table.extract_method:
metadata.append(("提取方式", page.table.extract_method))
if page.table.access_code:
metadata.append(("访问代码", page.table.access_code))
if page.scope:
metadata.append(("范围", page.scope))
lines.extend(["| 项目 | 内容 |", "| --- | --- |"])
lines.extend(f"| {label} | {_markdown_cell(value)} |" for label, value in metadata)
lines.extend(["", "## 字段", "", "| 字段 ID | 字段名 | 类型 | 中文名 | 单位 | 说明 | 对应 API |", "| --- | --- | --- | --- | --- | --- | --- |"])
for field in page.fields:
values = (
field.field_id,
field.field_name,
field.data_type,
field.display_name,
field.unit,
field.description,
field.api_name,
)
lines.append("| " + " | ".join(_markdown_cell(value) for value in values) + " |")
if page.notes or page.table.update_info:
lines.extend(["", "## 数据说明", ""])
for note in (*page.notes, *page.table.update_info):
lines.append(f"- {_render_note(note)}")
if page.examples:
lines.extend(["", "## 取数示例", ""])
for index, example in enumerate(page.examples, start=1):
if example.description:
lines.extend([f"范例{index:02d}{example.description}", ""])
lines.extend(["```tsl", example.code, "```", ""])
while lines and not lines[-1]:
lines.pop()
return "\n".join(lines) + "\n"
def _page_reference(page: DictionaryPage) -> str:
return f"references/data_dictionary/{dictionary_slug(page)}"
def _field_alias(field: DictionaryField) -> str:
values = []
for value in (field.display_name, field.field_name):
if value and value not in values:
values.append(value)
return "|".join(values)
def build_index(pages: Iterable[DictionaryPage]) -> list[dict[str, str]]:
rows = []
keys = set()
for page in sorted(pages, key=lambda item: (dictionary_slug(item), item.table.name.casefold())):
common = {
"scope": page.scope,
"table_id": page.table.table_id,
"table_name": page.table.name,
"table_alias": "|".join(page.table.aliases),
"extract_method": page.table.extract_method,
"access_code": page.table.access_code,
"page": _page_reference(page),
"tags": "|".join((*page.source_path, page.title)),
}
entity = {
**{column: "" for column in INDEX_COLUMNS},
**common,
"kind": page.kind,
"description": " ".join(page.notes),
}
rows.append(entity)
for field in page.fields:
row = {
**{column: "" for column in INDEX_COLUMNS},
**common,
"kind": "field",
"field_id": field.field_id,
"field_name": field.field_name,
"field_alias": _field_alias(field),
"data_type": field.data_type,
"unit": field.unit,
"description": field.description,
"api_name": field.api_name,
}
key = (
row["kind"],
row["table_id"],
row["table_name"].casefold(),
row["field_id"],
row["field_name"].casefold(),
row["field_alias"].casefold(),
row["page"],
)
if key in keys:
raise DictionaryParseError(f"duplicate dictionary index key: {key}")
keys.add(key)
rows.append(row)
return rows
def _atomic_write_text(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
encoded = content.encode("utf-8")
descriptor, temporary_name = tempfile.mkstemp(
dir=path.parent, prefix=f".{path.name}.", suffix=".tmp"
)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "wb") as handle:
handle.write(encoded)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
if path.read_bytes() != encoded:
raise OSError(f"generated file verification failed: {path}")
finally:
temporary.unlink(missing_ok=True)
def write_index(rows: Iterable[dict[str, str]], path: Path) -> None:
handle = io.StringIO(newline="")
writer = csv.DictWriter(handle, fieldnames=INDEX_COLUMNS, delimiter="\t", lineterminator="\n")
writer.writeheader()
for row in rows:
writer.writerow({column: row.get(column, "") for column in INDEX_COLUMNS})
_atomic_write_text(path, handle.getvalue())
def validate_lexicon(lexicon: object) -> dict[str, list[str]]:
if not isinstance(lexicon, dict):
raise ValueError("dictionary lexicon must be a JSON object")
validated = {}
for canonical, aliases in lexicon.items():
if not isinstance(canonical, str) or not canonical.strip():
raise ValueError("dictionary lexicon contains an empty canonical term")
if not isinstance(aliases, list) or not all(isinstance(item, str) and item.strip() for item in aliases):
raise ValueError(f"aliases for {canonical!r} must be non-empty strings")
validated[canonical] = list(dict.fromkeys(aliases))
return validated
def write_lexicon(lexicon: object, path: Path) -> None:
validated = validate_lexicon(lexicon)
_atomic_write_text(path, json.dumps(validated, ensure_ascii=False, indent=2) + "\n")
def load_lexicon(path: Path) -> dict[str, list[str]]:
return validate_lexicon(json.loads(path.read_text(encoding="utf-8")))
def build_snapshot(source_root: Path, output_root: Path, index_path: Path) -> BuildReport:
records = []
ignored = []
malformed = []
dictionary_count = 0
navigation_count = 0
source_paths = sorted(source_root.glob("*.html"), key=lambda path: int(path.stem) if path.stem.isdigit() else path.name)
for source_path in source_paths:
try:
document = parse_dictionary_document(source_path)
except NotDictionaryPage:
ignored.append(source_path.name)
continue
except DictionaryParseError as error:
dictionary_count += 1
malformed.append(f"{source_path.name}: {error}")
continue
dictionary_count += 1
if not document.records:
navigation_count += 1
continue
records.extend(document.records)
records = list(_resolve_slug_collisions(records))
slugs = {dictionary_slug(page): page.source_file for page in records}
output_root.mkdir(parents=True, exist_ok=True)
expected = set(slugs)
for stale in output_root.glob("*.md"):
if stale.name not in expected:
stale.unlink()
for page in records:
_atomic_write_text(output_root / dictionary_slug(page), render_dictionary_page(page))
rows = build_index(records)
write_index(rows, index_path)
return BuildReport(
scanned_page_count=len(source_paths),
dictionary_page_count=dictionary_count,
navigation_page_count=navigation_count,
page_count=len(records),
table_count=sum(page.kind == "table" for page in records),
source_count=sum(page.kind == "source" for page in records),
field_count=sum(len(page.fields) for page in records),
ignored_pages=tuple(ignored),
malformed_dictionary_pages=tuple(malformed),
)
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("--source-root", required=True, type=Path)
parser.add_argument("--output-root", required=True, type=Path)
parser.add_argument("--index", required=True, type=Path)
parser.add_argument("--lexicon", type=Path)
return parser.parse_args(argv)
def main(argv: Sequence[str] | None = None) -> int:
args = parse_args(argv)
if args.lexicon is not None:
load_lexicon(args.lexicon)
report = build_snapshot(args.source_root, args.output_root, args.index)
print(
f"dictionary_pages={report.dictionary_page_count} generated_pages={report.page_count} "
f"tables={report.table_count} sources={report.source_count} fields={report.field_count} "
f"navigation={report.navigation_page_count} malformed={len(report.malformed_dictionary_pages)}"
)
for item in report.malformed_dictionary_pages:
print(f"MALFORMED\t{item}")
return 1 if report.malformed_dictionary_pages else 0
if __name__ == "__main__":
raise SystemExit(main())
+93 -6
View File
@@ -16,6 +16,7 @@ SCRIPT = (
/ "lookup.py"
)
SKILL_MD = SCRIPT.parents[1] / "SKILL.md"
API_WORKFLOW = SCRIPT.parents[1] / "workflows" / "api-lookup.md"
def load_script():
@@ -573,7 +574,7 @@ class LookupTest(unittest.TestCase):
self.assertEqual(2, result.returncode)
self.assertIn("--limit must be >= 1", result.stderr)
def test_help_lists_all_bundled_scopes(self):
def test_help_derives_scopes_from_selected_index(self):
result = subprocess.run(
[sys.executable, str(SCRIPT), "--help"],
capture_output=True,
@@ -582,7 +583,9 @@ class LookupTest(unittest.TestCase):
)
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("builtin、dotnet、third 与 deprecated", result.stdout)
self.assertIn("可用值从当前 function_index.tsv 读取", result.stdout)
self.assertNotIn("deprecated", result.stdout)
self.assertNotIn("--scope third", result.stdout)
def test_exact_search_uses_anchor_for_duplicate_signatures(self):
module = load_script()
@@ -775,14 +778,98 @@ class LookupTest(unittest.TestCase):
self.assertIn("DemoUnit.Document.Save", visibility.stdout)
def test_skill_documents_qualified_queries_and_legacy_index_compatibility(self):
text = SKILL_MD.read_text(encoding="utf-8")
skill_text = SKILL_MD.read_text(encoding="utf-8")
text = API_WORKFLOW.read_text(encoding="utf-8")
self.assertIn("workflows/api-lookup.md", skill_text)
self.assertIn("完全限定名称", text)
self.assertIn("--scope third", text)
self.assertIn("--scope module", text)
self.assertIn("qualified_name", text)
self.assertIn("owner、kind、binding、visibility", text)
self.assertIn("owner、kind、", text)
self.assertIn("binding 和 visibility", text)
self.assertIn("class function", text)
self.assertIn("8 列", text)
self.assertIn("8 列索引", text)
self.assertIn("13 列索引", text)
def test_skill_routes_workflows_and_keeps_stop_resident(self):
skill_text = SKILL_MD.read_text(encoding="utf-8")
skill_root = SKILL_MD.parent
instruction_paths = sorted((skill_root / "workflows").glob("*.md"))
for path in instruction_paths:
self.assertIn(path.relative_to(skill_root).as_posix(), skill_text)
self.assertIn("🔴 CHECKPOINT · 🛑 STOP", skill_text)
self.assertIn("rc=1 不进入上述“API 缺失”分支", skill_text)
self.assertIn("原样名称或 `qualified_name`", skill_text)
self.assertIn("`profile_status=not_profiled`", skill_text)
def test_skill_instruction_package_has_no_reverse_dependencies(self):
skill_root = SKILL_MD.parent
package_text = "\n".join(
path.read_text(encoding="utf-8")
for path in [SKILL_MD, *sorted((skill_root / "workflows").glob("*.md"))]
)
for forbidden in (
"tsl-syntax-reference",
"references/maintenance.md",
"tools/tsl-codegen",
"scripts/build_dictionary.py",
"AGENTS.md",
"CONTEXT.md",
"docs/adr/",
):
self.assertNotIn(forbidden, package_text)
def test_repository_maintenance_doc_owns_api_data_maintenance(self):
readme = (
SKILL_MD.parents[2] / "tools" / "tsl-codegen" / "MAINTENANCE.md"
).read_text(encoding="utf-8")
for required in (
"TSL API 数据维护",
"class_lookup.py --check",
"tools/tsl-codegen/scripts/build_index.py",
"--skill-dir skills/tsl-api-reference",
"tools/tsl-codegen/scripts/build_dictionary.py",
"beautifulsoup4",
"profile_status=not_profiled",
):
self.assertIn(required, readme)
user_readme = (
SKILL_MD.parents[2] / "tools" / "tsl-codegen" / "README.md"
).read_text(encoding="utf-8")
for maintainer_only in (
"class_lookup.py --check",
"beautifulsoup4",
"--source-root <exported-html-dir>",
):
self.assertNotIn(maintainer_only, user_readme)
self.assertIn("profile_status=not_profiled", user_readme)
def test_public_cli_help_owns_status_and_exit_contracts(self):
skill_root = SKILL_MD.parent
expected = {
"lookup.py": ("退出码:", "--name 无匹配", "page#anchor"),
"class_lookup.py": ("输出状态:", "contract_status", "退出码:"),
"dictionary_lookup.py": ("status: ambiguous", "status: no_match", "退出码:"),
}
for script_name, fragments in expected.items():
with self.subTest(script=script_name):
result = subprocess.run(
[
sys.executable,
str(skill_root / "scripts" / script_name),
"--help",
],
capture_output=True,
text=True,
check=False,
)
self.assertEqual(0, result.returncode, result.stderr)
for fragment in fragments:
self.assertIn(fragment, result.stdout)
def test_keyword_search_includes_tags_and_summary(self):
module = load_script()
+50 -2
View File
@@ -1,3 +1,4 @@
import importlib.util
import json
import subprocess
import sys
@@ -11,6 +12,8 @@ try:
except ImportError:
yaml = None
BS4_AVAILABLE = importlib.util.find_spec("bs4") is not None
TOOL_ROOT = Path(__file__).resolve().parents[1]
REPO_ROOT = TOOL_ROOT.parents[1]
@@ -18,12 +21,28 @@ CONVERT = TOOL_ROOT / "scripts" / "convert_tsf.py"
GENERATE = TOOL_ROOT / "scripts" / "generate.py"
LINT = TOOL_ROOT / "scripts" / "lint.py"
BUILD_INDEX = TOOL_ROOT / "scripts" / "build_index.py"
BUILD_DICTIONARY = TOOL_ROOT / "scripts" / "build_dictionary.py"
LOOKUP = REPO_ROOT / "skills" / "tsl-api-reference" / "scripts" / "lookup.py"
STANDARD = TOOL_ROOT / "STANDARD.md"
README = TOOL_ROOT / "README.md"
EXAMPLE_JSON = TOOL_ROOT / "examples" / "example.json"
EXAMPLE_YAML = TOOL_ROOT / "examples" / "example.yaml"
SKILL = REPO_ROOT / "skills" / "tsl-api-reference" / "SKILL.md"
API_WORKFLOW = (
REPO_ROOT / "skills" / "tsl-api-reference" / "workflows" / "api-lookup.md"
)
def load_build_dictionary():
spec = importlib.util.spec_from_file_location(
"tsl_codegen_build_dictionary", BUILD_DICTIONARY
)
if spec is None or spec.loader is None:
raise RuntimeError(f"cannot load {BUILD_DICTIONARY}")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
class UnifiedPipelineTest(unittest.TestCase):
@@ -35,6 +54,33 @@ class UnifiedPipelineTest(unittest.TestCase):
def tearDown(self):
self.temp_dir.cleanup()
@unittest.skipUnless(BS4_AVAILABLE, "beautifulsoup4 is not installed")
def test_dictionary_notes_escape_source_numbering(self):
module = load_build_dictionary()
page = module.DictionaryPage(
kind="table",
scope="macro",
title="示例",
source_path=("宏观",),
source_file="example.html",
table=module.DictionaryTable(
name="示例",
table_id="1",
extract_method="",
access_code="",
update_info=("1、更新频率:月度",),
),
fields=(),
notes=("1. 第一条说明", "2. 第二条说明"),
examples=(),
)
rendered = module.render_dictionary_page(page)
self.assertIn("- 1\\. 第一条说明", rendered)
self.assertIn("- 2\\. 第二条说明", rendered)
self.assertIn("- 1、更新频率:月度", rendered)
def run_command(self, *args, cwd=None):
return subprocess.run(
[sys.executable, *map(str, args)],
@@ -212,6 +258,7 @@ class UnifiedPipelineTest(unittest.TestCase):
standard = STANDARD.read_text(encoding="utf-8")
readme = README.read_text(encoding="utf-8")
skill = SKILL.read_text(encoding="utf-8")
api_workflow = API_WORKFLOW.read_text(encoding="utf-8")
self.assertTrue(standard.startswith("# TSL API 文档标准\n"))
self.assertNotIn("根字段 `functions`、`class`、`unit`", standard)
@@ -273,8 +320,9 @@ class UnifiedPipelineTest(unittest.TestCase):
self.assertNotIn("generate.py tmp/my-api.yaml", readme)
self.assertNotIn("generate.py tmp/my-api.json", readme)
self.assertNotIn("已废弃,请使用 --file", readme)
self.assertIn("混合页面", skill)
self.assertIn("page#anchor", skill)
self.assertIn("workflows/api-lookup.md", skill)
self.assertIn("混合页面", api_workflow)
self.assertIn("page#anchor", api_workflow)
def test_standard_class_methods_reuse_top_level_function_structure(self):
standard = STANDARD.read_text(encoding="utf-8")
@@ -0,0 +1,542 @@
import csv
import json
import re
import subprocess
import sys
import tempfile
import unittest
from collections import defaultdict
from pathlib import Path
ROOT = Path(__file__).resolve().parents[3]
SKILL_ROOT = ROOT / "skills" / "tsl-api-reference"
LOOKUP = SKILL_ROOT / "scripts" / "lookup.py"
DICTIONARY_LOOKUP = SKILL_ROOT / "scripts" / "dictionary_lookup.py"
class ValueDomainCliTest(unittest.TestCase):
def run_cli(self, script, *args):
return subprocess.run(
[sys.executable, str(script), *args],
cwd=ROOT,
capture_output=True,
text=True,
encoding="utf-8",
check=False,
)
def test_dictionary_query_resolves_classification_by_label_and_code(self):
for query in ("申万煤炭", "SWHY740000"):
with self.subTest(query=query):
result = self.run_cli(DICTIONARY_LOOKUP, "--query", query)
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("status: ok", result.stdout)
self.assertIn("类型:参数取值", result.stdout)
self.assertIn(
"参数域:classification_code(分类属性代码)",
result.stdout,
)
self.assertIn("取值:SWHY740000", result.stdout)
self.assertIn("名称:申万煤炭", result.stdout)
self.assertIn("父级:SWHY(申万行业)", result.stdout)
self.assertIn("层级:1", result.stdout)
self.assertIn("关联表:138 股票.股票行业分类信息", result.stdout)
self.assertIn("快照日期:2023-10-13", result.stdout)
self.assertIn("来源:faq:31494", result.stdout)
def test_dictionary_query_uses_strong_values_without_hijacking_table_names(self):
for query in ("SWHY740000 是什么", "申万煤炭分类代码"):
with self.subTest(query=query):
result = self.run_cli(DICTIONARY_LOOKUP, "--query", query)
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("类型:参数取值", result.stdout)
self.assertIn("取值:SWHY740000", result.stdout)
table = self.run_cli(DICTIONARY_LOOKUP, "--query", "申万行业配置")
self.assertEqual(0, table.returncode, table.stderr)
self.assertIn("类型:table", table.stdout)
self.assertIn("表 ID629", table.stdout)
self.assertNotIn("类型:参数取值", table.stdout)
def test_pattern_punctuation_is_significant(self):
pattern = self.run_cli(DICTIONARY_LOOKUP, "--query", ".N")
single_letter = self.run_cli(DICTIONARY_LOOKUP, "--query", "N")
self.assertEqual(0, pattern.returncode, pattern.stderr)
self.assertIn("类型:参数取值", pattern.stdout)
self.assertIn("取值:<分类属性代码>.N", pattern.stdout)
self.assertEqual(0, single_letter.returncode, single_letter.stderr)
self.assertNotIn("类型:参数取值", single_letter.stdout)
def test_shared_board_type_enum_maps_multiple_api_parameters(self):
label = self.run_cli(DICTIONARY_LOOKUP, "--query", "申万三级行业")
numeric = self.run_cli(DICTIONARY_LOOKUP, "--query", "板块类别 7")
bare_numeric = self.run_cli(DICTIONARY_LOOKUP, "--query", "7")
api = self.run_cli(LOOKUP, "--name", "getbktypelist")
bare_lookup = self.run_cli(LOOKUP, "--kw", "7")
self.assertEqual(0, label.returncode, label.stderr)
self.assertIn("status: ambiguous", label.stdout)
self.assertIn("取值:7", label.stdout)
self.assertIn("名称:申万三级行业", label.stdout)
self.assertIn("绑定 APIgetbktypelist.bktype; stocksbklist.bktype", label.stdout)
self.assertIn("字段:申万三级行业", label.stdout)
self.assertEqual(0, numeric.returncode, numeric.stderr)
self.assertIn("名称:申万三级行业", numeric.stdout)
self.assertEqual(0, bare_numeric.returncode, bare_numeric.stderr)
self.assertNotIn("类型:参数取值", bare_numeric.stdout)
self.assertEqual(0, api.returncode, api.stderr)
self.assertIn("market_board_type", api.stdout)
self.assertIn("完整", api.stdout)
self.assertIn("`7` — 申万三级行业", api.stdout)
self.assertEqual(0, bare_lookup.returncode, bare_lookup.stderr)
self.assertNotIn("参数取值域:getbktypelist.bktype", bare_lookup.stdout)
def test_historical_market_board_codes_use_the_shared_domain(self):
dictionary = self.run_cli(DICTIONARY_LOOKUP, "--query", "TSI000001")
lookup = self.run_cli(LOOKUP, "--kw", "TSI000001")
self.assertEqual(0, dictionary.returncode, dictionary.stderr)
self.assertIn("取值:TSI000001", dictionary.stdout)
self.assertIn("名称:A股板块", dictionary.stdout)
self.assertIn("绑定 APIgetBkByDate.index_id", dictionary.stdout)
self.assertEqual(0, lookup.returncode, lookup.stderr)
self.assertIn("历史市场板块代码:TSI000001(A股板块)", lookup.stdout)
self.assertNotIn("参数取值域:getbktypelist.bktype", lookup.stdout)
def test_exact_code_value_ignores_incidental_dictionary_id_substrings(self):
result = self.run_cli(DICTIONARY_LOOKUP, "--query", "SWHY740000")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("status: ok", result.stdout)
self.assertIn("取值:SWHY740000", result.stdout)
self.assertNotIn("字段:截止日", result.stdout)
def test_named_historical_board_domain_is_complete_and_api_specific(self):
dictionary = self.run_cli(DICTIONARY_LOOKUP, "--query", "中小企业板")
lookup = self.run_cli(LOOKUP, "--kw", "中小企业板")
exact = self.run_cli(LOOKUP, "--name", "getAbkbyDate")
self.assertEqual(0, dictionary.returncode, dictionary.stderr)
self.assertIn("绑定 APIgetAbkbyDate.bk_name", dictionary.stdout)
self.assertIn("2021-04-06 并入主板", dictionary.stdout)
self.assertEqual(0, lookup.returncode, lookup.stderr)
self.assertIn("历史市场板块名:中小企业板", lookup.stdout)
self.assertEqual(0, exact.returncode, exact.stderr)
self.assertIn("historical_a_share_board_name", exact.stdout)
self.assertIn("`中小企业板` — 中小企业板(历史兼容)", exact.stdout)
def test_scope_filters_multi_table_value_domains(self):
stock = self.run_cli(
DICTIONARY_LOOKUP, "--query", "属性代码", "--scope", "stock"
)
fund = self.run_cli(
DICTIONARY_LOOKUP, "--query", "属性代码", "--scope", "fund"
)
self.assertEqual(0, stock.returncode, stock.stderr)
self.assertIn("关联表:138 股票.股票行业分类信息", stock.stdout)
self.assertNotIn("355 基金.基金分类信息", stock.stdout)
self.assertEqual(0, fund.returncode, fund.stderr)
self.assertIn("关联表:355 基金.基金分类信息", fund.stdout)
self.assertNotIn("138 股票.股票行业分类信息", fund.stdout)
def test_keyword_query_maps_value_to_current_and_historical_apis(self):
result = self.run_cli(LOOKUP, "--kw", "申万", "煤炭")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("getBk\tfunction\tgetBk(marketlist)", result.stdout)
self.assertIn(
"getBkByDate\tfunction\tgetBkByDate(index_id, end_t, extype)",
result.stdout,
)
self.assertIn("getBk.marketlist", result.stdout)
self.assertIn("当前板块候选:申万煤炭", result.stdout)
self.assertIn("运行时核验", result.stdout)
self.assertIn("getBkByDate.index_id", result.stdout)
self.assertIn("历史分类代码:SWHY740000", result.stdout)
def test_keyword_annotations_only_include_the_top_value_group(self):
exact_code = self.run_cli(LOOKUP, "--kw", "SWHY740000")
domain_query = self.run_cli(LOOKUP, "--kw", "属性", "代码")
self.assertEqual(0, exact_code.returncode, exact_code.stderr)
self.assertIn("历史分类代码:SWHY740000", exact_code.stdout)
self.assertNotIn("历史分类代码:SWHY(申万行业)", exact_code.stdout)
self.assertEqual(0, domain_query.returncode, domain_query.stderr)
self.assertIn("目录选择器:属性代码", domain_query.stdout)
self.assertIn("下级分类代码模式:<分类属性代码>.N", domain_query.stdout)
self.assertNotIn("历史分类代码:CAPCHY", domain_query.stdout)
def test_exact_getbk_query_explains_runtime_catalog_boundary(self):
result = self.run_cli(LOOKUP, "--name", "getBk")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("参数取值域", result.stdout)
self.assertIn("marketlist", result.stdout)
self.assertIn("runtime_catalog", result.stdout)
self.assertIn("静态记录不完整", result.stdout)
self.assertIn("getBkList2", result.stdout)
self.assertIn("getUserBkList2", result.stdout)
self.assertIn("按值查询", result.stdout)
self.assertNotIn("`申万煤炭`", result.stdout)
def test_runtime_domain_query_returns_resolvers_not_unrelated_market_codes(self):
result = self.run_cli(DICTIONARY_LOOKUP, "--query", "market_board")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("status: ok", result.stdout)
self.assertIn("类型:参数取值域", result.stdout)
self.assertIn("参数域:market_board", result.stdout)
self.assertIn("绑定 APIgetBk.marketlist", result.stdout)
self.assertIn("getBkList2(bktype)(系统目录)", result.stdout)
self.assertIn("getUserBkList2(bktype)(用户目录)", result.stdout)
self.assertIn("申万煤炭", result.stdout)
self.assertNotIn("TSI000001", result.stdout)
def test_domain_label_query_returns_one_domain_summary(self):
result = self.run_cli(DICTIONARY_LOOKUP, "--query", "分类属性代码")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("status: ok", result.stdout)
self.assertIn("参数域:classification_code", result.stdout)
self.assertIn("绑定 APIgetBkByDate.index_id", result.stdout)
self.assertIn("SWHY740000(申万煤炭)", result.stdout)
self.assertNotIn("类型:field", result.stdout)
def test_exact_api_query_does_not_expand_versioned_catalog(self):
result = self.run_cli(LOOKUP, "--name", "getBkByDate")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("分类属性代码", result.stdout)
self.assertIn("已记录值或候选:7 条", result.stdout)
self.assertIn("dictionary_lookup.py --query", result.stdout)
parameter_domain = result.stdout.split("### 参数取值域", maxsplit=1)[1]
self.assertNotIn("`SWHY740000`", parameter_domain)
def test_invalid_value_domain_data_is_a_deployment_error(self):
with tempfile.TemporaryDirectory() as temp_dir:
invalid = Path(temp_dir) / "value_domains.json"
invalid.write_text('{"version": 1, "domains": []}\n', encoding="utf-8")
for script, action in (
(LOOKUP, ("--kw", "申万")),
(DICTIONARY_LOOKUP, ("--query", "申万")),
):
with self.subTest(script=script.name):
result = self.run_cli(
script,
*action,
"--value-domains",
str(invalid),
)
self.assertEqual(1, result.returncode)
self.assertEqual("", result.stdout)
self.assertIn("domains must not be empty", result.stderr)
self.assertNotIn("Traceback", result.stderr)
def test_parameter_domain_binding_does_not_cross_scope_or_module(self):
with tempfile.TemporaryDirectory() as temp_dir:
data_dir = Path(temp_dir) / "data"
data_dir.mkdir()
tsv = data_dir / "function_index.tsv"
tsv.write_text(
"name\tscope\tmodule\tsignature\tpage\tanchor\ttags\tsummary\n"
"getBk\tdotnet\tdatawarehouse\tgetBk(marketlist)\t"
"dotnet/market.md\tgetbk\t\t系统板块\n"
"getBk\tdotnet\tdatawarehouse\tgetBk()\t"
"dotnet/market.md\tgetbk-empty\t\t无参数重载\n"
"getBk\tproject\tdemo\tgetBk(name)\tproject/demo.md\t"
"getbk\t\t项目函数\n",
encoding="utf-8",
)
domains = data_dir / "value_domains.json"
domains.write_text(
json.dumps(
{
"version": 1,
"domains": [
{
"id": "market_board",
"label": "市场板块",
"mode": "runtime_catalog",
"complete": False,
"bindings": [
{
"scope": "dotnet",
"module": "datawarehouse",
"api": "getBk",
"parameter": "marketlist",
"role": "current_components",
}
],
"resolvers": [
{
"scope": "dotnet",
"module": "datawarehouse",
"api": "getBkList2",
"parameter": "bktype",
"catalog": "system",
"source": "net_function:28964",
}
],
"related_tables": [],
"values": [
{
"value": "申万煤炭",
"label": "申万煤炭",
"sources": ["faq:31494"],
}
],
"sources": ["faq:31494"],
"as_of": "2026-08-20",
}
],
},
ensure_ascii=False,
),
encoding="utf-8",
)
result = self.run_cli(
LOOKUP,
"--kw",
"申万煤炭",
"--tsv",
str(tsv),
"--value-domains",
str(domains),
)
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("dotnet/market.md#getbk", result.stdout)
self.assertNotIn("dotnet/market.md#getbk-empty", result.stdout)
self.assertNotIn("project/demo.md#getbk", result.stdout)
def test_exact_query_attaches_domain_only_to_overload_with_bound_parameter(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
data_dir = root / "data"
codegen = root / "references" / "codegen" / "dotnet"
data_dir.mkdir()
codegen.mkdir(parents=True)
tsv = data_dir / "function_index.tsv"
tsv.write_text(
"name\tscope\tmodule\tsignature\tpage\tanchor\ttags\tsummary\n"
"getBk\tdotnet\tdatawarehouse\tgetBk(marketlist)\t"
"dotnet/market.md\tgetbk\t\t系统板块\n"
"getBk\tdotnet\tdatawarehouse\tgetBk()\t"
"dotnet/market.md\tgetbk-1\t\t无参数重载\n",
encoding="utf-8",
)
(codegen / "market.md").write_text(
"# Dotnet\n\n"
"## `getBk(marketlist)`\n\n声明:function\n\n系统板块\n\n"
"## `getBk()`\n\n声明:function\n\n无参数重载\n",
encoding="utf-8",
)
domains = data_dir / "value_domains.json"
domains.write_text(
json.dumps(
{
"version": 1,
"domains": [
{
"id": "market_board",
"label": "市场板块",
"mode": "runtime_catalog",
"complete": False,
"bindings": [
{
"scope": "dotnet",
"module": "datawarehouse",
"api": "getBk",
"parameter": "marketlist",
"role": "current_components",
}
],
"resolvers": [
{
"scope": "dotnet",
"module": "datawarehouse",
"api": "getBkList2",
"parameter": "bktype",
"catalog": "system",
"source": "net_function:28964",
}
],
"related_tables": [],
"values": [],
"sources": ["net_function:28960"],
"as_of": "2026-08-20",
}
],
},
ensure_ascii=False,
),
encoding="utf-8",
)
result = self.run_cli(
LOOKUP,
"--name",
"getBk",
"--tsv",
str(tsv),
"--value-domains",
str(domains),
)
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("## `getBk(marketlist)`", result.stdout)
self.assertIn("## `getBk()`", result.stdout)
self.assertEqual(1, result.stdout.count("### 参数取值域"))
def test_exact_query_does_not_attach_domain_to_same_name_in_other_scope(self):
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
data_dir = root / "data"
codegen = root / "references" / "codegen"
(codegen / "dotnet").mkdir(parents=True)
(codegen / "project").mkdir(parents=True)
data_dir.mkdir()
tsv = data_dir / "function_index.tsv"
tsv.write_text(
"name\tscope\tmodule\tsignature\tpage\tanchor\ttags\tsummary\n"
"getBk\tdotnet\tdatawarehouse\tgetBk(marketlist)\t"
"dotnet/market.md\tgetbk\t\t系统板块\n"
"getBk\tproject\tdemo\tgetBk(name)\tproject/demo.md\t"
"getbk\t\t项目函数\n",
encoding="utf-8",
)
(codegen / "dotnet" / "market.md").write_text(
"# Dotnet\n\n## `getBk(marketlist)`\n\n声明:function\n\n系统板块\n",
encoding="utf-8",
)
(codegen / "project" / "demo.md").write_text(
"# Project\n\n## `getBk(name)`\n\n声明:function\n\n项目函数\n",
encoding="utf-8",
)
domains = data_dir / "value_domains.json"
domains.write_text(
json.dumps(
{
"version": 1,
"domains": [
{
"id": "market_board",
"label": "市场板块",
"mode": "runtime_catalog",
"complete": False,
"bindings": [
{
"scope": "dotnet",
"module": "datawarehouse",
"api": "getBk",
"parameter": "marketlist",
"role": "current_components",
}
],
"resolvers": [
{
"scope": "dotnet",
"module": "datawarehouse",
"api": "getBkList2",
"parameter": "bktype",
"catalog": "system",
"source": "net_function:28964",
}
],
"related_tables": [],
"values": [],
"sources": ["net_function:28960"],
"as_of": "2026-08-20",
}
],
},
ensure_ascii=False,
),
encoding="utf-8",
)
result = self.run_cli(
LOOKUP,
"--scope",
"project",
"--name",
"getBk",
"--tsv",
str(tsv),
"--value-domains",
str(domains),
)
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("项目函数", result.stdout)
self.assertNotIn("参数取值域", result.stdout)
def test_bundled_value_domains_reference_existing_apis_parameters_and_tables(self):
document = json.loads(
(SKILL_ROOT / "data" / "value_domains.json").read_text(encoding="utf-8")
)
with (SKILL_ROOT / "data" / "function_index.tsv").open(
encoding="utf-8", newline=""
) as handle:
rows = list(csv.DictReader(handle, delimiter="\t"))
with (SKILL_ROOT / "data" / "dictionary_index.tsv").open(
encoding="utf-8", newline=""
) as handle:
dictionary_rows = list(csv.DictReader(handle, delimiter="\t"))
apis = defaultdict(set)
for row in rows:
match = re.search(r"\((.*)\)", row["signature"])
if not match:
continue
key = (
row["scope"].casefold(),
row["module"].casefold(),
row["name"].casefold(),
)
for parameter in match.group(1).split(","):
parameter = parameter.strip().strip("[]")
if parameter:
apis[key].add(parameter.casefold())
table_pages = {
row["table_id"]: row["page"]
for row in dictionary_rows
if row["kind"] in {"table", "source"} and row["table_id"]
}
for domain in document["domains"]:
for binding in domain["bindings"]:
self.assert_api_parameter_exists(apis, binding)
for resolver in domain["resolvers"]:
self.assert_api_parameter_exists(apis, resolver)
for table in domain["related_tables"]:
self.assertEqual(table["page"], table_pages.get(table["id"]))
self.assertTrue((SKILL_ROOT / table["page"]).is_file())
for value in domain["values"]:
for relation in value.get("relations", []):
self.assert_api_parameter_exists(apis, relation["verification"])
def assert_api_parameter_exists(self, apis, reference):
key = (
reference["scope"].casefold(),
reference["module"].casefold(),
(reference.get("api") or reference.get("resolver")).casefold(),
)
parameter = reference.get("parameter")
self.assertIn(key, apis)
if parameter:
self.assertIn(parameter.casefold(), apis[key])
if __name__ == "__main__":
unittest.main()