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