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_json1920log = get_logger("clean.stores")212223def load_regions_layer():24 cands = 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(map(str, 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 = g.to_crs(4326)43 g["_region"] = g[col].astype(str).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 main():49 st = pd.read_parquet(P["interim"] / "stores_raw.parquet")50 li = pd.read_parquet(P["interim"] / "listings_clean.parquet")51 live = st[(st.enabled == 1) & (st.product_count > 0)].copy()52 live["region_db"] = live.region.fillna("").replace({"": None})53 live["region_db"] = live.region_db.where(live.region_db.isin(REGIONS), None)5455 layer, layer_path = load_regions_layer()56 live["region_pip"] = None57 if layer is not None:58 import geopandas as gpd59 pts = live[live.lat.notna() & live.lng.notna()]60 gp = gpd.GeoDataFrame(pts[["id"]], geometry=gpd.points_from_xy(pts.lng, pts.lat), crs=4326)61 j = gpd.sjoin(gp, layer, how="left", predicate="within")62 j = j.drop_duplicates("id")63 live = live.merge(j[["id", "_region"]], on="id", how="left")64 live["region_pip"] = live["_region"]; live = live.drop(columns=["_region"])65 log.info("regions layer %s — PIP assigned for %s stores", layer_path, int(live.region_pip.notna().sum()))66 else:67 log.warning("no administrative-region layer found under data/external — region_pip left empty")6869 live["region_final"] = live.region_db.fillna(live.region_pip).fillna("Inconnue")70 live["region_source"] = np.where(live.region_db.notna(), "db", np.where(live.region_pip.notna(), "pip", "unknown"))71 live["region_conflict"] = live.region_db.notna() & live.region_pip.notna() & (live.region_db != live.region_pip)72 live["has_geo"] = live.lat.notna()73 ph = live.phone.fillna("").str.replace(r"\D", "", regex=True)74 cnt = ph[ph != ""].value_counts()75 live["phone_shared"] = ph.map(cnt).fillna(0).ge(3)76 live["origin_class"] = live.origin_class.fillna("E")77 live["certified"] = live.n_certifications.fillna(0) > 078 live["cert_adq"] = live.cert_labels.fillna("").str.contains("Aliments du Québec", regex=False)79 live["cert_lpq"] = live.cert_labels.fillna("").str.contains("Produits du Québec", regex=False)80 live["cert_bio"] = live.cert_labels.fillna("").str.contains("Biologique|CARTV", regex=True)81 live["store_kind"] = live.store_kind.fillna("").replace({"": "standard"})82 live["language"] = live.language.fillna("").replace({"": "unknown", "None": "unknown"})83 live["n_sources"] = live.discovery_sources_list.map(lambda s: len(json.loads(s)))84 live["first_source"] = live.discovery_sources_list.map(lambda s: (json.loads(s) or ["none"])[0])8586 a = li[li.f_analysis_assort]87 agg = a.groupby("store_id").agg(n_listings=("uid", "size"), n_categories=("category", "nunique"),88 share_autre=("category", lambda s: (s == "autre").mean()),89 median_price=("price", "median"), share_available=("available", lambda s: (s == 1).mean()),90 share_on_sale=("on_sale", "mean"), first_listing_utc=("first_seen_utc", "min"),91 last_listing_utc=("first_seen_utc", "max"))92 live = live.merge(agg, left_on="id", right_index=True, how="left")93 live["n_listings"] = live.n_listings.fillna(0).astype(int)94 top = a.groupby(["store_id", "category"]).size().reset_index(name="n").sort_values(["store_id", "n"], ascending=[True, False]).drop_duplicates("store_id")95 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"])9697 keep = ["id", "name", "platform", "city", "region_db", "region_pip", "region_final", "region_source", "region_conflict", "has_geo", "lat", "lng",98 "postal_prefix", "phone_shared", "origin_class", "origin_confidence", "origin_evidence", "certified", "cert_adq", "cert_lpq", "cert_bio",99 "cert_labels", "store_kind", "language", "n_sources", "first_source", "reg_status", "product_count", "n_listings", "n_categories",100 "share_autre", "median_price", "share_available", "share_on_sale", "main_category", "first_listing_utc", "last_listing_utc", "last_sync_utc", "fail_streak"]101 out = live[keep].copy()102 out.to_parquet(P["interim"] / "stores_master.parquet", index=False)103 register_output(P["interim"] / "stores_master.parquet", "clean.stores")104 summ = {"live_stores": len(out), "regions_layer": layer_path, "region_source": out.region_source.value_counts().to_dict(),105 "region_conflicts": int(out.region_conflict.sum()), "unknown_region_listings": int(out.loc[out.region_final == "Inconnue", "n_listings"].sum()),106 "phone_shared_stores": int(out.phone_shared.sum()), "certified_stores": int(out.certified.sum()),107 "region_final_counts": out.region_final.value_counts().to_dict(),108 "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)}109 write_json(summ, ROOT / "experiments" / "store_master.json", "clean.stores")110 ledger("clean.stores", **summ)111 log.info(json.dumps(summ, ensure_ascii=False))112113114if __name__ == "__main__":115 main()116