"""Reads CSV exports and maps their headers to canonical fields, with presets for HubSpot, Salesforce and Attio. The only module that opens input files; it never writes. """ import csv import hashlib import io import os import sys from collections import defaultdict from datetime import date from hygiene_core import ISO_RE, InputError, PERSONS, SLASH_RE, normh # Canonical fields per object. Presets below are header guesses; the mapping report confirms them. COMMON = ["record_id", "created_at", "updated_at", "owner", "source", "created_by", "last_activity_at", "last_engagement_at"] CANON = { "contacts": COMMON + ["email", "additional_emails", "first_name", "last_name", "full_name", "phone", "mobile", "title", "company_name", "associated_company_ids", "associated_deal_ids", "lifecycle_stage", "lead_status", "marketing_status", "opted_out", "do_not_call", "hard_bounced", "bounced_at", "bounce_reason", "last_form_submission_at", "last_click_at", "send_count", "country", "state", "postal_code", "is_converted", "converted_contact_id"], "companies": COMMON + ["name", "domain", "additional_domains", "website", "phone", "country", "city", "state", "postal_code", "industry", "lifecycle_stage", "parent_id", "associated_contact_ids", "associated_deal_ids"], "deals": COMMON + ["name", "amount", "currency", "close_date", "stage", "pipeline", "is_closed", "is_won", "is_lost", "deal_type", "lost_reason", "associated_company_ids", "associated_contact_ids", "next_step"], "users": ["user_id", "name", "email", "active", "kind"], "contact_roles": ["opportunity_id", "contact_id"], "queues": ["queue_id", "name"], } CANON["leads"] = CANON["contacts"] HS_COMMON = {"record_id": ["Record ID"], "created_at": ["Create Date", "Created Date"], "updated_at": ["Last Modified Date"], "source": ["Record source", "Original Traffic Source", "Original Source"], # the last is an older label "last_activity_at": ["Last Activity Date"]} HS = { "contacts": {"email": ["Email"], "additional_emails": ["Additional email addresses"], "first_name": ["First Name"], "last_name": ["Last Name"], "phone": ["Phone Number"], "mobile": ["Mobile Phone Number"], "title": ["Job Title"], "company_name": ["Company Name"], "associated_company_ids": ["Associated Company IDs"], "associated_deal_ids": ["Associated Deal IDs"], "lifecycle_stage": ["Lifecycle Stage"], "lead_status": ["Lead Status"], "marketing_status": ["Marketing contact status"], "opted_out": ["Unsubscribed from all email"], "bounce_reason": ["Email hard bounce reason"], "last_engagement_at": ["Last Engagement Date"], "last_form_submission_at": ["Recent Conversion Date"], "last_click_at": ["Last marketing email click date"], "send_count": ["Marketing emails delivered"], "country": ["Country/Region"], "state": ["State/Region"], "postal_code": ["Postal Code"], "owner": ["Contact owner"]}, "companies": {"name": ["Company name"], "domain": ["Company Domain Name"], "additional_domains": ["Additional domains"], "website": ["Website URL"], "phone": ["Phone Number"], "country": ["Country/Region"], "city": ["City"], "state": ["State/Region"], "postal_code": ["Postal Code"], "industry": ["Industry"], "lifecycle_stage": ["Lifecycle Stage"], "parent_id": ["Parent Company ID"], "associated_contact_ids": ["Associated Contact IDs"], "associated_deal_ids": ["Associated Deal IDs"], "owner": ["Company owner"]}, "deals": {"name": ["Deal Name"], "amount": ["Amount"], "currency": ["Currency"], "close_date": ["Close Date"], "stage": ["Deal Stage"], "pipeline": ["Pipeline"], "is_closed": ["Is Deal Closed?", "Is Closed"], "is_won": ["Is Closed Won"], "is_lost": ["Is closed lost"], "deal_type": ["Deal Type"], "lost_reason": ["Closed Lost Reason"], "associated_company_ids": ["Associated Company IDs"], "associated_contact_ids": ["Associated Contact IDs"], "next_step": ["Next Step"], "owner": ["Deal owner"]}, } SF_COMMON = {"record_id": ["Id"], "created_at": ["CreatedDate"], "updated_at": ["LastModifiedDate"], "owner": ["OwnerId"], "created_by": ["CreatedById"], "last_activity_at": ["LastActivityDate"]} SF_PERSON = {"email": ["Email"], "first_name": ["FirstName"], "last_name": ["LastName"], "full_name": ["Name"], "phone": ["Phone"], "mobile": ["MobilePhone"], "title": ["Title"], "opted_out": ["HasOptedOutOfEmail"], "do_not_call": ["DoNotCall"], "hard_bounced": ["IsEmailBounced"], "bounced_at": ["EmailBouncedDate"], # IsEmailBounced is on Contact only "bounce_reason": ["EmailBouncedReason"], "source": ["LeadSource"]} SF = { "contacts": dict(SF_PERSON, associated_company_ids=["AccountId"], country=["MailingCountry"], state=["MailingState"], postal_code=["MailingPostalCode"]), "leads": dict(SF_PERSON, company_name=["Company"], lead_status=["Status"], is_converted=["IsConverted"], converted_contact_id=["ConvertedContactId"], country=["Country"], state=["State"], postal_code=["PostalCode"]), "companies": {"name": ["Name"], "website": ["Website"], "phone": ["Phone"], "country": ["BillingCountry"], "city": ["BillingCity"], "state": ["BillingState"], "postal_code": ["BillingPostalCode"], "industry": ["Industry"], "parent_id": ["ParentId"]}, "deals": {"name": ["Name"], "amount": ["Amount"], "currency": ["CurrencyIsoCode"], "close_date": ["CloseDate"], "stage": ["StageName"], "is_closed": ["IsClosed"], "is_won": ["IsWon"], "deal_type": ["Type"], "associated_company_ids": ["AccountId"], "next_step": ["NextStep"], "source": ["LeadSource"]}, } AT_COMMON = {"record_id": ["Record ID"], "created_at": ["Created at"], "created_by": ["Created by"]} AT = { "contacts": {"full_name": ["Name"], "first_name": ["First name"], "last_name": ["Last name"], "email": ["Email addresses"], "phone": ["Phone numbers"], "title": ["Job title"], "company_name": ["Company"], "owner": ["Owner"]}, "companies": {"name": ["Name"], "domain": ["Domains"], "owner": ["Owner"]}, "deals": {"name": ["Deal name", "Name"], "stage": ["Deal stage", "Stage"], "amount": ["Deal value", "Value"], "owner": ["Deal owner", "Owner"], "close_date": ["Close date"]}, } AUX = { # HubSpot owner fields store owner IDs, not user IDs, so an Owner ID column wins when a file holds both. "users": {"user_id": ["Owner ID", "User ID", "Id"], "name": ["Name", "Full name"], "email": ["Email"], "active": ["Active", "IsActive", "Status"], "kind": ["Kind", "Type"]}, "contact_roles": {"opportunity_id": ["OpportunityId", "Deal ID"], "contact_id": ["ContactId", "Contact ID"]}, "queues": {"queue_id": ["Id", "Queue ID"], "name": ["Name", "Queue name"]}, } PRESETS = {"hubspot": (HS_COMMON, HS), "salesforce": (SF_COMMON, SF), "attio": (AT_COMMON, AT)} class Table: def __init__(self, obj, path): self.obj, self.path, self.name = obj, path, os.path.basename(path) self.headers, self.rows, self.malformed = [], [], [] self.encoding = self.delimiter = self.sha256 = "" self.col, self.mapinfo, self.blank_cols = {}, {}, set() self.extra = {} # rule-referenced headers: label -> column index self._dates = {} def raw(self, row, field): i = self.col.get(field) return row[i] if i is not None and i < len(row) else "" def v(self, row, field): return self.raw(row, field).strip() def rid(self, row): return self.v(row, "record_id") def usable(self, field): return field in self.col and self.col[field] not in self.blank_cols def why(self, *fields): out = [] for f in fields: if f not in self.col: state = self.mapinfo.get(f, ("unmapped",))[0] out.append(f"{self.obj}.{f} {'is ambiguous' if state == 'ambiguous' else 'is not mapped'}") elif self.col[f] in self.blank_cols: out.append(f"{self.obj}.{f} was not retrieved (blank on every row)") return "; ".join(out) def read_table(obj, path): t = Table(obj, path) try: with open(path, "rb") as fh: raw = fh.read() except OSError as e: raise InputError(f"cannot read {t.name}: {e.strerror}") t.sha256 = hashlib.sha256(raw).hexdigest() for enc in ("utf-8-sig", "cp1252", "latin-1"): try: text = raw.decode(enc) t.encoding = "utf-8" if enc == "utf-8-sig" else enc break except UnicodeDecodeError: continue first = text.split("\n", 1)[0] t.delimiter = max([",", ";", "\t", "|"], key=first.count) if first else "," csv.field_size_limit(min(sys.maxsize, 2 ** 31 - 1)) try: rows = list(csv.reader(io.StringIO(text, newline=""), delimiter=t.delimiter)) except csv.Error as e: raise InputError(f"{t.name} is not parseable CSV: {e}") rows = [r for r in rows if any(c.strip() for c in r)] if not rows: raise InputError(f"{t.name} is empty") if len(rows) < 2: # a header with no records would pass every check on nothing raise InputError(f"{t.name} has a header row and no records; re-export it, or leave that option out") t.headers = [h.strip() for h in rows[0]] for n, r in enumerate(rows[1:], start=2): if len(r) != len(t.headers): t.malformed.append(n) r = (r + [""] * len(t.headers))[:len(t.headers)] t.rows.append(r) return t def aliases_for(crm, obj): out = defaultdict(list) sources = [] if obj in AUX: sources = [AUX[obj]] elif crm in PRESETS: common, per = PRESETS[crm] sources = [per.get(obj, {}) if obj != "leads" or crm == "salesforce" else per.get("contacts", {}), common] else: for c in ("hubspot", "salesforce", "attio"): common, per = PRESETS[c] sources += [per.get(obj if obj in per else "contacts", {}) if obj in PERSONS else per.get(obj, {}), common] for src in sources: for field, names in src.items(): out[field] += [normh(n) for n in names if normh(n) not in out[field]] for field in CANON.get(obj, []): if normh(field) not in out[field]: out[field].append(normh(field)) return out def map_columns(t, crm, overrides): norm = [normh(h) for h in t.headers] for field, names in aliases_for(crm, t.obj).items(): if field in overrides: continue hits = [] for a in names: idx = [i for i, h in enumerate(norm) if h == a] if idx: hits.append(idx) if not hits: t.mapinfo[field] = ("unmapped", "", "") elif len(hits[0]) > 1: t.mapinfo[field] = ("ambiguous", "", "headers " + ", ".join(t.headers[i] for i in hits[0]) + " look the same") else: others = [t.headers[i[0]] for i in hits[1:]] t.col[field] = hits[0][0] t.mapinfo[field] = ("mapped", t.headers[hits[0][0]], ("preferred over " + ", ".join(others)) if others else "") for field, header in overrides.items(): idx = [i for i, h in enumerate(t.headers) if h == header] or [i for i, h in enumerate(norm) if h == normh(header)] if len(idx) != 1: raise InputError(f"--map {t.obj}.{field}: header '{header}' {'not found' if not idx else 'is not unique'} in {t.name}") t.col[field] = idx[0] t.mapinfo[field] = ("mapped", t.headers[idx[0]], "set with --map") for i in range(len(t.headers)): if all(not (r[i] if i < len(r) else "").strip() for r in t.rows): t.blank_cols.add(i) def header_index(t, label): """A rule may name a canonical field or a raw header.""" if label in t.col: return t.col[label] idx = [i for i, h in enumerate(t.headers) if normh(h) == normh(label)] return idx[0] if len(idx) == 1 else None def date_parser(t, field, order_arg): """Returns (parse, problem). parse(value) gives a date or None.""" if field in t._dates: return t._dates[field] if not t.usable(field): t._dates[field] = (None, t.why(field)) return t._dates[field] vals = [t.v(r, field) for r in t.rows if t.v(r, field)] iso = [v for v in vals if ISO_RE.match(v)] slash = [SLASH_RE.match(v) for v in vals if SLASH_RE.match(v)] dmy = any(int(m.group(1)) > 12 for m in slash) mdy = any(int(m.group(2)) > 12 for m in slash) order, problem = None, "" if slash: if dmy and mdy: problem = f"{t.obj}.{field} mixes day-first and month-first dates" elif dmy or mdy: order = "dmy" if dmy else "mdy" elif order_arg: order = order_arg else: problem = f"{t.obj}.{field} has ambiguous slash dates; pass --date-order mdy or dmy" if not iso and not slash: problem = f"{t.obj}.{field} has no recognisable dates" def parse(v): m = ISO_RE.match(v or "") try: if m: return date(int(m.group(1)), int(m.group(2)), int(m.group(3))) m = SLASH_RE.match(v or "") if m and order: a, b, y = int(m.group(1)), int(m.group(2)), int(m.group(3)) return date(y, b, a) if order == "dmy" else date(y, a, b) except ValueError: return None return None if problem and not iso: t._dates[field] = (None, problem) else: t._dates[field] = (parse, problem) return t._dates[field]