#!/usr/bin/env python3 """Query the independent Tinysoft table and field dictionary index.""" from __future__ import annotations import argparse import csv import json import re import sys import unicodedata from dataclasses import dataclass from pathlib import Path from typing import Iterable, Mapping, Sequence SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from value_domains import ( # noqa: E402 RelatedTable, ValueDomainCatalog, ValueMatch, domains_path_for_index, ) SKILL_ROOT = Path(__file__).resolve().parents[1] DEFAULT_TSV = SKILL_ROOT / "data" / "dictionary_index.tsv" DEFAULT_LEXICON = SKILL_ROOT / "data" / "dictionary_lexicon.json" DEFAULT_VALUE_DOMAINS = SKILL_ROOT / "data" / "value_domains.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", ) HELP_EPILOG = """\ 按中文业务含义查询表、数据源、字段或参数取值域： dictionary_lookup.py --query "股票现金流指标销售现金比率" dictionary_lookup.py --query "市值" --scope fund dictionary_lookup.py --query "买一量" --table TradeTable dictionary_lookup.py --query "9900700" --field 销售现金比率 dictionary_lookup.py --query "申万煤炭" 输出第一行是 in-band 状态： status: ok 最高分候选唯一 status: ambiguous 多个同分候选；必须补充 scope、table 或 field 范围 status: no_match 表/字段/参数取值域均无匹配，不等于 API not found 参数取值域结果会标明值、名称、绑定 API、关联表和来源；动态目录只提供已知样例， 完整名称必须使用结果中的运行时解析器重新确认。 退出码： 0 查询完成；ok、ambiguous、no_match 均由 status 表达 1 dictionary_index.tsv、词表或 value_domains.json 缺失、损坏、格式错误 2 参数不合法 """ @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 ValueDomainItem: kind: str domain_id: str domain_label: str value: str label: str parent: str parent_label: str level: int | None valid_from: str valid_to: str note: str related_tables: tuple[str, ...] bindings: tuple[str, ...] mode: str complete: bool as_of: str sources: tuple[str, ...] resolver_text: str @dataclass(frozen=True) class ValueDomainSummaryItem: kind: str domain_id: str label: str related_tables: tuple[str, ...] bindings: tuple[str, ...] mode: str complete: bool as_of: str sources: tuple[str, ...] resolvers: tuple[str, ...] candidates: tuple[str, ...] @dataclass(frozen=True) class QueryResult: status: str items: tuple[QueryItem | ValueDomainItem | ValueDomainSummaryItem, ...] @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 _value_item( catalog: ValueDomainCatalog, match: ValueMatch, related_tables: Sequence[RelatedTable] | None = None, ) -> ValueDomainItem: value = match.value hints = catalog.api_hints((match,)) bindings = tuple(dict.fromkeys(hint.target for hint in hints)) verification = [] for hint in hints: if not hint.verification: continue if hint.verification.status == "runtime_required": call = hint.verification.resolver if hint.verification.argument: call += f"('{hint.verification.argument}')" verification.append(f"{call}（运行时）") elif hint.verification.status == "documented": verification.append("文档已确认") resolver_text = "; ".join(dict.fromkeys(verification)) table = match.domain.related_table(value.related_table) selected_tables = tuple(related_tables) if related_tables is not None else () if not selected_tables: selected_tables = (table,) if table else match.domain.related_tables rendered_tables = tuple( f"{related.table_id} {related.name}" for related in selected_tables if related is not None ) return ValueDomainItem( kind="parameter_value", domain_id=match.domain.domain_id, domain_label=match.domain.label, value=value.value, label=value.label, parent=value.parent, parent_label=value.parent_label, level=value.level, valid_from=value.valid_from, valid_to=value.valid_to, note=value.note, related_tables=rendered_tables, bindings=bindings, mode=match.domain.mode, complete=match.domain.complete, as_of=value.as_of or match.domain.as_of, sources=value.sources, resolver_text=resolver_text, ) def _domain_summary_items( catalog: ValueDomainCatalog, query: str, *, scope: str | None, table: str | None, limit: int, ) -> tuple[ValueDomainSummaryItem, ...]: scope_key = _compact(scope or "") table_key = _compact(table or "") items = [] for domain in catalog.exact_domains(query): related_tables = domain.related_tables if scope_key: related_tables = tuple( related for related in related_tables if _compact(related.scope) == scope_key ) if table_key: related_tables = tuple( related for related in related_tables if table_key in _compact(related.name) or table_key in _compact(related.table_id) ) if (scope_key or table_key) and not related_tables: continue bindings = tuple( f"{binding.api}.{binding.parameter}" for binding in domain.bindings ) resolvers = [] for resolver in domain.resolvers: catalog_label = { "system": "系统", "user": "用户", }.get(resolver.catalog, resolver.catalog) resolvers.append( f"{resolver.api}({resolver.parameter})（{catalog_label}目录）" ) candidates = tuple( value if not label or label == value else f"{value}（{label}）" for value, label in catalog.recorded_candidates(domain) ) items.append( ValueDomainSummaryItem( kind="parameter_domain", domain_id=domain.domain_id, label=domain.label, related_tables=tuple( f"{related.table_id} {related.name}" for related in related_tables ), bindings=bindings, mode=domain.mode, complete=domain.complete, as_of=domain.as_of, sources=domain.sources, resolvers=tuple(resolvers), candidates=candidates, ) ) return tuple(items[:limit]) def _value_matches( catalog: ValueDomainCatalog, query: str, *, scope: str | None, table: str | None, field: str | None, limit: int, strong_only: bool, ) -> tuple[ValueDomainItem, ...]: if field: return () query_terms = tuple(query.split()) matches = catalog.search(query_terms) if len(query_terms) > 1 else () if not matches: matches = catalog.search(query) if re.fullmatch(r"\s*-?\d{1,2}\s*", query): matches = tuple( match for match in matches if not re.fullmatch(r"-?\d{1,2}", match.value.value) ) if strong_only: query_key = _compact(query) intent_markers = ("代码", "分类", "属性", "取值", "枚举") matches = tuple( match for match in matches if match.exact or ( len(_compact(match.value.value)) >= 4 and any( character.isascii() and character.isalpha() for character in match.value.value ) and any(character.isdigit() for character in match.value.value) and _compact(match.value.value) in query_key ) or ( match.domain.mode == "versioned_catalog" and _compact(match.value.label) in query_key and any(marker in query for marker in intent_markers) ) or ( match.domain.mode == "enum" and _compact(match.domain.label) in query_key and _compact(match.value.value) in query_key ) ) filtered = [] scope_key = _compact(scope or "") table_key = _compact(table or "") for match in matches: related_tables = match.domain.related_tables if match.value.related_table: related = match.domain.related_table(match.value.related_table) related_tables = (related,) if related else () if scope_key: related_tables = tuple( related for related in related_tables if _compact(related.scope) == scope_key ) if table_key: related_tables = tuple( related for related in related_tables if table_key in _compact(related.name) or table_key in _compact(related.table_id) ) if (scope_key or table_key) and not related_tables: continue filtered.append((match, related_tables)) matches_with_tables = tuple(filtered) if not matches_with_tables: return () top_score = matches_with_tables[0][0].score top = [item for item in matches_with_tables if item[0].score == top_score] return tuple( _value_item(catalog, match, related_tables) for match, related_tables in top[:limit] ) 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") rows = list(rows) 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 _direct_dictionary_items(result: QueryResult, query: str) -> tuple[QueryItem, ...]: query_key = _compact(query) if not query_key: return () direct = [] for item in result.items: if not isinstance(item, QueryItem): continue names = (_compact(item.table_name), _compact(item.field_name)) identifiers = (_compact(item.table_id), _compact(item.field_id)) access_code = _compact(item.access_code) if ( any(name and (query_key == name or name in query_key) for name in names) or any(identifier and query_key == identifier for identifier in identifiers) or (len(query_key) >= 3 and query_key in access_code) ): direct.append(item) return tuple(direct) 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_value_item(index: int, item: ValueDomainItem) -> list[str]: validity = "" if item.valid_from or item.valid_to: validity = f"{item.valid_from or '不限'} 至 {item.valid_to or '今'}" parent = item.parent or "无" if item.parent_label: parent = f"{parent}（{item.parent_label}）" return [ f"[{index}]", "类型：参数取值", f"参数域：{item.domain_id}（{item.domain_label}）", f"取值：{item.value}", f"名称：{item.label}", f"父级：{parent}", f"层级：{item.level if item.level is not None else ''}", f"有效期：{validity}", f"备注：{item.note}", f"关联表：{'; '.join(item.related_tables)}", f"绑定 API：{'; '.join(item.bindings)}", f"模式：{item.mode}", f"完整性：{'完整' if item.complete else '静态记录不完整'}", f"快照日期：{item.as_of}", f"核验：{item.resolver_text}", f"来源：{', '.join(item.sources)}", ] def _render_domain_summary_item( index: int, item: ValueDomainSummaryItem ) -> list[str]: return [ f"[{index}]", "类型：参数取值域", f"参数域：{item.domain_id}", f"名称：{item.label}", f"关联表：{'; '.join(item.related_tables)}", f"绑定 API：{'; '.join(item.bindings)}", f"模式：{item.mode}", f"完整性：{'完整' if item.complete else '静态记录不完整'}", f"快照日期：{item.as_of}", f"运行时解析器：{'; '.join(item.resolvers)}", f"已记录值或候选：{'; '.join(item.candidates)}", f"来源：{', '.join(item.sources)}", ] def render_result(result: QueryResult) -> str: lines = [f"status: {result.status}"] if result.status == "no_match": lines.append("未找到匹配的数据字典条目") elif result.status == "ambiguous": lines.append("存在多个同分候选，请补充 scope、表、数据源或字段范围") for index, item in enumerate(result.items, start=1): if isinstance(item, ValueDomainSummaryItem): renderer = _render_domain_summary_item elif isinstance(item, ValueDomainItem): renderer = _render_value_item else: renderer = _render_item lines.extend(["", *renderer(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="查询天软表、数据源、字段及 API 参数取值域。", 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="受控同义词词表路径" ) parser.add_argument( "--value-domains", type=Path, help="显式指定 value_domains.json；默认使用所选索引同目录的数据，" "自定义索引旁没有该文件时不加载参数取值域", ) return parser.parse_args(argv) def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) domain_path = ( args.value_domains if args.value_domains else domains_path_for_index( args.tsv, default_index=DEFAULT_TSV, default_domains=DEFAULT_VALUE_DOMAINS, ) ) try: rows = load_rows(args.tsv) lexicon = load_lexicon(args.lexicon) catalog = ValueDomainCatalog.load(domain_path) if domain_path else None dictionary_result = query_index( rows, args.query, scope=args.scope, table=args.table, field=args.field, limit=args.limit, lexicon=lexicon, ) domain_items = ( _domain_summary_items( catalog, args.query, scope=args.scope, table=args.table, limit=args.limit, ) if catalog and not args.field else () ) value_items = ( _value_matches( catalog, args.query, scope=args.scope, table=args.table, field=args.field, limit=args.limit, strong_only=True, ) if catalog else () ) if domain_items: result = QueryResult( "ambiguous" if len(domain_items) > 1 else "ok", domain_items, ) elif value_items: dictionary_items = _direct_dictionary_items(dictionary_result, args.query) combined = (*value_items, *dictionary_items)[: args.limit] result = QueryResult( "ambiguous" if len(value_items) > 1 or dictionary_items else "ok", combined, ) else: result = dictionary_result if result.status == "no_match" and catalog: fallback_items = _value_matches( catalog, args.query, scope=args.scope, table=args.table, field=args.field, limit=args.limit, strong_only=False, ) if fallback_items: result = QueryResult( "ambiguous" if len(fallback_items) > 1 else "ok", fallback_items, ) 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())