283 lines
8.8 KiB
Python
283 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import re
|
||
import zipfile
|
||
import xml.etree.ElementTree as ET
|
||
from dataclasses import dataclass, asdict
|
||
from pathlib import Path
|
||
|
||
|
||
FISH_SRC_DEFAULT = "coastline/C09-06.zip"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MatchRow:
|
||
fpc: str
|
||
prc: str | None
|
||
primary_name: str | None
|
||
secondary_name: str | None
|
||
source_id: str | None
|
||
score: tuple[int, int, str]
|
||
|
||
|
||
def local_name(tag: str) -> str:
|
||
return tag.split("}", 1)[-1]
|
||
|
||
|
||
def normalize_text(value: str) -> str:
|
||
value = value.strip()
|
||
value = value.replace(" ", " ")
|
||
value = re.sub(r"\s+", "", value)
|
||
return value.lower()
|
||
|
||
|
||
def load_xml(zip_path: Path) -> ET.Element:
|
||
with zipfile.ZipFile(zip_path) as zf:
|
||
xml_name = next(name for name in zf.namelist() if name.endswith(".xml") and "META" not in name)
|
||
return ET.fromstring(zf.read(xml_name))
|
||
|
||
|
||
def get_text(feature: ET.Element, key: str) -> str | None:
|
||
el = feature.find(f"{{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}}{key}")
|
||
if el is None or not (el.text or "").strip():
|
||
return None
|
||
return el.text.strip()
|
||
|
||
|
||
def feature_name_tokens(feature: ET.Element) -> list[str]:
|
||
tokens: list[str] = []
|
||
for key in ("NA2", "NA4", "FCF", "AAC", "CFP", "FPA"):
|
||
value = get_text(feature, key)
|
||
if value:
|
||
tokens.append(value)
|
||
return tokens
|
||
|
||
|
||
def feature_display_name(feature: ET.Element) -> tuple[str | None, str | None]:
|
||
na2 = get_text(feature, "NA2")
|
||
na4 = get_text(feature, "NA4")
|
||
fcf = get_text(feature, "FCF")
|
||
primary = na2 or na4 or fcf
|
||
secondary = None
|
||
if primary == na2:
|
||
secondary = fcf or na4
|
||
elif primary == na4:
|
||
secondary = na2 or fcf
|
||
else:
|
||
secondary = na2 or na4
|
||
return primary, secondary
|
||
|
||
|
||
def catalog_display_name(row: dict[str, str]) -> str | None:
|
||
for key in ("name", "name_ja", "port_name", "港名", "漁港名", "名称"):
|
||
value = row.get(key)
|
||
if value and value.strip():
|
||
return value.strip()
|
||
return None
|
||
|
||
|
||
def catalog_fpc(row: dict[str, str]) -> str | None:
|
||
for key in ("fpc", "FPC"):
|
||
value = row.get(key)
|
||
if value and value.strip():
|
||
return value.strip()
|
||
return None
|
||
|
||
|
||
def catalog_prc(row: dict[str, str]) -> str | None:
|
||
for key in ("prc", "PRC"):
|
||
value = row.get(key)
|
||
if value and value.strip():
|
||
return value.strip()
|
||
return None
|
||
|
||
|
||
def score_match(query: str, tokens: list[str], primary: str | None) -> tuple[int, int, str] | None:
|
||
q = normalize_text(query)
|
||
token_norms = [normalize_text(token) for token in tokens if token]
|
||
primary_norm = normalize_text(primary) if primary else ""
|
||
|
||
if primary_norm == q:
|
||
return (0, 0, primary_norm)
|
||
if q in token_norms:
|
||
return (1, 0, primary_norm)
|
||
if primary_norm and q in primary_norm:
|
||
return (2, len(primary_norm), primary_norm)
|
||
for token in token_norms:
|
||
if q in token:
|
||
return (3, len(token), primary_norm or token)
|
||
return None
|
||
|
||
|
||
def load_catalog_matches(catalog_path: Path, query: str) -> list[MatchRow]:
|
||
suffix = catalog_path.suffix.lower()
|
||
rows: list[dict[str, str]] = []
|
||
if suffix in (".csv", ".tsv"):
|
||
delimiter = "\t" if suffix == ".tsv" else ","
|
||
with catalog_path.open("r", encoding="utf-8-sig", newline="") as fh:
|
||
reader = csv.DictReader(fh, delimiter=delimiter)
|
||
rows.extend(dict(row) for row in reader)
|
||
elif suffix in (".json", ".jsonl"):
|
||
text = catalog_path.read_text(encoding="utf-8")
|
||
if suffix == ".jsonl":
|
||
rows.extend(json.loads(line) for line in text.splitlines() if line.strip())
|
||
else:
|
||
payload = json.loads(text)
|
||
if isinstance(payload, list):
|
||
rows.extend(dict(row) for row in payload)
|
||
elif isinstance(payload, dict) and isinstance(payload.get("items"), list):
|
||
rows.extend(dict(row) for row in payload["items"])
|
||
else:
|
||
raise SystemExit("catalog json must be a list or {items:[...]}")
|
||
else:
|
||
raise SystemExit("catalog must be .csv, .tsv, .json or .jsonl")
|
||
|
||
matches: list[MatchRow] = []
|
||
for row in rows:
|
||
name = catalog_display_name(row)
|
||
fpc = catalog_fpc(row)
|
||
if not name or not fpc:
|
||
continue
|
||
score = score_match(query, [name], name)
|
||
if score is None:
|
||
continue
|
||
matches.append(
|
||
MatchRow(
|
||
fpc=fpc,
|
||
prc=catalog_prc(row),
|
||
primary_name=name,
|
||
secondary_name=None,
|
||
source_id=row.get("source_id") or row.get("id") or row.get("sid"),
|
||
score=score,
|
||
)
|
||
)
|
||
matches.sort(key=lambda row: (row.score, row.fpc))
|
||
return matches
|
||
|
||
|
||
def collect_matches(zip_path: Path, query: str) -> list[MatchRow]:
|
||
root = load_xml(zip_path)
|
||
obj = root.find(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}OBJ")
|
||
if obj is None:
|
||
raise SystemExit("OBJ block missing in input package")
|
||
|
||
matches: list[MatchRow] = []
|
||
for feature in obj.findall(".//{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}CB03"):
|
||
fpc = get_text(feature, "FPC")
|
||
prc = get_text(feature, "PRC")
|
||
if not fpc:
|
||
continue
|
||
tokens = feature_name_tokens(feature)
|
||
primary, secondary = feature_display_name(feature)
|
||
score = score_match(query, tokens, primary)
|
||
if score is None:
|
||
continue
|
||
source_id = None
|
||
loc = feature.find("{http://nlftp.mlit.go.jp/ksj/schemas/ksj-app}LOC")
|
||
if loc is not None:
|
||
source_id = loc.attrib.get("idref")
|
||
matches.append(
|
||
MatchRow(
|
||
fpc=fpc,
|
||
prc=prc,
|
||
primary_name=primary,
|
||
secondary_name=secondary,
|
||
source_id=source_id,
|
||
score=score,
|
||
)
|
||
)
|
||
|
||
matches.sort(key=lambda row: (row.score, row.fpc))
|
||
return matches
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="按渔港名字查找 FPC")
|
||
parser.add_argument("name", help="渔港名字，支持包含匹配")
|
||
parser.add_argument(
|
||
"--source",
|
||
default=FISH_SRC_DEFAULT,
|
||
help="输入渔港原始包，默认 coastline/C09-06.zip",
|
||
)
|
||
parser.add_argument(
|
||
"--catalog",
|
||
default=None,
|
||
help="可选的港名对照表（CSV/TSV/JSON/JSONL）；如果原始包里找不到名字，建议用它",
|
||
)
|
||
parser.add_argument(
|
||
"--exact",
|
||
action="store_true",
|
||
help="只保留精确匹配",
|
||
)
|
||
parser.add_argument(
|
||
"--json",
|
||
action="store_true",
|
||
help="以 JSON 输出",
|
||
)
|
||
parser.add_argument(
|
||
"--max-results",
|
||
type=int,
|
||
default=20,
|
||
help="最多输出多少条候选结果",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
source_path = Path(args.source)
|
||
if not source_path.is_absolute():
|
||
source_path = Path(__file__).resolve().parent / source_path
|
||
if not source_path.exists():
|
||
raise SystemExit(f"missing input: {source_path}")
|
||
|
||
matches: list[MatchRow] = []
|
||
if args.catalog:
|
||
catalog_path = Path(args.catalog)
|
||
if not catalog_path.is_absolute():
|
||
catalog_path = Path(__file__).resolve().parent / catalog_path
|
||
if not catalog_path.exists():
|
||
raise SystemExit(f"missing catalog: {catalog_path}")
|
||
matches = load_catalog_matches(catalog_path, args.name)
|
||
|
||
if not matches:
|
||
matches = collect_matches(source_path, args.name)
|
||
if args.exact:
|
||
q = normalize_text(args.name)
|
||
matches = [row for row in matches if normalize_text(row.primary_name or "") == q]
|
||
|
||
matches = matches[: args.max_results]
|
||
if not matches:
|
||
if args.catalog:
|
||
raise SystemExit(
|
||
f"no FPC matched for name={args.name!r} in catalog={args.catalog!r}"
|
||
)
|
||
raise SystemExit(
|
||
f"no FPC matched for name={args.name!r}; "
|
||
"this source package may not contain human-readable port names, "
|
||
"try --catalog with a port-name mapping file"
|
||
)
|
||
|
||
if args.json:
|
||
print(
|
||
json.dumps(
|
||
[asdict(row) | {"score": list(row.score)} for row in matches],
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
)
|
||
return
|
||
|
||
for idx, row in enumerate(matches, start=1):
|
||
name_bits = [bit for bit in [row.primary_name, row.secondary_name] if bit]
|
||
print(
|
||
f"{idx}. FPC={row.fpc} PRC={row.prc or '-'} "
|
||
f"name={' / '.join(name_bits) if name_bits else '-'} "
|
||
f"source_id={row.source_id or '-'}"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|