♻️ refactor(tsl-api-reference): split lookup workflows and maintenance tooling
This commit is contained in:
@@ -1,832 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build an independent searchable snapshot of the Tinysoft data dictionary."""
|
||||
|
||||
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_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"- {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())
|
||||
@@ -16,6 +16,43 @@ from framework_lookup import (
|
||||
from lookup import DEFAULT_TSV, codegen_root_for_tsv, load_rows, normalize
|
||||
|
||||
|
||||
HELP_EPILOG = """\
|
||||
查询顺序：
|
||||
1. 先按 class 取类摘要或 Framework Profile
|
||||
class_lookup.py --class TStringList
|
||||
class_lookup.py --class TSBackTesting --format json
|
||||
2. 需要成员正文时，再按返回的 qualified_name 调用 lookup.py --name
|
||||
|
||||
输出状态：
|
||||
status resolved、not_found 或 ambiguous
|
||||
profile_status resolved 表示有 Framework Profile；not_profiled 表示普通 class
|
||||
scaffold_status 生命周期和成员引用是否完整，仅在有 Profile 时输出
|
||||
contract_status 回调字段契约是否完整；非 resolved 时不得生成回调字段
|
||||
diagnostics 消歧、配置或契约缺口及下一步动作
|
||||
|
||||
--check 校验全部 Framework Profile 对 class、成员正文和证据的引用。
|
||||
使用自定义 --tsv 时默认不加载 Framework Profile；普通 class 仍可正常查询。若该索引
|
||||
另有 curated Profile，必须同时显式传入 --profiles。
|
||||
|
||||
退出码：
|
||||
0 查询或校验完成；not_found、ambiguous 通过输出状态表达
|
||||
1 function/framework 索引缺失、格式错误、为空或引用校验失败
|
||||
2 参数不合法
|
||||
"""
|
||||
|
||||
|
||||
def empty_profile_index():
|
||||
return {"schema_version": 1, "frameworks": []}
|
||||
|
||||
|
||||
def profile_path_for(args):
|
||||
if args.profiles:
|
||||
return Path(args.profiles)
|
||||
if args.tsv:
|
||||
return None
|
||||
return DEFAULT_INDEX
|
||||
|
||||
|
||||
def class_rows(rows):
|
||||
return [row for row in rows if row.get("kind") == "class"]
|
||||
|
||||
@@ -232,24 +269,32 @@ def main(argv=None):
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Query indexed TSL classes and curated framework profiles.",
|
||||
description="查询 TSL class 摘要及其证据化的 Framework Profile。",
|
||||
epilog=HELP_EPILOG,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
allow_abbrev=False,
|
||||
)
|
||||
action = parser.add_mutually_exclusive_group(required=True)
|
||||
action.add_argument("--class", dest="class_name", help="exact class name")
|
||||
action.add_argument("--list", action="store_true", help="list indexed classes")
|
||||
action.add_argument("--check", action="store_true", help="validate framework profiles")
|
||||
parser.add_argument("--scope", help="filter class scope")
|
||||
parser.add_argument("--module", help="filter class module")
|
||||
action.add_argument("--class", dest="class_name", help="精确 class 名或完全限定名")
|
||||
action.add_argument("--list", action="store_true", help="列出索引中的 class")
|
||||
action.add_argument("--check", action="store_true", help="校验全部 Framework Profile 引用")
|
||||
parser.add_argument("--scope", help="按 class scope 过滤")
|
||||
parser.add_argument("--module", help="按 class module 过滤")
|
||||
parser.add_argument(
|
||||
"--config", action="append", default=[], metavar="KEY=VALUE", help="raw config"
|
||||
"--config",
|
||||
action="append",
|
||||
default=[],
|
||||
metavar="KEY=VALUE",
|
||||
help="传入原始框架配置，可重复",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--include-members", action="store_true", help="include the compact member list"
|
||||
"--include-members", action="store_true", help="在类摘要中包含紧凑成员清单"
|
||||
)
|
||||
parser.add_argument("--format", choices=("text", "json"), default="text")
|
||||
parser.add_argument("--profiles", metavar="PATH", help="framework profile index")
|
||||
parser.add_argument("--tsv", metavar="PATH", help="function index")
|
||||
parser.add_argument(
|
||||
"--format", choices=("text", "json"), default="text", help="输出格式"
|
||||
)
|
||||
parser.add_argument("--profiles", metavar="PATH", help="Framework Profile 索引路径")
|
||||
parser.add_argument("--tsv", metavar="PATH", help="function_index.tsv 路径")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if args.config and not args.class_name:
|
||||
@@ -257,10 +302,12 @@ def main(argv=None):
|
||||
if args.include_members and not args.class_name:
|
||||
parser.error("--include-members requires --class")
|
||||
|
||||
profile_path = Path(args.profiles) if args.profiles else DEFAULT_INDEX
|
||||
tsv_path = Path(args.tsv) if args.tsv else DEFAULT_TSV
|
||||
profile_path = profile_path_for(args)
|
||||
try:
|
||||
profiles = load_index(profile_path)
|
||||
profiles = (
|
||||
load_index(profile_path) if profile_path else empty_profile_index()
|
||||
)
|
||||
rows = load_rows(tsv_path)
|
||||
except (OSError, UnicodeError, ValueError) as error:
|
||||
print(f"ERROR: {error}", file=sys.stderr)
|
||||
|
||||
@@ -35,6 +35,24 @@ REQUIRED_COLUMNS = (
|
||||
"tags",
|
||||
)
|
||||
|
||||
HELP_EPILOG = """\
|
||||
按中文业务含义查询表、数据源或字段：
|
||||
dictionary_lookup.py --query "股票现金流指标销售现金比率"
|
||||
dictionary_lookup.py --query "市值" --scope fund
|
||||
dictionary_lookup.py --query "买一量" --table TradeTable
|
||||
dictionary_lookup.py --query "9900700" --field 销售现金比率
|
||||
|
||||
输出第一行是 in-band 状态：
|
||||
status: ok 最高分候选唯一
|
||||
status: ambiguous 多个同分候选；必须补充 scope、table 或 field 范围
|
||||
status: no_match 数据字典无匹配，不等于 API not found
|
||||
|
||||
退出码：
|
||||
0 查询完成；ok、ambiguous、no_match 均由 status 表达
|
||||
1 dictionary_index.tsv 或词表缺失、损坏、格式错误
|
||||
2 参数不合法
|
||||
"""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class QueryItem:
|
||||
@@ -318,14 +336,21 @@ def _positive_int(value: str) -> int:
|
||||
|
||||
|
||||
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
|
||||
parser.add_argument("--query", required=True)
|
||||
parser.add_argument("--scope")
|
||||
parser.add_argument("--table")
|
||||
parser.add_argument("--field")
|
||||
parser.add_argument("--limit", type=_positive_int, default=10)
|
||||
parser.add_argument("--tsv", type=Path, default=DEFAULT_TSV)
|
||||
parser.add_argument("--lexicon", type=Path, default=DEFAULT_LEXICON)
|
||||
parser = argparse.ArgumentParser(
|
||||
description="查询独立的天软表、数据源与字段字典索引。",
|
||||
epilog=HELP_EPILOG,
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
allow_abbrev=False,
|
||||
)
|
||||
parser.add_argument("--query", required=True, help="中文业务描述、名称或 ID")
|
||||
parser.add_argument("--scope", help="按业务范围过滤")
|
||||
parser.add_argument("--table", help="按表名或数据源名过滤")
|
||||
parser.add_argument("--field", help="按字段名或字段 ID 过滤")
|
||||
parser.add_argument("--limit", type=_positive_int, default=10, help="候选条数上限")
|
||||
parser.add_argument("--tsv", type=Path, default=DEFAULT_TSV, help="字典索引路径")
|
||||
parser.add_argument(
|
||||
"--lexicon", type=Path, default=DEFAULT_LEXICON, help="受控同义词词表路径"
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
|
||||
@@ -1,15 +1,10 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Resolve framework-class profiles without replacing ordinary API lookup."""
|
||||
"""Library for validating and resolving evidence-backed framework profiles."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from lookup import (
|
||||
DEFAULT_TSV,
|
||||
codegen_root_for_tsv,
|
||||
load_rows,
|
||||
normalize,
|
||||
search_exact,
|
||||
slice_entry,
|
||||
@@ -230,33 +225,6 @@ def validate_index(data, rows, codegen_root):
|
||||
return errors
|
||||
|
||||
|
||||
def find_framework(data, name):
|
||||
requested = normalize(name)
|
||||
return next(
|
||||
(
|
||||
framework
|
||||
for framework in data.get("frameworks", [])
|
||||
if isinstance(framework, dict)
|
||||
and normalize(framework.get("qualified_name", "")) == requested
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
def parse_config(values):
|
||||
config = {}
|
||||
errors = []
|
||||
for value in values:
|
||||
key, separator, raw_value = value.partition("=")
|
||||
if not separator or not key.strip() or not raw_value.strip():
|
||||
errors.append(
|
||||
f"invalid --config {value!r}; expected a non-empty key=value pair"
|
||||
)
|
||||
continue
|
||||
config[key.strip()] = raw_value.strip()
|
||||
return config, errors
|
||||
|
||||
|
||||
def scaffold_packet(framework, config):
|
||||
configured_names = {
|
||||
entry["api"].rsplit(".", 1)[-1]
|
||||
@@ -320,135 +288,3 @@ def scaffold_packet(framework, config):
|
||||
"diagnostics": diagnostics,
|
||||
"evidence": framework.get("evidence", []),
|
||||
}
|
||||
|
||||
|
||||
def print_text(packet):
|
||||
print(f"framework: {packet['framework']}")
|
||||
print(f"scope/module: {packet['scope']}/{packet['module']}")
|
||||
print(f"scaffold_status: {packet['scaffold_status']}")
|
||||
print(f"contract_status: {packet['contract_status']}")
|
||||
print("lifecycle: " + " -> ".join(item["phase"] for item in packet["lifecycle"]))
|
||||
print("configuration: " + ", ".join(item["api"] for item in packet["configuration"]))
|
||||
required_hooks = [item["api"] for item in packet["hooks"] if item.get("required")]
|
||||
print("required_hooks: " + (", ".join(required_hooks) or "none"))
|
||||
print("state_apis: " + ", ".join(item["api"] for item in packet["state_apis"]))
|
||||
print("execution: " + ", ".join(item["api"] for item in packet["execution"]))
|
||||
print("result_apis: " + ", ".join(item["api"] for item in packet["result_apis"]))
|
||||
if packet["configured_values"]:
|
||||
print("configured_values: " + ", ".join(
|
||||
f"{key}={value}" for key, value in packet["configured_values"].items()
|
||||
))
|
||||
if packet["diagnostics"]:
|
||||
print("diagnostics:")
|
||||
for diagnostic in packet["diagnostics"]:
|
||||
print(f"- {diagnostic['code']}: {diagnostic['message']}")
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Resolve framework-class scaffolds for TSL API consumers.",
|
||||
allow_abbrev=False,
|
||||
)
|
||||
action = parser.add_mutually_exclusive_group(required=True)
|
||||
action.add_argument("--framework", help="exact framework qualified name")
|
||||
action.add_argument(
|
||||
"--check", action="store_true", help="validate all framework profile references"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--config",
|
||||
action="append",
|
||||
default=[],
|
||||
metavar="KEY=VALUE",
|
||||
help="raw framework configuration; may be repeated",
|
||||
)
|
||||
parser.add_argument("--format", choices=("text", "json"), default="text")
|
||||
parser.add_argument("--index", metavar="PATH", help="framework_index.json path")
|
||||
parser.add_argument("--tsv", metavar="PATH", help="function_index.tsv path")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
index_path = Path(args.index) if args.index else DEFAULT_INDEX
|
||||
tsv_path = Path(args.tsv) if args.tsv else DEFAULT_TSV
|
||||
try:
|
||||
data = load_index(index_path)
|
||||
rows = load_rows(tsv_path)
|
||||
except (OSError, UnicodeError, ValueError) as error:
|
||||
print(f"ERROR: {error}", file=sys.stderr)
|
||||
return 1
|
||||
if not rows:
|
||||
print(f"ERROR: {tsv_path} has no entries", file=sys.stderr)
|
||||
return 1
|
||||
codegen_root = codegen_root_for_tsv(tsv_path)
|
||||
|
||||
errors = validate_index(data, rows, codegen_root)
|
||||
if args.check:
|
||||
if errors:
|
||||
for error in errors:
|
||||
print(f"ERROR: {error}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"OK: {len(data['frameworks'])} framework profile(s) validated")
|
||||
return 0
|
||||
|
||||
if errors:
|
||||
packet = {
|
||||
"status": "data_error",
|
||||
"framework": args.framework,
|
||||
"diagnostics": [
|
||||
{
|
||||
"code": "PROFILE_REFERENCE_ERROR",
|
||||
"severity": "error",
|
||||
"message": error,
|
||||
"next_action": "Fix framework_index.json or rebuild the API index.",
|
||||
}
|
||||
for error in errors
|
||||
],
|
||||
}
|
||||
if args.format == "json":
|
||||
print(json.dumps(packet, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
print(f"Framework profile data is invalid for {args.framework!r}.")
|
||||
for diagnostic in packet["diagnostics"]:
|
||||
print(f"- {diagnostic['code']}: {diagnostic['message']}")
|
||||
return 1
|
||||
|
||||
framework = find_framework(data, args.framework)
|
||||
if framework is None:
|
||||
packet = {
|
||||
"status": "not_found",
|
||||
"framework": args.framework,
|
||||
"diagnostics": [
|
||||
{
|
||||
"code": "FRAMEWORK_NOT_FOUND",
|
||||
"severity": "error",
|
||||
"message": f"No framework profile named {args.framework!r}.",
|
||||
"next_action": "Use --check or add a framework profile.",
|
||||
}
|
||||
],
|
||||
}
|
||||
if args.format == "json":
|
||||
print(json.dumps(packet, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
print(f"No framework profile named {args.framework!r}.")
|
||||
return 0
|
||||
|
||||
config, config_errors = parse_config(args.config)
|
||||
packet = scaffold_packet(framework, config)
|
||||
for error in config_errors:
|
||||
packet["diagnostics"].append(
|
||||
{
|
||||
"code": "INVALID_CONFIGURATION",
|
||||
"severity": "error",
|
||||
"message": error,
|
||||
"next_action": "Use KEY=VALUE syntax with non-empty values.",
|
||||
}
|
||||
)
|
||||
if args.format == "json":
|
||||
print(json.dumps(packet, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
print_text(packet)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
||||
@@ -73,11 +73,16 @@ HELP_EPILOG = """\
|
||||
lookup.py --name arrDropDuplicate
|
||||
|
||||
已知确切名称时可以直接用 --name，跳过第 1 步。
|
||||
只查询一个 API scope 时使用 --scope，例如 --scope builtin、
|
||||
--scope dotnet 或 --scope module。
|
||||
只查询一个 API scope 时使用 --scope；可用值从当前 function_index.tsv 读取，
|
||||
不存在的 scope 会列出当前索引实际提供的值并返回 2。
|
||||
|
||||
结果：
|
||||
--kw 输出候选摘要表；选定后仍须运行 --name
|
||||
--name 输出完整条目正文和 scope/module、page#anchor 来源标记；
|
||||
简单成员名可能返回多个 owner，qualified_name 可消歧；重载会全部返回
|
||||
|
||||
退出码：
|
||||
0 取回成功；--name 无匹配也是 0（打印提示，不算错误）
|
||||
0 查询完成；--name 无匹配和 --kw 零候选也通过正文提示表达
|
||||
1 function_index.tsv 缺失、格式错误，或候选指向的条目正文找不到
|
||||
2 参数不合法（缺少动作、空值、--limit 小于 1、使用短选项或缩写）
|
||||
"""
|
||||
@@ -423,7 +428,7 @@ def main(argv=None):
|
||||
type=non_empty,
|
||||
metavar="SCOPE",
|
||||
help="只查询指定 scope，大小写不敏感；作用于 --name 和 --kw。"
|
||||
"内置索引当前提供 builtin、dotnet 与 module",
|
||||
"可用值从所选 function_index.tsv 读取并在加载后校验",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
|
||||
Reference in New Issue
Block a user