#!/usr/bin/env python3 """Structured sweep: resolve remaining companies via structured street/city queries and named-feature (building / ngõ) queries against a Nominatim instance. Complements v4's free-form waterfall (which fails on ambiguous street names): a candidate is accepted only if it is unambiguous — - exactly one result matches the queried street/feature name, and - when the address names a ward/district, that candidate's display name contains a matching ward/district token. Usage (local instance): GEOCODE_BASE_URL=http://127.0.0.1:8088 GEOCODE_GAP=0.2 \ python3 geocode_structured_sweep.py [--limit N] Accepted rows are written with method="structured" into the same progress/CSV files v4 uses. Every examined row is logged to forms/v9_sweep_review.json. """ import argparse import json import os import re import sys import time import urllib.parse import urllib.request sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import geocode_companies_v4 as g4 REVIEW_FILE = "forms/v9_sweep_review.json" METHOD = "structured" _PARENS = re.compile(r"\([^)]*\)") _ADMIN_SEG = re.compile( r"(phường|quận|xã|huyện|thị trấn|thành phố|khu chế xuất|khu công nghiệp|" r"khu phố|khu công nghệ|ấp\b|tỉnh\b|tp\b|kcn|kcx|q\.|p\.|h\.)", re.I) _UNIT_LEAD = re.compile( r"^\s*(?:số|lô|tầng|lầu|phòng|căn|kiốt|kiosk|tt|thửa)\s*[\w./-]*\s*", re.I) _LEAD_JUNK = re.compile(r"^[\s\-–+0-9/.]+") _UNIT_WORD = re.compile(r"\b(?:tầng|lầu|phòng|căn|kiốt|kiosk|thửa)\b", re.I) _STREET_MARK = re.compile(r"\b(?:đường|phố|đ\.)\s+", re.I) _BUILDING = re.compile(r"\b(?:tòa nhà|toà nhà|building|cao ốc|tòa|toà)\s+(.+)", re.I) _ALLEY = re.compile(r"\b(?:ngõ|hẻm|ngách)\s*([0-9]+[a-z0-9/]*)\s+(?:phố\s+)?(.+)", re.I) _ALLEY_LEAD = re.compile(r"^(?:ngõ|hẻm|ngách)\s*\S+\s+", re.I) _NUM_STREET = re.compile(r"^(?:đ\.|đường)\s+(số\s+[\w.]+.*)$", re.I) _SHORT_CODE = re.compile(r"^[a-z]{1,3}\d+[a-z0-9]*$") _DISTRICT = re.compile(r"^(?:q\.|quận|h\.|huyện)\s*(\d+|[A-Za-zÀ-ỹ][\wÀ-ỹ\s]*)$", re.I) _PROVINCE = re.compile(r"^(?:tỉnh)\s+([\wÀ-ỹ\s]+)$", re.I) _DIGITS = re.compile(r"^\d+[a-z]?$", re.I) CITY_PATTERNS = [ (re.compile(r"hồ\s*chí\s*minh|hcm|sài gòn|tphcm", re.I), "Thành phố Hồ Chí Minh"), (re.compile(r"hà nội", re.I), "Hà Nội"), (re.compile(r"đà nẵng", re.I), "Đà Nẵng"), (re.compile(r"huế", re.I), "Huế"), (re.compile(r"nha trang", re.I), "Nha Trang"), (re.compile(r"biên hòa|đồng nai", re.I), "Biên Hòa"), (re.compile(r"tây ninh", re.I), "Tây Ninh"), (re.compile(r"cà mau", re.I), "Cà Mau"), ] def norm(s): return re.sub(r"[^a-z0-9]+", " ", g4._fold(s or "")).strip() def _street_candidate(seg): """-> (street, explicit) | None. Explicit = numbered/marked street segment.""" m = _NUM_STREET.match(seg) if m: return "Đường " + m.group(1).strip(), True explicit = bool(_STREET_MARK.search(seg)) or bool(re.match(r"^\s*\d", seg)) s = _STREET_MARK.sub("", seg) s = _UNIT_LEAD.sub("", s) s = _LEAD_JUNK.sub("", s) if _ALLEY_LEAD.match(s): s = _ALLEY_LEAD.sub("", s) explicit = True s = _LEAD_JUNK.sub("", s).strip() if not s or _DIGITS.match(s): return None if not (len(norm(s)) >= 4 or _SHORT_CODE.match(norm(s))): return None return s, explicit _ADMIN_PREFIX = re.compile( r"^(?:phuong|p\d*|quan|q\d*|xa|huyen|h\d*|tp|thanh pho|thi tran|tinh|" r"khu che xuat|khu cong nghiep|khu cong nghe|khu pho|kcn|kcx|ap)\s+") _ADMIN_BARE = re.compile(r"^(?:p|q|h|tp)\s+\S", re.I) _CITY_JUNK = re.compile(r"\b(?:tphcm|tp hcm|hcm|viet nam|vn)\b") def _admin_name(seg): """Normalized ward/district name with its admin prefix stripped, or None. City/province-level segments (thành phố / tỉnh / tp.) are not ward names.""" s = norm(seg) for _ in range(4): if re.match(r"^(?:thanh pho|tp|tinh)\b", s): return None m = _ADMIN_PREFIX.match(s) if not m: break s = s[m.end():].strip() s = re.sub(r"\s+", " ", _CITY_JUNK.sub(" ", s)).strip() if not s or re.fullmatch(r"[\d/]+", s) or len(s) < 4: return None return s def parse_address(addr): """-> (city, county, state, street, feature, admin_names); feature = (kind, name) | None; admin_names = normalized ward/district names.""" addr = _PARENS.sub(" ", addr or "") segments = [s.strip() for s in re.split(r"[,\n]|\s+-\s+", addr) if s.strip()] city = county = state = None for seg in segments: for pat, name in CITY_PATTERNS: if pat.search(seg): city = name break if city: break if city is None: for seg in segments: m = _DISTRICT.match(seg) if m: prefix = "Huyện" if re.match(r"^\s*(?:h\.|huyện)", seg, re.I) else "Quận" county = f"{prefix} {m.group(1).strip()}" break else: for seg in segments: m = _PROVINCE.match(seg) if m: state = m.group(1).strip() break admin_names = [] street = feature = None generic = None for seg in segments: if _ADMIN_SEG.search(seg) or _ADMIN_BARE.match(seg): name = _admin_name(seg) if name: admin_names.append(name) continue if any(pat.search(seg) for pat, _ in CITY_PATTERNS): continue if feature is None: m = _BUILDING.search(seg) if m and len(norm(m.group(1))) >= 4: feature = ("building", m.group(1).strip()) continue if feature is None: m = _ALLEY.search(seg) if m: feature = ("alley", f"ngõ {m.group(1)} {m.group(2).strip()}") continue if _UNIT_WORD.search(seg) and not _STREET_MARK.search(seg): continue cand = _street_candidate(seg) if cand: if cand[1] and street is None: street = cand[0] elif generic is None: generic = cand[0] if street is None: street = generic return city, county, state, street, feature, admin_names def evaluate(rows, needle, admin_names, region_value, relax_tokens=frozenset()): """rows -> (hit | None, reason). Accept only unambiguous, region-consistent hits.""" region_tokens = {t for t in norm(region_value or "").split() if len(t) >= 2} cands = [] for r in rows: d = norm(r.get("display_name", "")) if not needle or needle not in d: continue if region_tokens and not region_tokens <= set(d.split()): continue cands.append(r) if admin_names: warded = [r for r in cands if any(n in norm(r.get("display_name", "")) for n in admin_names)] else: warded = cands if len(warded) == 1: return warded[0], ("unique_ward_match" if admin_names else "unique_match") if len(warded) > 1: return None, "ambiguous" if len(cands) == 1 and relax_tokens & set(norm(cands[0].get("display_name", "")).split()): return cands[0], "relaxed_unique_match" return None, ("ward_mismatch" if cands else "no_match") class Sweeper: def __init__(self): self.client = g4.Nominatim() def _get(self, params): params = dict(params, format="jsonv2", limit=10, addressdetails=1, countrycodes=g4.COUNTRY_CODE) url = f"{g4.NOMINATIM_URL}/search?{urllib.parse.urlencode(params)}" req = urllib.request.Request(url, headers={"User-Agent": g4.USER_AGENT}) for attempt in (1, 2): self.client._pace() self.client.requests += 1 try: with urllib.request.urlopen(req, timeout=g4.REQUEST_TIMEOUT) as resp: return json.loads(resp.read().decode()) except Exception as e: if attempt == 2: print(f" request failed: {e}", file=sys.stderr, flush=True) return [] time.sleep(2) return [] def structured(self, street, region_key, region_value): p = {"street": street} if region_key: p[region_key] = region_value return self._get(p) def freeform(self, q): return self._get({"q": q}) def slim(rows): return [{"lat": r.get("lat"), "lon": r.get("lon"), "display_name": r.get("display_name", "")[:140]} for r in rows] def sweep_row(s, location): """-> (hit, variant_label, reason, attempts)""" city, county, state, street, feature, admin_names = parse_address(location) attempts = [] region_key, region_value = (("city", city) if city else ("county", county) if county else ("state", state) if state else (None, None)) extra = g4._distinctive_tokens(location) for src in (street, region_value): extra -= set(norm(src or "").split()) for n in admin_names: extra -= set(n.split()) if feature: kind, name = feature parts = [x for x in (name, street, region_value) if x] if len(parts) >= 2: q = ", ".join(parts) rows = s.freeform(q) attempts.append({"query": q, "n": len(rows), "results": slim(rows)}) relax = extra | {t for t in norm(street or "").split() if len(t) >= 4} hit, why = evaluate(rows, norm(name), admin_names, region_value, relax) if hit: return hit, f"{kind} query: {q} ({why})", why, attempts if street: label = f"street={street} {region_key or 'area'}={region_value or '-'}" rows = s.structured(street, region_key, region_value) attempts.append({"query": label, "n": len(rows), "results": slim(rows)}) hit, why = evaluate(rows, norm(street), admin_names, region_value, extra) if hit: return hit, f"{label} ({why})", why, attempts reason = attempts[-1]["query"] if attempts else "unparsed" return None, None, reason, attempts def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--limit", type=int, default=None) args = parser.parse_args() with open(g4.INPUT_FILE, "r", encoding="utf-8") as f: companies = json.load(f) results = g4.load_results() pending = [c for c in companies if not g4.is_resolved(results.get(c.get("id")))] if args.limit is not None: pending = pending[:args.limit] print(f"{len(companies)} companies | {len(pending)} pending | " f"base URL {g4.NOMINATIM_URL}") s = Sweeper() started = time.time() review, accepted = [], [] for i, company in enumerate(pending, 1): cid = company.get("id", "") name = company.get("shortname", "?") location = company.get("location", "") entry = {"id": cid, "shortname": name, "location": location} if not g4.restrict_to_vn(location): entry["outcome"] = "foreign_skipped" review.append(entry) print(f"[{i}/{len(pending)}] {name}: foreign address, skipped") continue hit, label, reason, attempts = sweep_row(s, location) entry["attempts"] = attempts if hit: row = results[cid] row["latitude"], row["longitude"] = float(hit["lat"]), float(hit["lon"]) row["method"] = METHOD row["matched_variant"] = label row["matched_address"] = hit.get("display_name", "") entry["outcome"] = "accepted" entry["hit"] = {"lat": hit["lat"], "lon": hit["lon"], "display_name": hit.get("display_name", "")} accepted.append(name) print(f"[{i}/{len(pending)}] {name}: -> ({row['latitude']:.6f}, " f"{row['longitude']:.6f}) [{label}]") else: entry["outcome"] = reason print(f"[{i}/{len(pending)}] {name}: {reason} " f"({sum(a['n'] for a in attempts)} results over " f"{len(attempts)} queries)") review.append(entry) g4.save_progress(results) g4.save_csv(companies, results) with open(REVIEW_FILE, "w", encoding="utf-8") as f: json.dump({"pending": len(pending), "accepted": len(accepted), "rows": review}, f, ensure_ascii=False, indent=1) resolved = sum(1 for c in companies if g4.is_resolved(results.get(c.get("id")))) print(f"\nDone: {len(accepted)} accepted this run | {resolved}/{len(companies)} " f"resolved | {s.client.requests} requests | " f"{(time.time() - started) / 60:.1f} min | review -> {REVIEW_FILE}") if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\ninterrupted - progress saved per row, rerun to resume", file=sys.stderr) sys.exit(130)