# How to Merge Skip Trace Results Back into Your Property List with SQLite and Python You pulled an absentee owner list or a tax-delinquent list and sent it out for skip tracing. The return file that comes back rarely lines up one-to-one with your list. One owner has four phones, another has none, and a few returned IDs aren't in your list at all. This guide shows how to merge skip trace results back into your property list so every phone lands on the right record. If you're the person on the team who gets the vendor file and has to hand acquisitions a dialer upload by Monday, this is the script you rerun every batch. A plain join and pivot can cause three problems. It can send your dialer a number that texted STOP last month. It can call the same owner twice about two properties. And it can let you pay for a batch that doesn't match your list. This guide treats the merge as an audit instead. The script stops whenever counts, flags or duplicate copies of a phone disagree, reports phones shared across owners so you can assign them before the upload, and never edits the raw file. ## How do you merge skip trace results back into your property list? SQLite ships with Python as the `sqlite3` module, so there's nothing to install. The seven steps: 1. Check that SQLite supports window functions 2. Load and normalize the source list and the return file 3. Reconcile your counts against the vendor's hit count 4. Check duplicate phones before ranking 5. Rank and dedupe phones per owner with `ROW_NUMBER()` 6. Suppress your own opt-outs 7. Export the audit, dialer and no-dialable files All data here is fictional. The names, addresses and 555 numbers are made up. To run the script, paste the Python blocks from Steps 1 through 7 into one `.py` file in order. It needs no input files and writes three CSVs to the current folder. The appendix loader is for real files and is not part of that sequence. ## Why does every record need a stable record_id? Before the list goes out, give every row a `record_id` that never changes and is never reused (like `R0001`), and store it with the record in your CRM. Spreadsheet row numbers don't work because they shift as soon as someone sorts or deletes a row. If the list combines several lead sources, such as a driving-for-dollars sheet, an absentee pull and a tax-delinquent list, dedupe owners before you assign IDs. Otherwise you pay to trace the same person twice. The loader stops on any repeated `record_id`. ## Step 1: Does your Python's SQLite support ROW_NUMBER()? Window functions like `ROW_NUMBER() OVER (...)` need SQLite 3.25 or newer, so the script checks the version first. ```python import csv import io import re import sqlite3 from collections import Counter from datetime import datetime SCRIPT_VERSION = "merge-v2" # bump whenever the logic changes, and log it print("Script version:", SCRIPT_VERSION) print("SQLite version:", sqlite3.sqlite_version) if sqlite3.sqlite_version_info < (3, 25, 0): raise SystemExit("This script needs SQLite 3.25 or newer for window functions.") con = sqlite3.connect(":memory:") con.row_factory = sqlite3.Row ``` For real files, swap `":memory:"` for `"merge.db"` so you can inspect the results afterward. The script rebuilds its tables on every run, so copy `merge.db` somewhere else if you want to keep a batch. ## Step 2: How do you load a skip trace return file into SQLite? The return file has one row per phone. Each row carries: - `vendor_rank` (lower is better) - `phone_type` - `dnc_flag`, the vendor's do-not-call indicator - `litigator_flag`, explained below - `contact_name`, the person the phone is listed under `litigator_flag = Y` marks numbers the vendor associates with people who have filed telephone consumer protection lawsuits. Coverage and definitions vary by vendor, so check their data dictionary, and don't treat an `N` as proof a number is safe. The script passes both flags through as data, not as legal determinations. The sample covers the cases that break simple merges: - `R0001` has four phones. The fourth is listed under a relative and written `937.555.0104` - `R0002` has the same landline twice, plus a mobile labeled `Wireless` - `R0003`'s mobile also comes back under `R0005` - `R0004` wasn't returned, `R0006` came back with no phone (a no-hit), and `R0009` isn't in your list - `R0005` has a phone with a blank rank, listed under `RUIZ, ELENA` ```python SOURCE_CSV = """record_id,owner_name,property_address,city,state,zip R0001,Dana Whitfield,412 Alder Ln,Springfield,OH,45501 R0002,Marcus Bell,88 Quarry Rd,Springfield,OH,45503 R0003,Priya Raman,1450 Cedar Ct,Fairborn,OH,45324 R0004,Tom Okafor,9 Lantern Way,Xenia,OH,45385 R0005,Elena Ruiz,301 Mill St,Fairborn,OH,45324 R0006,Jordan Pike,77 Birch Ave,Xenia,OH,45385 """ RETURN_CSV = """record_id,phone,phone_type,vendor_rank,dnc_flag,litigator_flag,contact_name R0001,+19375550101,mobile,1,N,N,Dana Whitfield R0001,+19375550102,landline,2,Y,N,Dana Whitfield R0001,+19375550103,mobile,3,N,N,Dana Whitfield R0001,937.555.0104,voip,4,N,N,Kim Whitfield R0002,+19375550121,landline,1,N,N,Marcus Bell R0002,+19375550122,Wireless,1,N,N,Marcus Bell R0002,+19375550121,landline,3,N,N,Marcus Bell R0003,+19375550131,mobile,2,Y,N,Priya Raman R0005,+19375550151,mobile,1,N,Y,Elena Ruiz R0005,+19375550131,mobile,2,Y,N,Priya Raman R0005,+19375550152,landline,,N,N,"RUIZ, ELENA" R0006,,,,,, R0009,+19375550191,mobile,1,N,N,Sam Doyle """ con.executescript(""" DROP VIEW IF EXISTS ranked_phones; DROP TABLE IF EXISTS vendor_return; DROP TABLE IF EXISTS source_list; CREATE TABLE source_list (record_id TEXT PRIMARY KEY, owner_name TEXT, property_address TEXT, city TEXT, state TEXT, zip TEXT); CREATE TABLE vendor_return (record_id TEXT, phone TEXT, phone_type TEXT, vendor_rank INTEGER, dnc_flag TEXT, litigator_flag TEXT, contact_name TEXT); CREATE INDEX idx_return_record ON vendor_return(record_id); """) # Translate the vendor's spellings to yours. Add whatever your vendor sends. VALUE_MAP = { "phone_type": {"Wireless": "mobile", "Cell": "mobile", "Landline": "landline"}, "dnc_flag": {"Yes": "Y", "No": "N"}, "litigator_flag": {"Yes": "Y", "No": "N"}, } ENTITY_WORDS = {"LLC", "INC", "TRUST", "CORP", "LP", "ESTATE"} BAD_VALUES = [] def normalize_phone(raw): digits = re.sub(r"\D", "", raw) if len(digits) == 11 and digits.startswith("1"): digits = digits[1:] if len(digits) != 10: BAD_VALUES.append(("phone", raw)) return None return "+1" + digits def normalize_name(raw): # "WHITFIELD, DANA" -> "DANA WHITFIELD"; "Oak Holdings, LLC" stays as-is parts = [p.strip() for p in raw.split(",")] if len(parts) != 2 or not parts[1] or parts[1].upper().strip(".") in ENTITY_WORDS: return raw return f"{parts[1]} {parts[0]}" def normalize_date(raw): # CRM exports mix 9/14/2026 and 2026-09-14; store everything as ISO YYYY-MM-DD for fmt in ("%Y-%m-%d", "%m/%d/%Y"): try: return datetime.strptime(raw, fmt).date().isoformat() except ValueError: pass BAD_VALUES.append(("date", raw)) return None def clean(col, value): if value is None or value.strip() == "": return None value = VALUE_MAP.get(col, {}).get(value.strip(), value.strip()) if col == "phone": return normalize_phone(value) if col in ("owner_name", "contact_name"): return normalize_name(value) if col == "added_on": return normalize_date(value) return value def load_csv(table, text): reader = csv.DictReader(io.StringIO(text)) cols = reader.fieldnames if not cols: raise SystemExit(f"{table}: the file is empty (no header row).") rows = list(reader) if table == "source_list": dupes = sorted(i for i, n in Counter(r["record_id"] for r in rows).items() if n > 1) if dupes: raise SystemExit(f"source_list: duplicate record_id values: {dupes[:20]}") sql = f"INSERT INTO {table} ({','.join(cols)}) VALUES ({','.join('?' * len(cols))})" con.executemany(sql, [[clean(c, r[c]) for c in cols] for r in rows]) if BAD_VALUES: raise SystemExit(f"{table}: {len(BAD_VALUES)} values failed to normalize: {BAD_VALUES[:20]}") return len(rows) print("source rows loaded:", load_csv("source_list", SOURCE_CSV)) print("return rows loaded:", load_csv("vendor_return", RETURN_CSV)) bad = con.execute(""" SELECT 'phone_type' AS col, phone_type AS val FROM vendor_return WHERE phone_type IS NOT NULL AND phone_type NOT IN ('mobile','landline','voip') UNION SELECT 'dnc_flag', dnc_flag FROM vendor_return WHERE dnc_flag IS NOT NULL AND dnc_flag NOT IN ('Y','N') UNION SELECT 'litigator_flag', litigator_flag FROM vendor_return WHERE litigator_flag IS NOT NULL AND litigator_flag NOT IN ('Y','N') UNION SELECT 'vendor_rank', vendor_rank FROM vendor_return WHERE typeof(vendor_rank) NOT IN ('integer','null')""").fetchall() if bad: raise SystemExit("Unexpected vendor values (add them to VALUE_MAP): " + ", ".join(f"{r['col']}={r['val']!r}" for r in bad)) print("vendor values check: passed") ``` Output of Steps 1 and 2 (your SQLite version may differ): ```text Script version: merge-v2 SQLite version: 3.45.1 source rows loaded: 6 return rows loaded: 13 vendor values check: passed ``` Two checks stop the run on purpose: - **Phones and dates that won't normalize.** An extension, an international number or a date like `Sept 14` goes on `BAD_VALUES`. The run then stops and prints the count and the first 20 bad values, so you can fix a whole batch in one or two runs instead of one per row. Without the stop, a bad phone would load as blank and look like a no-hit. - **Unmapped vendor values.** A `Cell` or `Yes` missing from `VALUE_MAP`, or a rank like `1st`, would land in the wrong ranking bucket or slip past a DNC filter. Add each value the check lists to `VALUE_MAP` and rerun. `normalize_name` matters more than it looks. If your vendor writes names as `WHITFIELD, DANA` and you don't flip them, Step 5 ranks every one of the owner's own phones as a relative's. ## Step 3: Does the skip trace return file match your list? Count before you merge. The counts tell you whether the file belongs to this list, and they give you something concrete to raise with the vendor before you pay. ```python counts = con.execute(""" SELECT (SELECT COUNT(*) FROM source_list) AS source_records, (SELECT COUNT(DISTINCT s.record_id) FROM source_list s JOIN vendor_return v ON v.record_id = s.record_id) AS matched_records, (SELECT COUNT(DISTINCT s.record_id) FROM source_list s JOIN vendor_return v ON v.record_id = s.record_id WHERE v.phone IS NOT NULL) AS matched_with_phone, (SELECT COUNT(*) FROM source_list s WHERE NOT EXISTS (SELECT 1 FROM vendor_return v WHERE v.record_id = s.record_id)) AS unmatched_records, (SELECT COUNT(*) FROM vendor_return v WHERE NOT EXISTS (SELECT 1 FROM source_list s WHERE s.record_id = v.record_id)) AS orphan_return_rows """).fetchone() for key in counts.keys(): print(f"{key:>20}: {counts[key]}") REPORTS = { "Unmatched source records": """SELECT s.record_id, s.owner_name FROM source_list s WHERE NOT EXISTS (SELECT 1 FROM vendor_return v WHERE v.record_id = s.record_id)""", "Returned with no phone (no-hit)": """SELECT s.record_id, s.owner_name FROM source_list s JOIN vendor_return v ON v.record_id = s.record_id GROUP BY s.record_id HAVING COUNT(v.phone) = 0""", "Orphan return rows (ID not in source)": """SELECT v.record_id, v.phone FROM vendor_return v WHERE NOT EXISTS (SELECT 1 FROM source_list s WHERE s.record_id = v.record_id)""", } for title, sql in REPORTS.items(): print(f"\n{title}:") for r in con.execute(sql + " ORDER BY 1, 2"): print(" ", r[0], r[1]) VENDOR_REPORTED_HITS = 4 # fictional; copy the real figure from the vendor's batch summary HIT_COLUMN = "matched_with_phone" # or "matched_records", per the vendor's definition ACCEPT_HIT_MISMATCH = False # set True only after settling the gap, and log why if counts[HIT_COLUMN] == VENDOR_REPORTED_HITS: print(f"\nVendor-reported hits ({VENDOR_REPORTED_HITS}) match {HIT_COLUMN}.") elif ACCEPT_HIT_MISMATCH: print(f"\nACCEPTED MISMATCH: vendor {VENDOR_REPORTED_HITS}, {HIT_COLUMN} {counts[HIT_COLUMN]}.") else: raise SystemExit(f"Stopping: the vendor reports {VENDOR_REPORTED_HITS} hits, but " f"{HIT_COLUMN} is {counts[HIT_COLUMN]}. Settle it with the vendor.") ``` ```text source_records: 6 matched_records: 5 matched_with_phone: 4 unmatched_records: 1 orphan_return_rows: 1 Unmatched source records: R0004 Tom Okafor Returned with no phone (no-hit): R0006 Jordan Pike Orphan return rows (ID not in source): R0009 +19375550191 Vendor-reported hits (4) match matched_with_phone. ``` | Count | What it means | What to do when it's off | |---|---|---| | `matched_records` | Source IDs that came back, with or without a phone | Send the unmatched IDs (R0004) to the vendor and ask whether they were traced | | `matched_with_phone` | Matched records with at least one phone. The gap from `matched_records` is the no-hits (R0006) | If your no-hit count is far above the vendor's, check the phone column mapping | | `orphan_return_rows` | Return rows whose ID isn't in your list. They never reach an export | A large count usually means the wrong file or a mangled ID column | Set `HIT_COLUMN` to your vendor's definition of a hit. Here the vendor's 4 matches `matched_with_phone`. ## Step 4: How do you find duplicate and conflicting phones in a skip trace file? Step 5 keeps only one copy of each repeated phone. Before that happens, confirm that the copies agree on type, both flags and name. If they don't, the dedupe would pick a `dnc_flag` by sort order alone. `COUNT(DISTINCT ...)` ignores `NULL`, so the query wraps each column in `COALESCE(..., '')` to count blanks as their own value. Settle conflicts in `RESOLUTIONS`, never in the vendor's file. The sample has no conflicts, so the dictionary is empty. The commented entry shows the format. ```python RESOLUTIONS = { # ("R0002", "+19375550121"): {"dnc_flag": "Y"}, # vendor confirmed by email } for (record_id, phone), fix in RESOLUTIONS.items(): if not fix or not set(fix) <= {"phone_type", "dnc_flag", "litigator_flag", "contact_name"}: raise SystemExit(f"Resolution for {record_id} {phone}: unsupported columns {sorted(fix)}") cur = con.execute( f"UPDATE vendor_return SET {', '.join(f'{k} = ?' for k in fix)} " "WHERE record_id = ? AND phone = ?", [*fix.values(), record_id, phone]) if cur.rowcount == 0: raise SystemExit(f"Resolution for {record_id} {phone} matched no rows.") print("resolutions applied:", len(RESOLUTIONS)) print("\nDuplicate phones within a record:") conflicts = [] for r in con.execute(""" SELECT record_id, phone, COUNT(*) AS copies, MAX(COUNT(DISTINCT COALESCE(phone_type, '')), COUNT(DISTINCT COALESCE(dnc_flag, '')), COUNT(DISTINCT COALESCE(litigator_flag, '')), COUNT(DISTINCT COALESCE(contact_name, ''))) AS max_values FROM vendor_return WHERE phone IS NOT NULL AND record_id IN (SELECT record_id FROM source_list) GROUP BY record_id, phone HAVING COUNT(*) > 1 ORDER BY record_id, phone"""): status = "CONFLICT" if r["max_values"] > 1 else "type, flags, and name agree" if r["max_values"] > 1: conflicts.append(f"{r['record_id']} {r['phone']}") print(f" {r['record_id']} {r['phone']} copies={r['copies']} ({status})") if conflicts: raise SystemExit("Stopping: add a RESOLUTIONS entry for " + ", ".join(conflicts)) print("\nSame phone returned for more than one owner record:") for r in con.execute(""" SELECT phone, GROUP_CONCAT(record_id || ' ' || owner_name || ', listed under ' || COALESCE(contact_name, 'no name'), '|') AS details FROM (SELECT DISTINCT v.phone, v.record_id, s.owner_name, v.contact_name FROM vendor_return v JOIN source_list s ON s.record_id = v.record_id WHERE v.phone IS NOT NULL) GROUP BY phone HAVING COUNT(*) > 1 ORDER BY phone"""): print(" ", r["phone"]) for detail in sorted(r["details"].split("|")): # sorted so reruns print identically print(" ", detail) SHARED_REMOVALS = { # ("R0005", "+19375550131"): "listed under Priya Raman; assigned to R0003", } for (record_id, phone), why in SHARED_REMOVALS.items(): cur = con.execute("DELETE FROM vendor_return WHERE record_id = ? AND phone = ?", [record_id, phone]) if cur.rowcount == 0: raise SystemExit(f"Shared removal for {record_id} {phone} matched no rows.") print("shared removals applied:", len(SHARED_REMOVALS)) ``` ```text resolutions applied: 0 Duplicate phones within a record: R0002 +19375550121 copies=2 (type, flags, and name agree) Same phone returned for more than one owner record: +19375550131 R0003 Priya Raman, listed under Priya Raman R0005 Elena Ruiz, listed under Priya Raman shared removals applied: 0 ``` If you can't confirm the right value for a conflict, resolve toward the restrictive one (`dnc_flag = 'Y'`) and log why. A resolution whose key matches no row also stops the run, so an outdated fix can't quietly do nothing. A phone returned under two different records means one of two things: - **Same person.** One owner holds several properties under name variants your list dedupe missed, and your dialer would call them twice in one campaign. - **Different person.** It's a relative or a shared line, which makes it a weaker match for one of the records. Here the vendor lists the number under Priya Raman both times. It's most likely Priya's phone, showing up as an associated contact on Elena's record. The script reports the shared phone but doesn't change anything. Decide which record it belongs to before the upload, and log your decision. ## Step 5: How do you rank and dedupe phone numbers per owner with ROW_NUMBER()? One view uses two window functions. The first numbers the copies of each phone within a record, and `copy_num = 1` keeps the best-ranked copy. The second ranks the remaining phones for each owner, sorting by: 1. **Name match** (`name_order`). Phones listed under the owner come before relatives' phones. If you trust the vendor's rank more than its name data, remove this term. 2. **Vendor rank.** SQLite sorts `NULL` before numbers, so without the `vendor_rank IS NULL` term, blank-rank phones would jump to the top. That term pushes them to the end. 3. **Type.** Mobile first, then landline, then everything else. 4. **The phone itself**, as a tiebreaker so the order is the same on every run. ```python con.executescript(""" CREATE VIEW ranked_phones AS WITH matched AS ( SELECT v.*, CASE WHEN UPPER(v.contact_name) = UPPER(s.owner_name) THEN 0 ELSE 1 END AS name_order, CASE v.phone_type WHEN 'mobile' THEN 1 WHEN 'landline' THEN 2 ELSE 3 END AS type_order FROM vendor_return v JOIN source_list s ON s.record_id = v.record_id WHERE v.phone IS NOT NULL ), deduped AS ( SELECT *, ROW_NUMBER() OVER (PARTITION BY record_id, phone ORDER BY vendor_rank IS NULL, vendor_rank) AS copy_num FROM matched ) SELECT record_id, phone, phone_type, vendor_rank, dnc_flag, litigator_flag, contact_name, ROW_NUMBER() OVER (PARTITION BY record_id ORDER BY name_order, vendor_rank IS NULL, vendor_rank, type_order, phone) AS phone_rank FROM deduped WHERE copy_num = 1; """) print(f" {'record':<6} {'rk':>2} {'phone':<13} {'type':<8} {'v_rank':>6} dnc lit listed under") for r in con.execute("SELECT * FROM ranked_phones ORDER BY record_id, phone_rank"): vr = "" if r["vendor_rank"] is None else r["vendor_rank"] # blank cells come back as None; "or ''" keeps the width formatting from failing print(f" {r['record_id']:<6} {r['phone_rank']:>2} {r['phone']:<13} " f"{r['phone_type'] or '':<8} {vr!s:>6} {r['dnc_flag'] or '':<3} " f"{r['litigator_flag'] or '':<3} {r['contact_name'] or ''}") ``` ```text record rk phone type v_rank dnc lit listed under R0001 1 +19375550101 mobile 1 N N Dana Whitfield R0001 2 +19375550102 landline 2 Y N Dana Whitfield R0001 3 +19375550103 mobile 3 N N Dana Whitfield R0001 4 +19375550104 voip 4 N N Kim Whitfield R0002 1 +19375550122 mobile 1 N N Marcus Bell R0002 2 +19375550121 landline 1 N N Marcus Bell R0003 1 +19375550131 mobile 2 Y N Priya Raman R0005 1 +19375550151 mobile 1 N Y Elena Ruiz R0005 2 +19375550152 landline N N ELENA RUIZ R0005 3 +19375550131 mobile 2 Y N Priya Raman ``` For R0002, the duplicate landline collapsed to one row. The mobile and the landline tied at vendor rank 1, and the mobile won on type. For R0005, `normalize_name` turned `RUIZ, ELENA` into a match for the owner. That landline therefore ranks above the shared number listed under Priya, even though its vendor rank is blank. The SQL never filters out or rewrites `dnc_flag` or `litigator_flag`. Ranking decides the column order, not which numbers you're allowed to call. The vendor's flags reflect their check at trace time and don't replace your own scrub before each campaign. Numbers also get reassigned. Texting a mobile can be treated differently from calling a landline, which is why every export slot keeps `phone_type`. Calls and texts can fall under the TCPA, federal and state do-not-call rules, and state telemarketing laws. Check the rules that apply to you with your counsel. ## Step 6: How do you suppress your own opt-outs and STOP replies after a re-trace? Your team already keeps its own do-not-contact list: STOP replies, "don't call me" notes from acquisitions calls and numbers flagged in your CRM. A re-trace can bring those numbers back flagged `N`. A vendor's `N` never overrides your own opt-out. Export the list from your CRM fresh on every run. It goes through the same `clean` function, so `(937) 555-0103` matches `+19375550103` and `9/14/2026` becomes `2026-09-14`. ```python SUPPRESSION_CSV = """phone,reason,added_on (937) 555-0103,replied STOP,9/14/2026 937-555-0103,asked not to be called,2026-10-02 """ # fictional; in practice, export this from your CRM's do-not-contact field con.executescript(""" DROP TABLE IF EXISTS suppression_raw; DROP TABLE IF EXISTS internal_suppression; CREATE TABLE suppression_raw (phone TEXT, reason TEXT, added_on TEXT); """) print("suppression rows loaded:", load_csv("suppression_raw", SUPPRESSION_CSV)) con.executescript(""" CREATE TABLE internal_suppression AS SELECT phone, MIN(reason) AS reason, MAX(added_on) AS added_on FROM suppression_raw WHERE phone IS NOT NULL GROUP BY phone; CREATE UNIQUE INDEX idx_supp_phone ON internal_suppression(phone); """) print("Ranked phones on your internal suppression list:") for r in con.execute(""" SELECT p.record_id, p.phone, i.reason, i.added_on FROM ranked_phones p JOIN internal_suppression i ON i.phone = p.phone ORDER BY p.record_id, p.phone"""): print(f" {r['record_id']} {r['phone']} {r['reason']} (latest {r['added_on']})") ``` ```text suppression rows loaded: 2 Ranked phones on your internal suppression list: R0001 +19375550103 asked not to be called (latest 2026-10-02) ``` The two CRM rows collapse into one. The unique index guarantees one row per phone, so the Step 7 join can't repeat rows. `MAX(added_on)` compares text, so it only gives the right answer for `YYYY-MM-DD` dates. That's why `clean` converts the dates first. Unconverted, `9/14/2026` would sort after `2026-10-02` and be picked as the latest date. The collapsed row keeps one `reason` (the alphabetically first). See `suppression_raw` for all of them. The vendor returned R0001's third number with `dnc_flag = N`, but the owner opted out. Step 7 marks the number in the audit file and keeps it out of the dialer file. ## Step 7: How do you export phone_1, phone_2, phone_3 columns for your CRM and dialer? Most CRMs and dialers want one row per property, with `phone_1` through `phone_3` side by side. Conditional aggregation (`MAX(CASE WHEN phone_rank = n ...)`) builds each slot with its own type and flags, so a flag can't drift onto a neighboring number. The query starts from `source_list` with a `LEFT JOIN`. That keeps the unmatched and no-hit records and drops the orphans. The dialer file removes flagged numbers first and then renumbers the ones that are left. ```python TOP_N = 3 FLAGS = {"type": "p.phone_type", "dnc": "p.dnc_flag", "litigator": "p.litigator_flag", "internal_dnc": "CASE WHEN i.phone IS NULL THEN 'N' ELSE 'Y' END"} slot_cols = [] for n in range(1, TOP_N + 1): slot_cols.append(f"MAX(CASE WHEN p.phone_rank = {n} THEN p.phone END) AS phone_{n}") for name, expr in FLAGS.items(): slot_cols.append(f"MAX(CASE WHEN p.phone_rank = {n} THEN {expr} END) AS phone_{n}_{name}") export_sql = f""" SELECT s.*, COUNT(p.phone) AS distinct_phones, {', '.join(slot_cols)} FROM source_list s LEFT JOIN ranked_phones p ON p.record_id = s.record_id LEFT JOIN internal_suppression i ON i.phone = p.phone GROUP BY s.record_id ORDER BY s.record_id""" DIALER_POLICY = "exclude dnc_flag=Y, litigator_flag=Y, internal suppression; blank flags as N" DIAL_FILTER = """COALESCE(p.dnc_flag, 'N') <> 'Y' AND COALESCE(p.litigator_flag, 'N') <> 'Y' AND NOT EXISTS (SELECT 1 FROM internal_suppression i WHERE i.phone = p.phone)""" dial_cols = [] for n in range(1, TOP_N + 1): dial_cols.append(f"MAX(CASE WHEN d.dial_rank = {n} THEN d.phone END) AS phone_{n}") dial_cols.append(f"MAX(CASE WHEN d.dial_rank = {n} THEN d.phone_type END) AS phone_{n}_type") dialer_sql = f""" WITH dialable AS ( SELECT p.record_id, p.phone, p.phone_type, ROW_NUMBER() OVER (PARTITION BY p.record_id ORDER BY p.phone_rank) AS dial_rank FROM ranked_phones p WHERE {DIAL_FILTER}) SELECT s.*, {', '.join(dial_cols)} FROM source_list s JOIN dialable d ON d.record_id = s.record_id GROUP BY s.record_id ORDER BY s.record_id""" no_dial_sql = f""" SELECT s.* FROM source_list s WHERE s.record_id NOT IN (SELECT p.record_id FROM ranked_phones p WHERE {DIAL_FILTER}) ORDER BY s.record_id""" def write_csv(path, sql): cur = con.execute(sql) header = [d[0] for d in cur.description] rows = cur.fetchall() with open(path, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(header) writer.writerows([["" if v is None else v for v in row] for row in rows]) print(f"Wrote {len(rows)} rows ({len(header)} columns) to {path}") return rows for row in write_csv("merged_export.csv", export_sql): slots = [] for n in range(1, TOP_N + 1): cells = [row[f"phone_{n}{s}"] for s in ("", "_dnc", "_litigator", "_internal_dnc")] slots.append("/".join(c or "-" for c in cells)) print(f" {row['record_id']} n={row['distinct_phones']} " + " ".join(slots)) print("\nDialer policy:", DIALER_POLICY) for row in write_csv("dialer_upload.csv", dialer_sql): print(f" {row['record_id']} " + " ".join(row[f"phone_{n}"] or "-" for n in range(1, TOP_N + 1))) print() for row in write_csv("no_dialable_phone.csv", no_dial_sql): print(f" {row['record_id']} {row['owner_name']}") con.commit() con.close() ``` Each audit slot prints as `phone/dnc/litigator/internal_dnc`: ```text Wrote 6 rows (22 columns) to merged_export.csv R0001 n=4 +19375550101/N/N/N +19375550102/Y/N/N +19375550103/N/N/Y R0002 n=2 +19375550122/N/N/N +19375550121/N/N/N -/-/-/- R0003 n=1 +19375550131/Y/N/N -/-/-/- -/-/-/- R0004 n=0 -/-/-/- -/-/-/- -/-/-/- R0005 n=3 +19375550151/N/Y/N +19375550152/N/N/N +19375550131/Y/N/N R0006 n=0 -/-/-/- -/-/-/- -/-/-/- Dialer policy: exclude dnc_flag=Y, litigator_flag=Y, internal suppression; blank flags as N Wrote 3 rows (12 columns) to dialer_upload.csv R0001 +19375550101 +19375550104 - R0002 +19375550122 +19375550121 - R0005 +19375550152 - - Wrote 3 rows (6 columns) to no_dialable_phone.csv R0003 Priya Raman R0004 Tom Okafor R0006 Jordan Pike ``` **`merged_export.csv` is your audit record and CRM import.** It has 22 columns: the 6 source columns, `distinct_phones` and five columns per slot. - Use `record_id` as the external ID, so a later re-trace updates existing records instead of creating duplicates. - Map the three flag columns to your CRM's do-not-contact fields. - Don't upload this file to a dialer. Every number in it still carries its flags, so the dialer would call R0003's DNC number, R0001's opted-out number and R0005's litigator-flagged number. **`dialer_upload.csv` is the only file that goes to the dialer.** The filter runs before the slots are renumbered. As a result, R0001's fourth phone, which `TOP_N` cut from the audit export, becomes its `phone_2` here. To dial only numbers that appear in the audit export, add this line right after `DIAL_FILTER` is defined: ``` DIAL_FILTER += f"\n AND p.phone_rank <= {TOP_N}" ``` Both the dialer file and the no-dialable file pick it up. (It's not part of the sample run, so the outputs above don't include it.) Two more settings to check: - **Blank flags count as `N`.** That's a policy choice, not a fact about the number. To skip numbers the vendor didn't check, change `'N'` to `'Y'` in both `COALESCE` calls and update `DIALER_POLICY` to match. - **Shared numbers.** A clean shared number would land in both owners' rows. Add a `SHARED_REMOVALS` entry for the record it doesn't belong to. **`no_dialable_phone.csv` is still a lead list.** R0004 wasn't returned and R0006 was a no-hit, so send them to your mailer or a second-source trace. For R0003, a DNC flag on the phone doesn't bar mail. Still check whether the owner asked not to be contacted at all before you send anything. These filters are a starting point, not a legal determination. Scrub against current do-not-call lists before each campaign, and confirm your policy with counsel. ## What to log for every skip trace batch Save the vendor's return file and your suppression export exactly as they arrived, with the batch date in the file names, and never edit them. Weeks later someone may ask why a number was or wasn't dialed. You can then rerun the script on the untouched originals with the logged settings and get the same answer. **Per-batch checklist (log each item):** - `SCRIPT_VERSION`, and confirmation that the vendor returned `record_id` unchanged - Every `VALUE_MAP` entry (and `COLUMN_MAP` entry, if you use the appendix loader) you added for this vendor, and how you handled each `BAD_VALUES` item (in code, never in the raw file) - The Step 3 counts, the vendor's hit count, `HIT_COLUMN`, and the reason if you set `ACCEPT_HIT_MISMATCH` - Each `RESOLUTIONS` entry and where the value came from (restrictive when unconfirmed) - Each shared phone, the record you assigned it to, and its `SHARED_REMOVALS` entry - The suppression export date and how many ranked phones it caught - `TOP_N`, the ranking order and `DIALER_POLICY` - That only `dialer_upload.csv` went to the dialer, that `no_dialable_phone.csv` went to mail or a re-trace, and that the campaign was scrubbed first ## Appendix: How do you load real CSV files instead of inline text? This version replaces the `load_csv` function in Step 2 and its two calls. Don't run both loaders. If you do, `source_list` loads twice and the duplicate IDs break the `PRIMARY KEY`. ```python COLUMN_MAP = {"Phone Number": "phone", "Record ID": "record_id"} # examples def load_csv(table, path, encoding="utf-8-sig"): with open(path, newline="", encoding=encoding) as f: reader = csv.DictReader(f, restkey="_extra") if not reader.fieldnames: raise SystemExit(f"{table}: {path} is empty (no header row).") cols = [COLUMN_MAP.get(c, c) for c in reader.fieldnames] if len(set(cols)) != len(cols): raise SystemExit(f"{table}: two header columns map to the same name: {cols}") rows = [] for r in reader: if "_extra" in r or None in r.values(): raise SystemExit(f"{table}: {path} line {reader.line_num} doesn't match the header.") rows.append({COLUMN_MAP.get(k, k): v for k, v in r.items()}) known = {r[1] for r in con.execute(f"PRAGMA table_info({table})")} unknown = [c for c in cols if c not in known] if unknown: raise SystemExit(f"{table}: header columns not in the table: {unknown}") if table == "source_list": dupes = sorted(i for i, n in Counter(r["record_id"] for r in rows).items() if n > 1) if dupes: raise SystemExit(f"source_list: duplicate record_id values: {dupes[:20]}") sql = f"INSERT INTO {table} ({','.join(cols)}) VALUES ({','.join('?' * len(cols))})" con.executemany(sql, [[clean(c, r[c]) for c in cols] for r in rows]) if BAD_VALUES: raise SystemExit(f"{table}: {len(BAD_VALUES)} values failed to normalize: {BAD_VALUES[:20]}") return len(rows) # print("source rows loaded:", load_csv("source_list", "source_list.csv")) # print("return rows loaded:", load_csv("vendor_return", "vendor_return.csv")) # Step 6: load_csv("suppression_raw", "crm_do_not_contact.csv") ``` Each of these problems stops the run with a clear message: - **Header problems.** `PRAGMA table_info` catches vendor headers you haven't mapped yet. A `set` comparison catches two headers mapped to the same name. - **Rows that don't match the header.** A row with more or fewer fields than the header usually means an unquoted comma in an owner name or address. The message gives the line number. - **Repeated `record_id` values.** `Counter` counts them in one pass, so a 40,000-row absentee pull doesn't stall the way a nested `list.count` check would. `encoding="utf-8-sig"` strips the invisible byte order mark that Excel often adds to the start of a CSV. Without it, `record_id` would read as `\ufeffrecord_id`. If the load fails with `UnicodeDecodeError`, the file was most likely saved in a Windows encoding such as cp1252. Open it with `encoding="cp1252"`, or re-save it as CSV UTF-8 from Excel. Published by Locate Next (locatenext.com).