✨ feat(tsl-api-reference): update API catalog and data dictionary
This commit is contained in:
@@ -0,0 +1,354 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Query the independent Tinysoft table and field dictionary index."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import sys
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Iterable, Mapping, Sequence
|
||||
|
||||
|
||||
SKILL_ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_TSV = SKILL_ROOT / "data" / "dictionary_index.tsv"
|
||||
DEFAULT_LEXICON = SKILL_ROOT / "data" / "dictionary_lexicon.json"
|
||||
REQUIRED_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",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class QueryItem:
|
||||
kind: str
|
||||
scope: str
|
||||
table_id: str
|
||||
table_name: str
|
||||
field_id: str
|
||||
field_name: str
|
||||
field_alias: str
|
||||
data_type: str
|
||||
unit: str
|
||||
description: str
|
||||
extract_method: str
|
||||
access_code: str
|
||||
api_name: str
|
||||
page: str
|
||||
score: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class QueryResult:
|
||||
status: str
|
||||
items: tuple[QueryItem, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _ScoredRow:
|
||||
score: int
|
||||
field_match: int
|
||||
table_match: int
|
||||
row: Mapping[str, str]
|
||||
|
||||
|
||||
def _compact(value: str) -> str:
|
||||
normalized = unicodedata.normalize("NFKC", value).casefold()
|
||||
return "".join(
|
||||
character
|
||||
for character in normalized
|
||||
if character == "_" or unicodedata.category(character)[0] in {"L", "N"}
|
||||
)
|
||||
|
||||
|
||||
def _terms(value: str) -> tuple[str, ...]:
|
||||
terms = []
|
||||
for item in value.split("|"):
|
||||
compact = _compact(item)
|
||||
if compact and compact not in terms:
|
||||
terms.append(compact)
|
||||
return tuple(terms)
|
||||
|
||||
|
||||
def _row_terms(row: Mapping[str, str], name: str, alias: str) -> tuple[str, ...]:
|
||||
values = []
|
||||
for value in (_compact(row.get(name, "")), *_terms(row.get(alias, ""))):
|
||||
if value and value not in values:
|
||||
values.append(value)
|
||||
return tuple(values)
|
||||
|
||||
|
||||
def validate_lexicon(value: object) -> dict[str, list[str]]:
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError("dictionary lexicon must be a JSON object")
|
||||
result = {}
|
||||
for canonical, aliases in value.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(alias, str) and alias.strip() for alias in aliases):
|
||||
raise ValueError(f"aliases for {canonical!r} must be non-empty strings")
|
||||
result[canonical] = list(dict.fromkeys(aliases))
|
||||
return result
|
||||
|
||||
|
||||
def load_lexicon(path: Path) -> dict[str, list[str]]:
|
||||
return validate_lexicon(json.loads(path.read_text(encoding="utf-8")))
|
||||
|
||||
|
||||
def load_rows(path: Path) -> list[dict[str, str]]:
|
||||
with path.open(encoding="utf-8", newline="") as handle:
|
||||
reader = csv.DictReader(handle, delimiter="\t")
|
||||
if reader.fieldnames is None:
|
||||
raise ValueError("dictionary index has no header")
|
||||
missing = [column for column in REQUIRED_COLUMNS if column not in reader.fieldnames]
|
||||
if missing:
|
||||
raise ValueError(f"dictionary index is missing columns: {', '.join(missing)}")
|
||||
if len(reader.fieldnames) != len(set(reader.fieldnames)):
|
||||
raise ValueError("dictionary index has duplicate columns")
|
||||
return [dict(row) for row in reader]
|
||||
|
||||
|
||||
def _query_variants(query: str, lexicon: Mapping[str, Sequence[str]]) -> tuple[str, ...]:
|
||||
base = _compact(query)
|
||||
variants = {base}
|
||||
for canonical, aliases in lexicon.items():
|
||||
canonical_key = _compact(canonical)
|
||||
if not canonical_key:
|
||||
continue
|
||||
for alias in aliases:
|
||||
alias_key = _compact(alias)
|
||||
if alias_key and alias_key in base:
|
||||
variants.add(base.replace(alias_key, canonical_key))
|
||||
return tuple(sorted(variant for variant in variants if variant))
|
||||
|
||||
|
||||
def _name_score(query: str, terms: Iterable[str], *, exact: int, contained: int) -> int:
|
||||
score = 0
|
||||
for term in terms:
|
||||
if query == term:
|
||||
score = max(score, exact)
|
||||
elif term and term in query:
|
||||
score = max(score, contained + min(len(term), 100))
|
||||
return score
|
||||
|
||||
|
||||
def _score_row(row: Mapping[str, str], query_variants: Sequence[str]) -> _ScoredRow:
|
||||
table_terms = _row_terms(row, "table_name", "table_alias")
|
||||
field_terms = _row_terms(row, "field_name", "field_alias")
|
||||
field_id = _compact(row.get("field_id", ""))
|
||||
table_id = _compact(row.get("table_id", ""))
|
||||
api_name = _compact(row.get("api_name", ""))
|
||||
description = _compact(row.get("description", ""))
|
||||
tags = _terms(row.get("tags", ""))
|
||||
best = _ScoredRow(0, 0, 0, row)
|
||||
for query in query_variants:
|
||||
table_match = _name_score(query, table_terms, exact=1000, contained=600)
|
||||
field_match = _name_score(query, field_terms, exact=2000, contained=1300)
|
||||
if field_id and query == field_id:
|
||||
field_match = max(field_match, 3000)
|
||||
elif field_id and len(field_id) >= 3 and field_id in query:
|
||||
field_match = max(field_match, 2600)
|
||||
if table_id and query == table_id:
|
||||
table_match = max(table_match, 2400)
|
||||
elif table_id and len(table_id) >= 3 and table_id in query:
|
||||
table_match = max(table_match, 2100)
|
||||
if api_name and query == api_name:
|
||||
field_match = max(field_match, 2200)
|
||||
elif api_name and api_name in query:
|
||||
field_match = max(field_match, 1700 + min(len(api_name), 100))
|
||||
fallback = 0
|
||||
if not field_match and not table_match:
|
||||
if description and query in description:
|
||||
fallback = 200
|
||||
else:
|
||||
fallback = _name_score(query, tags, exact=180, contained=120)
|
||||
score = table_match + field_match + fallback
|
||||
if score > best.score:
|
||||
best = _ScoredRow(score, field_match, table_match, row)
|
||||
return best
|
||||
|
||||
|
||||
def _matches_filter(row: Mapping[str, str], value: str, name: str, alias: str) -> bool:
|
||||
key = _compact(value)
|
||||
return any(key == term or key in term or term in key for term in _row_terms(row, name, alias))
|
||||
|
||||
|
||||
def _item(scored: _ScoredRow) -> QueryItem:
|
||||
row = scored.row
|
||||
return QueryItem(
|
||||
kind=row.get("kind", ""),
|
||||
scope=row.get("scope", ""),
|
||||
table_id=row.get("table_id", ""),
|
||||
table_name=row.get("table_name", ""),
|
||||
field_id=row.get("field_id", ""),
|
||||
field_name=row.get("field_name", ""),
|
||||
field_alias=row.get("field_alias", ""),
|
||||
data_type=row.get("data_type", ""),
|
||||
unit=row.get("unit", ""),
|
||||
description=row.get("description", ""),
|
||||
extract_method=row.get("extract_method", ""),
|
||||
access_code=row.get("access_code", ""),
|
||||
api_name=row.get("api_name", ""),
|
||||
page=row.get("page", ""),
|
||||
score=scored.score,
|
||||
)
|
||||
|
||||
|
||||
def query_index(
|
||||
rows: Iterable[Mapping[str, str]],
|
||||
query: str,
|
||||
*,
|
||||
scope: str | None = None,
|
||||
table: str | None = None,
|
||||
field: str | None = None,
|
||||
limit: int = 10,
|
||||
lexicon: Mapping[str, Sequence[str]] | None = None,
|
||||
) -> QueryResult:
|
||||
if not query.strip():
|
||||
raise ValueError("query must not be empty")
|
||||
if limit < 1:
|
||||
raise ValueError("limit must be at least 1")
|
||||
variants = _query_variants(query, lexicon or {})
|
||||
scope_key = _compact(scope or "")
|
||||
scored = []
|
||||
for row in rows:
|
||||
if scope_key and _compact(row.get("scope", "")) != scope_key:
|
||||
continue
|
||||
if table and not _matches_filter(row, table, "table_name", "table_alias"):
|
||||
continue
|
||||
if field and not _matches_filter(row, field, "field_name", "field_alias"):
|
||||
continue
|
||||
candidate = _score_row(row, variants)
|
||||
if candidate.score:
|
||||
scored.append(candidate)
|
||||
if not scored:
|
||||
return QueryResult("no_match", ())
|
||||
|
||||
field_candidates = [item for item in scored if item.field_match]
|
||||
if field_candidates:
|
||||
scored = field_candidates
|
||||
else:
|
||||
entity_candidates = [item for item in scored if item.row.get("kind") in {"table", "source"}]
|
||||
if entity_candidates:
|
||||
scored = entity_candidates
|
||||
scored.sort(
|
||||
key=lambda item: (
|
||||
-item.score,
|
||||
_compact(item.row.get("table_name", "")),
|
||||
_compact(item.row.get("field_name", "")),
|
||||
item.row.get("field_id", ""),
|
||||
)
|
||||
)
|
||||
top_score = scored[0].score
|
||||
top = [candidate for candidate in scored if candidate.score == top_score]
|
||||
unique = []
|
||||
seen = set()
|
||||
for candidate in top:
|
||||
row = candidate.row
|
||||
key = (
|
||||
row.get("table_name", "").casefold(),
|
||||
row.get("field_id", ""),
|
||||
row.get("field_name", "").casefold(),
|
||||
row.get("page", ""),
|
||||
)
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
unique.append(_item(candidate))
|
||||
items = tuple(unique[:limit])
|
||||
status = "ambiguous" if len(unique) > 1 else "ok"
|
||||
return QueryResult(status, items)
|
||||
|
||||
|
||||
def _render_item(index: int, item: QueryItem) -> list[str]:
|
||||
return [
|
||||
f"[{index}]",
|
||||
f"类型：{item.kind}",
|
||||
f"范围：{item.scope}",
|
||||
f"表或数据源：{item.table_name}",
|
||||
f"表 ID：{item.table_id}",
|
||||
f"字段：{item.field_name}",
|
||||
f"字段别名或中文名：{item.field_alias}",
|
||||
f"字段 ID：{item.field_id}",
|
||||
f"字段类型：{item.data_type}",
|
||||
f"单位：{item.unit}",
|
||||
f"提取方式：{item.extract_method}",
|
||||
f"访问代码：{item.access_code}",
|
||||
f"对应 API：{item.api_name}",
|
||||
f"说明：{item.description}",
|
||||
f"页面：{item.page}",
|
||||
]
|
||||
|
||||
|
||||
def render_result(result: QueryResult) -> str:
|
||||
lines = [f"status: {result.status}"]
|
||||
if result.status == "no_match":
|
||||
lines.append("未找到匹配的数据字典条目")
|
||||
elif result.status == "ambiguous":
|
||||
lines.append("存在多个同分候选，请补充表或数据源范围")
|
||||
for index, item in enumerate(result.items, start=1):
|
||||
lines.extend(["", *_render_item(index, item)])
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def _positive_int(value: str) -> int:
|
||||
try:
|
||||
parsed = int(value)
|
||||
except ValueError as error:
|
||||
raise argparse.ArgumentTypeError("must be an integer") from error
|
||||
if parsed < 1:
|
||||
raise argparse.ArgumentTypeError("must be at least 1")
|
||||
return parsed
|
||||
|
||||
|
||||
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)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
try:
|
||||
rows = load_rows(args.tsv)
|
||||
lexicon = load_lexicon(args.lexicon)
|
||||
result = query_index(
|
||||
rows,
|
||||
args.query,
|
||||
scope=args.scope,
|
||||
table=args.table,
|
||||
field=args.field,
|
||||
limit=args.limit,
|
||||
lexicon=lexicon,
|
||||
)
|
||||
except (OSError, ValueError, json.JSONDecodeError) as error:
|
||||
print(f"ERROR: {error}", file=sys.stderr)
|
||||
return 1
|
||||
print(render_result(result), end="")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user