833 lines
29 KiB
Python
833 lines
29 KiB
Python
#!/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())
|