#!/usr/bin/env python3 """ new_customer_porter.py — Swing Shift new-customer auto-porting Reads the latest uSchedule CSV export, diffs against a known-emails ledger, and imports net-new customers into the MailerLite Welcome group. Usage: python3 13_Scripts/new_customer_porter.py # uses default input python3 13_Scripts/new_customer_porter.py --input 11_Input/foo.csv python3 13_Scripts/new_customer_porter.py --dry-run # no ML write Env: MAILERLITE_API_KEY (set in automation/n8n/.env.local) Output: 12_Output/porter_ledger.csv — cumulative seen-emails ledger 12_Output/porter_log.txt — append-only run log """ import csv import json import os import sys import argparse import urllib.request import urllib.error from datetime import datetime from pathlib import Path ROOT = Path(__file__).resolve().parent.parent DEFAULT_INPUT = ROOT / "11_Input" / "CustomerList.csv" LEDGER = ROOT / "12_Output" / "porter_ledger.csv" LOG = ROOT / "12_Output" / "porter_log.txt" # MailerLite group to add new customers to WELCOME_GROUP_ID = "182257597706929274" MAILERLITE_API = "https://connect.mailerlite.com/api" # ── Email column detection ───────────────────────────────────────────────────── EMAIL_COLS = [ "EmailAddress", "email", "Email", "email_address", "CustomerEmail", "customer_email", "EMAIL", "e_mail", ] PHONE_COLS = [ "CellPhone", "phone", "Phone", "cell_phone", "PhoneNumber", "phone_number", "mobile", "Mobile", ] NAME_COLS = [ "CustomerName", "full_name", "FullName", "name", "Name", "FirstName", "LastName", "customer_name", ] def find_col(headers: list, candidates: list): for c in candidates: if c in headers: return c return None # ── Ledger ───────────────────────────────────────────────────────────────────── def load_ledger() -> set: if not LEDGER.exists(): return set() with open(LEDGER) as f: return {row["email"].strip().lower() for row in csv.DictReader(f) if row.get("email")} def update_ledger(new_emails: list[dict]): """Append new rows to ledger.""" exists = LEDGER.exists() with open(LEDGER, "a", newline="") as f: writer = csv.DictWriter(f, fieldnames=["email", "name", "added_at"]) if not exists: writer.writeheader() for row in new_emails: writer.writerow({ "email": row["email"], "name": row.get("name", ""), "added_at": datetime.now().isoformat(), }) # ── MailerLite import ────────────────────────────────────────────────────────── def mailerlite_import(subscribers: list[dict]) -> dict: api_key = os.environ.get("MAILERLITE_API_KEY", "") if not api_key: # Try loading from .env.local env_file = ROOT / "automation" / "n8n" / ".env.local" if env_file.exists(): for line in env_file.read_text().splitlines(): if line.startswith("MAILERLITE_API_KEY="): api_key = line.split("=", 1)[1].strip() break if not api_key: raise EnvironmentError("MAILERLITE_API_KEY not set") url = f"{MAILERLITE_API}/groups/{WELCOME_GROUP_ID}/subscribers/import" payload = json.dumps({ "subscribers": subscribers, "resubscribe": False, "autoresponders": True, # trigger Welcome automation }).encode() req = urllib.request.Request( url, data=payload, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "application/json", }, method="POST" ) try: with urllib.request.urlopen(req) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: raise RuntimeError(f"MailerLite error {e.code}: {e.read().decode()}") from e # ── Log ──────────────────────────────────────────────────────────────────────── def log(msg: str): ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") line = f"[{ts}] {msg}" print(line) with open(LOG, "a") as f: f.write(line + "\n") # ── Main ─────────────────────────────────────────────────────────────────────── def run(input_path: Path, dry_run: bool = False) -> dict: log(f"Starting porter — input={input_path} dry_run={dry_run}") if not input_path.exists(): log(f"ERROR: input file not found: {input_path}") return {"error": "input not found", "path": str(input_path)} # Load seen emails seen = load_ledger() log(f"Ledger: {len(seen)} known emails") # Parse CSV new_customers = [] skipped = 0 invalid = 0 with open(input_path, newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) headers = reader.fieldnames or [] email_col = find_col(headers, EMAIL_COLS) name_col = find_col(headers, NAME_COLS) phone_col = find_col(headers, PHONE_COLS) if not email_col: log(f"ERROR: no email column found. Headers: {headers}") return {"error": "no email column", "headers": headers} log(f"Columns — email={email_col} name={name_col} phone={phone_col}") for row in reader: email = (row.get(email_col) or "").strip().lower() # Stricter email validation import re if not email or "@" not in email or not re.match( r'^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$', email ): invalid += 1 continue if email in seen: skipped += 1 continue name = (row.get(name_col) or "").strip() if name_col else "" phone = (row.get(phone_col) or "").strip() if phone_col else "" entry = {"email": email} if name: entry["name"] = name if phone: entry["fields"] = {"phone": phone} new_customers.append(entry) log(f"Found {len(new_customers)} net-new, {skipped} already known, {invalid} invalid") if not new_customers: log("Nothing to import — all known.") return {"new": 0, "skipped": skipped, "invalid": invalid} if dry_run: log(f"[DRY RUN] Would import {len(new_customers)} subscribers to Welcome group") for c in new_customers[:5]: log(f" {c['email']}") if len(new_customers) > 5: log(f" ... and {len(new_customers) - 5} more") return {"new": len(new_customers), "skipped": skipped, "invalid": invalid, "dry_run": True} # Import to MailerLite in batches of 200 batch_size = 200 total_imported = 0 for i in range(0, len(new_customers), batch_size): batch = new_customers[i:i + batch_size] try: result = mailerlite_import(batch) data = result.get("data", {}) imported = data.get("imported", 0) total_imported += imported log(f"Batch {i//batch_size + 1}: processed={data.get('processed',0)} " f"imported={imported} unchanged={data.get('unchanged_count',0)}") except RuntimeError as e: log(f"Batch {i//batch_size + 1} failed: {e}") # Retry one-by-one to find and skip bad entries for entry in batch: try: r = mailerlite_import([entry]) d = r.get("data", {}) total_imported += d.get("imported", 0) except RuntimeError: log(f" Skipping bad email: {entry.get('email')}") # Update ledger with newly added update_ledger(new_customers) log(f"Ledger updated — {len(new_customers)} emails added") return { "new": len(new_customers), "imported_to_mailerlite": total_imported, "skipped": skipped, "invalid": invalid, } def main(): parser = argparse.ArgumentParser(description="Port new uSchedule customers to MailerLite") parser.add_argument("--input", default=str(DEFAULT_INPUT), help="Path to uSchedule CSV export") parser.add_argument("--dry-run", action="store_true", help="Detect new customers but do not write to MailerLite") args = parser.parse_args() result = run(Path(args.input), dry_run=args.dry_run) print(json.dumps(result, indent=2)) if __name__ == "__main__": main()