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.

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

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):

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.

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.")

      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.

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.

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))

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.

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 ''}")

  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.

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']})")

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.

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:

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.

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).