path: root/scripts/make-park-match-review.py

"""OSMの公園ポリゴンのうち、市の公園一覧と一致しなかったものをレビュー用に出力する。

出力:
  reference/park-matching-review.csv      … Excelで開ける一覧（UTF-8 BOM）
  reference/park-matching-review.geojson  … 未一致ポリゴン（QGIS用）
  reference/kodaira-parks-points.geojson  … 市の公園の位置（QGIS用）
  src/data/park-name-overrides.json       … OSM ID→市の公園名の手動上書き（未作成時のみ）

CSVやQGISで確認したうえで、JSONの "city" を埋めて `npm run osm:update` を
実行すると照合結果に反映される。市の公園一覧にない施設・広場は
"status": "unlisted" にすると、候補の提示を止めて「一覧外」として記録する。
"""

import csv
import difflib
import json
import math
import os
import sys
import time
import unicodedata

GEOJSON = "src/data/kodaira-parks-osm.json"
OURS = "src/data/kodaira-parks.json"
OUT_CSV = "reference/park-matching-review.csv"
OUT_REVIEW_GEOJSON = "reference/park-matching-review.geojson"
OUT_POINTS_GEOJSON = "reference/kodaira-parks-points.geojson"
OUT_OVERRIDES = "src/data/park-name-overrides.json"

CSV_FIELDS = [
    "osm_type",
    "osm_id",
    "OSM名",
    "OSM URL",
    "緯度",
    "経度",
    "面積㎡",
    "頂点数",
    "候補1",
    "距離1m",
    "候補2",
    "距離2m",
    "候補3",
    "距離3m",
    "名前類似候補",
    "類似度",
    "類似候補距離m",
    "推定",
    "判定",
    "メモ",
]


def norm(s):
    s = unicodedata.normalize("NFKC", str(s)).replace(" ", "").replace("　", "")
    return s.replace("ヶ", "ケ").replace("ツ", "ッ")


def rings(geom):
    if geom["type"] == "Polygon":
        return [geom["coordinates"][0]]
    if geom["type"] == "MultiPolygon":
        return [poly[0] for poly in geom["coordinates"]]
    return []


def center(geom):
    pts = [p for ring in rings(geom) for p in ring]
    if not pts:
        return None, None
    lats = [p[1] for p in pts]
    lons = [p[0] for p in pts]
    return (min(lats) + max(lats)) / 2, (min(lons) + max(lons)) / 2


def area_m2(geom):
    total = 0.0
    for ring in rings(geom):
        if len(ring) < 4:
            continue
        lat0 = sum(p[1] for p in ring) / len(ring)
        kx = 111320 * math.cos(math.radians(lat0))
        ky = 111320
        s = 0.0
        for (x1, y1), (x2, y2) in zip(ring, ring[1:] + ring[:1]):
            s += (x1 * kx) * (y2 * ky) - (x2 * kx) * (y1 * ky)
        total += abs(s) / 2
    return total


def dist_m(lat1, lon1, lat2, lon2):
    dy = (lat1 - lat2) * 111320
    dx = (lon1 - lon2) * 111320 * math.cos(math.radians((lat1 + lat2) / 2))
    return math.hypot(dx, dy)


