#!/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())