56 lines
3.1 KiB
Python
56 lines
3.1 KiB
Python
"""Map only unique canonical names within the same province, then export privately."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import unicodedata
|
|
|
|
def norm(v):
|
|
return ' '.join(''.join(c for c in unicodedata.normalize('NFKD',v).casefold() if not unicodedata.combining(c)).split())
|
|
|
|
p=argparse.ArgumentParser()
|
|
p.add_argument('--preview',required=True)
|
|
p.add_argument('--geography',required=True)
|
|
a=p.parse_args()
|
|
root=Path(a.preview)
|
|
geos=json.loads(Path(a.geography).read_text(encoding='utf-8'))
|
|
byid={r['id']:r for r in geos}
|
|
territories={}
|
|
provinces={}
|
|
subterritories={}
|
|
def province_name(v):
|
|
n=norm(v)
|
|
return 'maputo provincia' if n=='maputo' else n
|
|
for g in geos:
|
|
if g['location_type']=='province':provinces.setdefault(norm(g['name']),[]).append(g['id'])
|
|
if g['location_type'] not in ('district','admin_post','locality'):continue
|
|
parent=byid.get(g['parent_id']);visited=set()
|
|
while parent and parent['location_type']!='province' and parent['id'] not in visited:
|
|
visited.add(parent['id']);parent=byid.get(parent['parent_id'])
|
|
if not parent:continue
|
|
target=territories if g['location_type']=='district' else subterritories
|
|
target.setdefault((norm(parent['name']),norm(g['name'])),[]).append(g['id'])
|
|
db=sqlite3.connect(root/'supply-preview.sqlite')
|
|
db.row_factory=sqlite3.Row
|
|
resolution={}
|
|
with (root/'observations.jsonl').open('w',encoding='utf-8',newline='\n') as f:
|
|
for row in db.execute('select * from records order by source_hash,sheet,row_number'):
|
|
r=dict(row)
|
|
key=(province_name(r['province']),norm(r['district']))
|
|
hits=territories.get(key,[])
|
|
lower_level=False
|
|
if not hits:
|
|
hits=subterritories.get(key,[])
|
|
lower_level=bool(hits)
|
|
province_hits=provinces.get(province_name(r['province']),[])
|
|
province_only=not r['district'] and len(province_hits)==1
|
|
r['territory_id']=hits[0] if len(hits)==1 else (province_hits[0] if province_only else None)
|
|
r['territory_resolution']=('exact_normalized_province_subterritory' if lower_level else 'exact_normalized_province_district') if len(hits)==1 else ('province_only_district_unknown' if province_only else ('ambiguous' if hits else 'unresolved'))
|
|
r['identity_conflict']=bool(r['identity_conflict'])
|
|
resolution[r['territory_resolution']]=resolution.get(r['territory_resolution'],0)+1
|
|
f.write(json.dumps(r,ensure_ascii=False)+'\n')
|
|
result={'rows':db.execute('select count(*) from records').fetchone()[0],'territory_resolution':resolution,'sha256':hashlib.file_digest((root/'observations.jsonl').open('rb'),'sha256').hexdigest(),'mapping':'case/accent/whitespace; exact district in province, otherwise unique named subordinate territory within that province; source label Maputo distinct from Maputo Cidade maps explicitly to Maputo Província; city names retain canonical administrative level, no fuzzy matching','publication':False}
|
|
(root/'export.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
|
|
print(json.dumps(result,ensure_ascii=False))
|