def main():
    osm = json.load(open(GEOJSON, encoding="utf-8"))
    ours = json.load(open(OURS, encoding="utf-8"))
    parks = []
    for r in ours:
        try:
            lat = float(r.get("緯度") or "")
            lon = float(r.get("経度") or "")
        except ValueError:
            continue
        parks.append({"name": r["公園名"], "lat": lat, "lon": lon, "address": r.get("住所", "")})
    park_names = {q["name"] for q in parks}

    decisions = {}
    if os.path.exists(OUT_OVERRIDES):
        raw = json.load(open(OUT_OVERRIDES, encoding="utf-8"))
        decisions = {k: v for k, v in raw.items() if isinstance(v, dict)}

    unmatched = [f for f in osm["features"] if not f["properties"].get("matched")]
    rows = []
    review_features = []
    for f in unmatched:
        p = f["properties"]
        lat, lon = center(f["geometry"])
        if lat is None:
            continue
        area = area_m2(f["geometry"])
        pts = sum(len(r) for r in rings(f["geometry"]))
        near = sorted(
            (
                {
                    "name": q["name"],
                    "dist": dist_m(lat, lon, q["lat"], q["lon"]),
                    "similar": difflib.SequenceMatcher(None, norm(p.get("name", "")), norm(q["name"])).ratio(),
                }
                for q in parks
            ),
            key=lambda x: x["dist"],
        )[:3]
        same = sorted(
            (
                {
                    "name": q["name"],
                    "dist": dist_m(lat, lon, q["lat"], q["lon"]),
                    "similar": difflib.SequenceMatcher(None, norm(p.get("name", "")), norm(q["name"])).ratio(),
                }
                for q in parks
                if p.get("name")
            ),
            key=lambda x: -x["similar"],
        )
        best_similar = same[0] if same and same[0]["similar"] >= 0.6 else None

        suggestion = ""
        if near[0]["dist"] <= 30:
            suggestion = near[0]["name"]
        elif best_similar and best_similar["similar"] >= 0.75 and best_similar["dist"] <= 300:
            suggestion = best_similar["name"]
        elif best_similar and best_similar["dist"] <= 100:
            suggestion = best_similar["name"]

        memo = []
        if area < 5:
            memo.append("面積が極小（形状エラーの可能性）")
        if near[0]["dist"] <= 30:
            memo.append("最寄り公園の30m以内（同一の可能性）")
        if p.get("name") and best_similar:
            memo.append(f"名前が類似: {best_similar['name']} ({best_similar['similar']:.2f})")

        key = f"{p.get('osm_type', '')}/{p.get('osm_id', '')}"
        decision = decisions.get(key, {})
        decided_city = str(decision.get("city", "") or "").strip()
        status = str(decision.get("status", "") or "").strip()
        invalid_city = bool(decided_city) and decided_city not in park_names

        if invalid_city:
            memo.append(f"上書き先が一覧にありません: {decided_city}")
            decided_city = ""
        if decided_city:
            suggestion = decided_city
            memo.append("手動で対応済み")
        elif status == "unlisted":
            suggestion = ""
            memo = [m for m in memo if not m.startswith("名前が類似")]
            memo.append("一覧外（手動確認済み）")

        if decided_city:
            verdict = f"対応済み: {decided_city}"
        elif status == "unlisted":
            verdict = "一覧外"
        elif invalid_city:
            verdict = "要確認: 一覧にない名前"
        else:
            verdict = ""

        row = {
            "osm_type": p.get("osm_type", ""),
            "osm_id": p.get("osm_id", ""),
            "OSM名": p.get("name", ""),
            "OSM URL": f"https://www.openstreetmap.org/{p.get('osm_type', '')}/{p.get('osm_id', '')}",
            "緯度": f"{lat:.6f}",
            "経度": f"{lon:.6f}",
            "面積㎡": f"{area:.1f}",
            "頂点数": pts,
            "候補1": near[0]["name"] if len(near) > 0 else "",
            "距離1m": f"{near[0]['dist']:.0f}" if len(near) > 0 else "",
            "候補2": near[1]["name"] if len(near) > 1 else "",
            "距離2m": f"{near[1]['dist']:.0f}" if len(near) > 1 else "",
            "候補3": near[2]["name"] if len(near) > 2 else "",
            "距離3m": f"{near[2]['dist']:.0f}" if len(near) > 2 else "",
            "名前類似候補": best_similar["name"] if best_similar else "",
            "類似度": f"{best_similar['similar']:.2f}" if best_similar else "",
            "類似候補距離m": f"{best_similar['dist']:.0f}" if best_similar else "",
            "推定": suggestion,
            "判定": verdict,
            "メモ": " / ".join(memo),
        }
        rows.append(row)
        review_features.append(
            {
                "type": "Feature",
                "properties": {
                    "osm": f"{row['osm_type']}/{row['osm_id']}",
                    "osm_name": row["OSM名"],
                    "area_m2": round(area, 1),
                    "nearest": row["候補1"],
                    "dist_m": row["距離1m"],
                    "similar": row["名前類似候補"],
                    "similar_ratio": row["類似度"],
                    "similar_dist_m": row["類似候補距離m"],
                    "status": status or ("mapped" if decided_city else ""),
                    "memo": row["メモ"],
                },
                "geometry": f["geometry"],
            }
        )

    rows.sort(key=lambda r: (not r["OSM名"], float(r["距離1m"] or 1e9), r["OSM名"]))
    os.makedirs(os.path.dirname(OUT_CSV), exist_ok=True)
    out_csv = OUT_CSV
    try:
        fh = open(out_csv, "w", encoding="utf-8-sig", newline="")
    except PermissionError:
        out_csv = OUT_CSV.replace(".csv", f"-{time.strftime('%H%M%S')}.csv")
        print(f"warning: {OUT_CSV} is locked; wrote {out_csv}", file=sys.stderr)
        fh = open(out_csv, "w", encoding="utf-8-sig", newline="")
    with fh:
        w = csv.DictWriter(fh, fieldnames=CSV_FIELDS)
        w.writeheader()
        w.writerows(rows)

    json.dump(
        {"type": "FeatureCollection", "features": review_features},
        open(OUT_REVIEW_GEOJSON, "w", encoding="utf-8"),
        ensure_ascii=False,
        separators=(",", ":"),
    )
    points = []
    for q in parks:
        points.append(
            {
                "type": "Feature",
                "properties": {"name": q["name"], "address": q["address"]},
                "geometry": {"type": "Point", "coordinates": [q["lon"], q["lat"]]},
            }
        )
    json.dump(
        {"type": "FeatureCollection", "features": points},
        open(OUT_POINTS_GEOJSON, "w", encoding="utf-8"),
        ensure_ascii=False,
        separators=(",", ":"),
    )

    if not os.path.exists(OUT_OVERRIDES):
        data = {
            f"{r['osm_type']}/{r['osm_id']}": {
                "osm_name": r["OSM名"],
                "suggested": r.get("推定", ""),
                "status": "",
                "city": "",
                "candidates": [c for c in (r["候補1"], r["候補2"], r["候補3"]) if c],
            }
            for r in rows
        }
        with open(OUT_OVERRIDES, "w", encoding="utf-8") as fh:
            json.dump(data, fh, ensure_ascii=False, indent=2)
            fh.write("\n")
        print(f"overrides template -> {OUT_OVERRIDES}")
    else:
        print(f"overrides exists (not overwritten) -> {OUT_OVERRIDES}")

    named = [r for r in rows if r["OSM名"]]
    print(f"unmatched {len(rows)} (named {len(named)}, unnamed {len(rows) - len(named)})")
    print(f"  csv: {out_csv}")
    print(f"  geojson: {OUT_REVIEW_GEOJSON} / {OUT_POINTS_GEOJSON}")
    print(f"  near<=30m: {sum(1 for r in rows if r['距離1m'] and float(r['距離1m']) <= 30)}")
    print(f"  suggested: {sum(1 for r in rows if r.get('推定'))}")
    print(f"  area<5m2: {sum(1 for r in rows if float(r['面積㎡']) < 5)}")
    print("named:")
    for r in named:
        print(
            f"  {r['osm_type']}/{r['osm_id']} {r['OSM名']} -> "
            f"類似 {r['名前類似候補'] or '-'} ({r['類似度'] or '-'}, {r['類似候補距離m'] or '-'}m) / "
            f"最寄り {r['候補1']} {r['距離1m']}m"
        )


if __name__ == "__main__":
    main()