添加全国三层生成与港名查FPC工具
This commit is contained in:
OpenAI Codex
282
find_fpc_by_port_name.py
Normal file
282
find_fpc_by_port_name.py
Normal file
@@ -0,0 +1,282 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user