`warmth.py`

127 lines
"""
Warmth scoring for RENTV leads — one definition of "warm", shared by the live
console (live_server.py) and the top-N export (export_top.py) so they never drift.

A lead's warmth blends the three signals that already live in the sheet:
  1. Centrality  — # Appearances (how many source tabs a contact shows up on).
  2. Intent      — Status (color): a flagged/left-message lead outranks a cold one.
  3. Reachability— do we actually have an email / phone / LinkedIn / enriched VP?
Plus a Strong-Prospects membership bonus (Steve's own hand-tagged short list).

The score is a plain int. Higher = work first. Section/header rows get a large
negative so they sink to the bottom under a descending sort.

Pure/stdlib, no deps — same input always yields the same score ($0, offline).
"""

WARMTH_HEADER = "🔥 Warmth"

# Status (color) -> intent weight. Keys match the sheet's inferred status labels.
_STATUS_WEIGHT = {
    "Flag / Priority": 40,
    "In Progress / Left Message": 32,
    "Prospect / Follow-up": 26,
    "Active / Contacted": 20,
    "Needs Email / Outreach Pending": 12,
    "On Hold / Inactive": -25,
    "Section / Category": -100,   # header/divider rows — sink them
}


def _cell(row, idx, name):
    i = idx.get(name)
    if i is None or i >= len(row):
        return ""
    v = row[i]
    return "" if v is None else str(v).strip()


def _int(s):
    try:
        return int(float(str(s).strip() or 0))
    except (ValueError, TypeError):
        return 0


def score(row, idx):
    """row = list of cell values; idx = {header_name: col_index}. Returns int warmth."""
    pts = 0

    # 1. Centrality — capped so a contact spammed across every tab can't dominate.
    appearances = _int(_cell(row, idx, "# Appearances"))
    pts += min(appearances, 6) * 8            # 0..48

    # 2. Intent
    status = _cell(row, idx, "Status (color)")
    pts += _STATUS_WEIGHT.get(status, 0)

    # 3. Strong-Prospects membership (Steve's hand-tagged list)
    src = _cell(row, idx, "Source Tabs").lower()
    if "strong prospects" in src:
        pts += 30

    # 4. Reachability — a warm lead we can't contact is worth less.
    email = _cell(row, idx, "Email") or _cell(row, idx, "Updated Email (found)") \
        or _cell(row, idx, "Likely Email (inferred)")
    if email:
        pts += 10
    phone = any(_cell(row, idx, p) for p in
                ("Phone 1", "Phone 2", "Phone 3", "Company Phone (found)", "Marketing/VP Phone"))
    if phone:
        pts += 6
    if _cell(row, idx, "Marketing/VP Name") or _cell(row, idx, "Marketing/VP Contact"):
        pts += 8
    if _cell(row, idx, "LinkedIn (Contact)") or _cell(row, idx, "LinkedIn (Company)"):
        pts += 3

    return pts


def reason(row, idx):
    """Short human 'why warm' string for the export — mirrors score()'s inputs."""
    bits = []
    app = _int(_cell(row, idx, "# Appearances"))
    if app:
        bits.append(f"{app} tab{'s' if app != 1 else ''}")
    status = _cell(row, idx, "Status (color)")
    if status and status != "Section / Category":
        bits.append(status)
    if "strong prospects" in _cell(row, idx, "Source Tabs").lower():
        bits.append("Strong Prospect")
    reach = []
    if _cell(row, idx, "Email") or _cell(row, idx, "Updated Email (found)") \
            or _cell(row, idx, "Likely Email (inferred)"):
        reach.append("email")
    if any(_cell(row, idx, p) for p in
           ("Phone 1", "Phone 2", "Phone 3", "Company Phone (found)", "Marketing/VP Phone")):
        reach.append("phone")
    if _cell(row, idx, "Marketing/VP Name"):
        reach.append("VP contact")
    if reach:
        bits.append("has " + "/".join(reach))
    return "; ".join(bits)


def can_score(headers):
    """Only augment tabs that actually carry the signals — others get no column."""
    hs = set(headers)
    return ("Status (color)" in hs) or ("# Appearances" in hs)


def augment(payload):
    """Given a tab/combined payload {headers, rows, ...}, append the Warmth column
    (in the RESPONSE only — never written back to the sheet). Returns the payload."""
    headers = payload.get("headers", [])
    if not can_score(headers):
        return payload
    idx = {h: i for i, h in enumerate(headers)}
    new_headers = list(headers) + [WARMTH_HEADER]
    new_rows = []
    for r in payload.get("rows", []):
        new_rows.append(list(r) + [score(r, idx)])
    payload = dict(payload)
    payload["headers"] = new_headers
    payload["rows"] = new_rows
    payload["warmth_col"] = len(new_headers) - 1   # so the UI can find it
    return payload