1"""Phase 2.3 — Store master with region recovery and geographic-quality flags.23Input : data/interim/stores_raw.parquet, data/interim/listings_clean.parquet, optional data/external/<regions layer>4Output: data/interim/stores_master.parquet, experiments/store_master.json56Region logic (documented in reports/phase2_data_audit.md):7 region_db : Fabri-Ka's region (directory source / site mentions / FSA map), blank for 22 % of live stores8 region_pip : administrative region containing the FSA centroid (lat, lng) — point-in-polygon against the official9 MERN layer (if available; precision = neighbourhood, never the civic address)10 region_final : region_db if present, else region_pip, else 'Inconnue'11 region_source : db | pip | unknown12 region_conflict: region_db and region_pip both present and different (kept, flagged; sensitivity analyses drop them)13 phone_shared : phone number shared by >= 3 live stores (directory contamination; phone never used for geography)14"""15from __future__ import annotations16import glob, json17import numpy as np, pandas as pd18from src.common import ROOT, P, REGIONS, get_logger, ledger, register_output, write_json, BASE_PROC, BASE_EXP1920log = get_logger("clean.stores")212223def load_regions_layer():24 cands = sorted(glob.glob(str(ROOT / "data" / "external" / "**" / "*mern*.gpkg"), recursive=True)) + sorted(glob.glob(str(ROOT / "data" / "external" / "**" / "*.gpkg"), recursive=True) +25 glob.glob(str(ROOT / "data" / "external" / "**" / "*.shp"), recursive=True) +26 glob.glob(str(ROOT / "data" / "external" / "**" / "*.geojson"), recursive=True))27 try:28 import geopandas as gpd29 except ImportError:30 return None, None31 for c in cands:32 try:33 g = gpd.read_file(c)34 except Exception:35 continue36 # find a name column whose values cover the 17 regions37 for col in g.columns:38 if g[col].dtype == object:39 vals = set(str(v).replace("--", "–") for v in g[col].dropna().unique())40 hit = len([r for r in REGIONS if r in vals or r.replace("–", "-") in vals])41 if hit >= 15:42 g["geometry"] = g.geometry.make_valid(); g = g.to_crs(4326)43 g["_region"] = g[col].astype(str).str.replace("--", "–").map(lambda v: next((r for r in REGIONS if r == v or r.replace("–", "-") == v), v))44 return g[["_region", "geometry"]].dissolve(by="_region").reset_index(), c45 return None, None464748def city_region_lookup():49 """Map the zippopotam sector label (e.g. 'Granby Central', 'Estrie-Ouest (Fulford)') to an administrative region via the50 official MERN municipality table; only unambiguous municipality names are used."""51 import unicodedata, re52 f = ROOT / "data" / "external" / "mern_municipalites_attributes.csv"53 if not f.exists():54 return lambda c: None55 mu = pd.read_csv(f)56 norm = lambda x: re.sub(r"[^a-z]", "", unicodedata.normalize("NFKD", str(x)).encode("ascii", "ignore").decode().lower())57 mu["k"] = mu.MUS_NM_MUN.map(norm)58 amb = mu.groupby("k").MUS_NM_REG.nunique(); table = mu.drop_duplicates("k").set_index("k").MUS_NM_REG; table = table[amb[table.index] == 1]59 def look(c):60 if not isinstance(c, str) or not c: return None61 base = re.sub(r"\(.*?\)", "", c); base = re.sub(r"\b(Central|West|East|North|South|Northwest|Northeast|Southwest|Southeast|Centre|Ouest|Est|Nord|Sud)\b", "", base).strip(" -")62 k = norm(base) # the parenthesised town of rural FSAs is only a representative place, never used63 return table[k] if k and k in table.index else None64 return look656667def main():68 st = pd.read_parquet(P["interim"] / "stores_raw.parquet")69 li = pd.read_parquet(P["interim"] / "listings_clean.parquet")70 live = st[(st.enabled == 1) & (st.product_count > 0)].copy()71 live["region_db"] = live.region.fillna("").replace({"": None})72 live["region_db"] = live.region_db.where(live.region_db.isin(REGIONS), None)7374 layer, layer_path = load_regions_layer()75 live["region_pip"] = None76 if layer is not None:77 import geopandas as gpd78 pts = live[live.lat.notna() & live.lng.notna()]79 gp = gpd.GeoDataFrame(pts[["id"]], geometry=gpd.points_from_xy(pts.lng, pts.lat), crs=4326)80 j = gpd.sjoin(gp, layer, how="left", predicate="within")81 j = j.drop_duplicates("id")82 live = live.merge(j[["id", "_region"]], on="id", how="left")83 live["region_pip"] = live["_region"]; live = live.drop(columns=["_region"])84 log.info("regions layer %s — PIP assigned for %s stores", layer_path, int(live.region_pip.notna().sum()))85 else:86 log.warning("no administrative-region layer found under data/external — region_pip left empty")8788 look = city_region_lookup(); urban = live.postal_prefix.fillna("").str.len().eq(3) & ~live.postal_prefix.fillna("").str[1].eq("0")89 live["region_city"] = live.city.where(urban).map(look) # municipality lookup only for urban FSAs (second character != 0)90 live["region_city"] = live.region_city.where(live.region_city.isin(REGIONS), None)91 # Amendment A-001 (2026-11-01): official administrative geography first — municipality lookup, then FSA-centroid92 # point-in-polygon, then Fabri-Ka's directory/site label (directory regions are often tourism/brand regions, e.g.93 # 'Cantons-de-l'Est' members in Montérégie labelled Estrie). region_db kept for the db-first sensitivity variant.94 # A-001 (final form): municipality lookup (urban FSAs, exact) > Fabri-Ka label > FSA-centroid point-in-polygon95 # (centroids of rural FSAs are unreliable near region borders); pip-first kept as a sensitivity variant.96 live["region_final"] = live.region_city.fillna(live.region_db).fillna(live.region_pip).fillna("Inconnue")97 live["region_source"] = np.where(live.region_city.notna(), "city", np.where(live.region_db.notna(), "db", np.where(live.region_pip.notna(), "pip", "unknown")))98 live["region_dbfirst"] = live.region_pip.fillna(live.region_city).fillna(live.region_db).fillna("Inconnue") # name kept for the pipeline: this is the PIP-first sensitivity variant99 live["region_conflict"] = live.region_db.notna() & (live.region_final != "Inconnue") & (live.region_db != live.region_final)100 live["has_geo"] = live.lat.notna()101 ph = live.phone.fillna("").str.replace(r"\D", "", regex=True)102 cnt = ph[ph != ""].value_counts()103 live["phone_shared"] = ph.map(cnt).fillna(0).ge(3)104 live["origin_class"] = live.origin_class.fillna("E")105 live["certified"] = live.n_certifications.fillna(0) > 0106 live["cert_adq"] = live.cert_labels.fillna("").str.contains("Aliments du Québec", regex=False)107 live["cert_lpq"] = live.cert_labels.fillna("").str.contains("Produits du Québec", regex=False)108 live["cert_bio"] = live.cert_labels.fillna("").str.contains("Biologique|CARTV", regex=True)109 live["store_kind"] = live.store_kind.fillna("").replace({"": "standard"})110 live["language"] = live.language.fillna("").replace({"": "unknown", "None": "unknown"})111 live["n_sources"] = live.discovery_sources_list.map(lambda s: len(json.loads(s)))112 live["first_source"] = live.discovery_sources_list.map(lambda s: (json.loads(s) or ["none"])[0])113114 a = li[li.f_analysis_assort]115 agg = a.groupby("store_id").agg(n_listings=("uid", "size"), n_categories=("category", "nunique"),116 share_autre=("category", lambda s: (s == "autre").mean()),117 median_price=("price", "median"), share_available=("available", lambda s: (s == 1).mean()),118 share_on_sale=("on_sale", "mean"), first_listing_utc=("first_seen_utc", "min"),119 last_listing_utc=("first_seen_utc", "max"))120 live = live.merge(agg, left_on="id", right_index=True, how="left")121 live["n_listings"] = live.n_listings.fillna(0).astype(int)122 top = a.groupby(["store_id", "category"]).size().reset_index(name="n").sort_values(["store_id", "n"], ascending=[True, False]).drop_duplicates("store_id")123 live = live.merge(top.rename(columns={"category": "main_category"})[["store_id", "main_category"]], left_on="id", right_on="store_id", how="left").drop(columns=["store_id"])124125 keep = ["id", "name", "platform", "city", "region_db", "region_pip", "region_city", "region_dbfirst", "region_final", "region_source", "region_conflict", "has_geo", "lat", "lng",126 "postal_prefix", "phone_shared", "origin_class", "origin_confidence", "origin_evidence", "certified", "cert_adq", "cert_lpq", "cert_bio",127 "cert_labels", "store_kind", "language", "n_sources", "first_source", "reg_status", "product_count", "n_listings", "n_categories",128 "share_autre", "median_price", "share_available", "share_on_sale", "main_category", "first_listing_utc", "last_listing_utc", "last_sync_utc", "fail_streak"]129 out = live[keep].copy()130 out.to_parquet(P["interim"] / "stores_master.parquet", index=False)131 register_output(P["interim"] / "stores_master.parquet", "clean.stores")132 summ = {"live_stores": len(out), "regions_layer": layer_path, "region_source": out.region_source.value_counts().to_dict(),133 "region_conflicts": int(out.region_conflict.sum()), "unknown_region_listings": int(out.loc[out.region_final == "Inconnue", "n_listings"].sum()),134 "phone_shared_stores": int(out.phone_shared.sum()), "certified_stores": int(out.certified.sum()),135 "region_final_counts": out.region_final.value_counts().to_dict(),136 "agreement_db_vs_pip": (round(float((out.region_db == out.region_pip)[out.region_db.notna() & out.region_pip.notna()].mean()), 4) if out.region_pip.notna().any() else None),137 "agreement_city_vs_pip": (round(float((out.region_city == out.region_pip)[out.region_city.notna() & out.region_pip.notna()].mean()), 4) if out.region_city.notna().any() else None),138 "agreement_db_vs_final": round(float((out.region_db == out.region_final)[out.region_db.notna()].mean()), 4)}139 write_json(summ, BASE_EXP / "store_master.json", "clean.stores")140 ledger("clean.stores", **summ)141 log.info(json.dumps(summ, ensure_ascii=False))142143144if __name__ == "__main__":145 main()146