gigafibre-fsm/scripts/campaigns/contacts_from_legacy.py

#!/usr/bin/env python3
"""
contacts_from_legacy.py — Build a Giftbit-compatible contacts CSV from
the legacy "adresses postales" export (selectionAdressesMap*.csv).

The legacy export is pipe-delimited with a 1-line title preamble. It
contains messy fields (multi-email cells, mixed-case addresses, etc.).
This script extracts the columns Giftbit needs, normalizes them, and
writes a clean comma-separated CSV ready for either:
  - Upload to Giftbit's UI for English delivery, or
  - Pairing with this folder's `send_gift_campaign.js` for French delivery.

Multi-email handling (when a cell contains "a@x.com;b@y.com" — typically
a couple living at the same address):
  --multi=first  (default) — keep the first email, 1 gift per household
  --multi=split             — split into multiple rows, 1 gift per person
                              (note: both rows share the same name)
  --multi=skip              — drop those rows for manual review

Usage:
  python3 contacts_from_legacy.py \\
    --src ~/Downloads/selectionAdressesMap9-12.csv \\
    --out ~/Downloads/giftbit-contacts.csv \\
    --multi first
"""
import argparse
import csv
import re
import sys
from pathlib import Path

LOWER_WORDS = {"de", "du", "des", "la", "le", "les", "au", "aux", "à",
               "et", "sur", "en"}
EMAIL_SPLIT = re.compile(r"\s*[;,]\s*")


def split_name(full):
    """First whitespace-delimited token = firstname, rest = lastname."""
    full = (full or "").strip()
    if not full:
        return "", ""
    parts = full.split(None, 1)
    return (parts[0], parts[1].strip()) if len(parts) > 1 else (parts[0], "")


def title_address(addr):
    """Title-case an address with French article rules.

    '25 Rue Des Hirondelles ' -> '25 Rue des Hirondelles'
    'chemin de la 1Re-Concession' -> 'Chemin de la 1re-Concession'
    """
    if not addr:
        return ""
    out = []
    for i, word in enumerate(addr.split()):
        lw = word.lower()
        if i > 0 and lw in LOWER_WORDS:
            out.append(lw)
        elif "-" in word:
            chunks = []
            for c in word.split("-"):
                cl = c.lower()
                if cl in LOWER_WORDS:
                    chunks.append(cl)
                else:
                    chunks.append(c[:1].upper() + c[1:].lower())
            out.append("-".join(chunks))
        else:
            out.append(word[:1].upper() + word[1:].lower())
    return " ".join(out)


def is_email(s):
    return bool(s and "@" in s and "." in s.split("@")[-1])


def pick_emails(row):
    """Pull clean email list from the first non-empty source field."""
    for k in ("email au compte", "email à l'adresse"):
        raw = (row.get(k) or "").strip()
        if not raw:
            continue
        emails = [e.strip().lower() for e in EMAIL_SPLIT.split(raw) if is_email(e.strip())]
        if emails:
            return emails
    return []


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--src", required=True, help="Source legacy CSV (pipe-delimited).")
    p.add_argument("--out", required=True, help="Output Giftbit-compatible CSV.")
    p.add_argument("--multi", choices=("first", "split", "skip"),
                   default="first",
                   help="How to handle cells with multiple emails.")
    args = p.parse_args()

    src = Path(args.src).expanduser()
    out = Path(args.out).expanduser()

    rows = []
    with open(src, encoding="utf-8") as f:
        next(f)  # skip "adresses postales" title line
        reader = csv.DictReader(f, delimiter="|")
        reader.fieldnames = [fn.strip() for fn in reader.fieldnames]
        for r in reader:
            rows.append({(k.strip() if k else k): (v.strip() if v else v)
                         for k, v in r.items()})

    contacts, seen = [], set()
    skipped_no_email = skipped_no_name = skipped_multi = 0

    for r in rows:
        emails = pick_emails(r)
        if not emails:
            skipped_no_email += 1
            continue
        if len(emails) > 1:
            if args.multi == "skip":
                skipped_multi += 1
                continue
            if args.multi == "first":
                emails = emails[:1]
        full = (r.get("nom au compte") or r.get("nom à l'adresse") or "").strip()
        if not full:
            skipped_no_name += 1
            continue
        firstname, lastname = split_name(full)
        desc = title_address(r.get("adresse dans F") or "")
        for em in emails:
            if em in seen:
                continue
            seen.add(em)
            contacts.append({
                "firstname": firstname,
                "lastname": lastname,
                "email": em,
                "description": desc,
            })

    with open(out, "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["firstname", "lastname", "email", "description"],
                           quoting=csv.QUOTE_MINIMAL)
        w.writeheader()
        for c in contacts:
            w.writerow(c)

    print(f"  source rows:                    {len(rows)}")
    print(f"  contacts written:               {len(contacts)}")
    print(f"  skipped (no email):             {skipped_no_email}")
    print(f"  skipped (multi-email, --skip):  {skipped_multi}")
    print(f"  skipped (no name):              {skipped_no_name}")
    print(f"  → {out}")


if __name__ == "__main__":
    sys.exit(main() or 0)