#!/usr/bin/env python3
"""Build Network Audit HTML from Google Sheet + DM scan CSV."""
import csv
import re
from datetime import datetime
from pathlib import Path
SHEET_CSV = Path("/tmp/keith_network.csv")
DM_CSV = Path(__file__).parent / "follow_up_FULL.csv"
OUTPUT_HTML = Path(__file__).parent / "network_audit.html"
# ── Parse Google Sheet ────────────────────────────────────────────────────────
sheet_rows = []
with open(SHEET_CSV, encoding="utf-8") as f:
reader = csv.reader(f)
for row in reader:
if len(row) < 2:
continue
name = row[0].strip()
handle = row[1].strip() if len(row) > 1 else ""
reached_out = (row[2].strip().upper() == "TRUE") if len(row) > 2 else False
followed_up = (row[3].strip().upper() == "TRUE") if len(row) > 3 else False
note = row[4].strip() if len(row) > 4 else ""
if name.lower() in ("contact info", "marc.paffrath") or handle == "Contact Info":
continue
if not handle.startswith("@"):
continue
if not name:
name = handle # use handle as display name if name is empty
sheet_rows.append({
"name": name,
"handle": handle.lower(),
"reached_out": reached_out,
"followed_up": followed_up,
"note": note,
})
# ── Parse DM scan CSV ─────────────────────────────────────────────────────────
dm_by_handle = {}
with open(DM_CSV, encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
h = row["handle"].strip().lower()
dm_by_handle[h] = {
"follow_up_type": row["follow_up_type"],
"last_msg_date": row["last_msg_date"],
"last_sender": row["last_sender"],
"thread_id": row["thread_id"],
"keith_last_msg": row["keith_last_msg"][:180] + "…" if len(row["keith_last_msg"]) > 180 else row["keith_last_msg"],
"other_last_msg": row["other_last_msg"][:180] + "…" if len(row["other_last_msg"]) > 180 else row["other_last_msg"],
"conversation_summary":row["conversation_summary"],
"suggested_reply": row.get("suggested_reply", ""),
"lead_score": row.get("lead_score", "0"),
"intent": row.get("intent", ""),
}
# ── Merge + sort ──────────────────────────────────────────────────────────────
def parse_note_date(note: str) -> datetime:
note = note.strip()
if not note:
return datetime.min
parts = re.split(r"[+,]", note)
last = parts[-1].strip()
for fmt in ("%d%b%y", "%d%b%Y", "%d %b %y", "%d %b %Y"):
try:
return datetime.strptime(last, fmt)
except Exception:
pass
return datetime.min
def parse_dm_date(d: str) -> datetime:
if not d:
return datetime.min
try:
return datetime.strptime(d, "%Y-%m-%d %H:%M")
except Exception:
return datetime.min
merged = []
for r in sheet_rows:
dm = dm_by_handle.get(r["handle"])
dm_date = parse_dm_date(dm["last_msg_date"]) if dm else datetime.min
note_date = parse_note_date(r["note"])
best_date = max(dm_date, note_date)
merged.append({**r, "dm": dm, "best_date": best_date, "dm_date": dm_date})
def lead_score(r):
dm = r["dm"]
return int(dm["lead_score"]) if dm and dm.get("lead_score") else 0
merged.sort(key=lambda x: x["best_date"], reverse=True)
# ── Status logic ──────────────────────────────────────────────────────────────
def status(row):
dm = row["dm"]
if dm:
if dm["follow_up_type"] == "NEEDS_REPLY":
return "needs_reply"
return "bump_up"
if row["followed_up"]:
return "done"
if row["reached_out"]:
return "reached_out"
return "not_contacted"
STATUS_LABEL = {
"needs_reply": "⚡ NEEDS REPLY",
"bump_up": "↑ BUMP UP",
"done": "✓ DONE",
"reached_out": "→ REACHED OUT",
"not_contacted": "· NOT YET",
}
STATUS_COLOR = {
"needs_reply": "#ff3b30",
"bump_up": "#ff9500",
"done": "#30d158",
"reached_out": "#0a84ff",
"not_contacted": "#636366",
}
def days_ago(dt: datetime) -> str:
if dt == datetime.min:
return "—"
delta = datetime.now() - dt
d = delta.days
if d == 0: return "today"
if d == 1: return "yesterday"
if d < 7: return f"{d}d ago"
if d < 30: return f"{d//7}w ago"
if d < 365: return f"{d//30}mo ago"
return f"{d//365}y ago"
def esc(s: str) -> str:
return s.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
# ── Build rows ────────────────────────────────────────────────────────────────
counts = {k: 0 for k in STATUS_LABEL}
rows_html = ""
for r in merged:
st = status(r)
color = STATUS_COLOR[st]
label = STATUS_LABEL[st]
dm = r["dm"]
counts[st] += 1
handle_raw = r["handle"].lstrip("@")
ig_url = f"https://www.instagram.com/{handle_raw}/"
dm_url = f"https://www.instagram.com/direct/t/{dm['thread_id']}" if dm else ig_url
date_display = days_ago(r["best_date"])
date_raw = (dm["last_msg_date"] if dm else r["note"]) or "—"
# last message preview
if dm:
if dm["follow_up_type"] == "NEEDS_REPLY":
preview_label = "they said:"
preview_text = esc(dm["other_last_msg"])
preview_cls = "preview-other"
else:
preview_label = "keith said:"
preview_text = esc(dm["keith_last_msg"])
preview_cls = "preview-keith"
else:
preview_label = ""
preview_text = ""
preview_cls = ""
# suggested reply
suggestion = ""
if dm and dm.get("suggested_reply") and not dm["suggested_reply"].startswith("["):
suggestion = esc(dm["suggested_reply"])
# intent + score
score = int(dm["lead_score"]) if dm and dm.get("lead_score") else 0
intent_raw = dm.get("intent", "") if dm else ""
intent_label = intent_raw.split("|")[0].strip() if "|" in intent_raw else intent_raw
intent_signal = intent_raw.split("|")[1].strip() if "|" in intent_raw else ""
INTENT_COLORS = {"HOT": "#ff3b30", "WARM": "#ff9500", "COLD": "#636366", "SUPPORT": "#0a84ff", "COLLAB": "#bf5af2", "SPAM": "#3a3a3c"}
intent_color = INTENT_COLORS.get(intent_label, "#555")
score_bar = f'
=70 else "#ff9500" if score>=40 else "#555"}">
{score}'
intent_badge = f'{intent_label}' if intent_label else ""
# flags
flag_reached = 'reached out' if r["reached_out"] else 'not reached'
flag_followed = 'followed up' if r["followed_up"] else ""
flag_note = f'{esc(r["note"])}' if r["note"] else ""
# ── first-name extraction + prefab templates ────────────────────────
_src_name = (r["name"] or r["handle"].lstrip("@") or "").strip()
_tokens = re.split(r"[\s,._\-]+", _src_name)
_first = ""
for _tok in _tokens:
_clean = re.sub(r"[^A-Za-z]", "", _tok)
if len(_clean) >= 2:
_first = _clean[0].upper() + _clean[1:].lower()
break
_greet = f"Hey {_first}" if _first else "Hey there"
t_bump = f"{_greet}, just bumping this up. I know how easy it is for messages to get buried. \U0001f9f6\U0001f64f\U0001fab7"
t_circle = f"{_greet}, just circling back, how are you doing? Did you have a chance to try out shaking yet? If you got any questions, feel free to message me anytime \U0001f9f6\U0001f64f\U0001fab7"
_ai_link = (f'{suggestion}'
f''
if suggestion else '')
_tpl_block = (f'
{date_display}{date_raw[:10] if date_raw != '—' else '—'}
{preview_html}
{suggestion_html}
{flag_reached}
{flag_followed}
{flag_note}
"""
# ── Summary bar ───────────────────────────────────────────────────────────────
summary_pills = "".join(
f''
for k in STATUS_LABEL if counts[k] > 0
)
summary_pills += ''
summary_pills += ''
# ── Full HTML ─────────────────────────────────────────────────────────────────
html = f"""
Keith — Network Audit {datetime.now().strftime('%d %b %Y')}
Keith — Network Audit {datetime.now().strftime('%d %b %Y')}
{summary_pills}
{len(merged)} shown
quick filters:
Status
Person
Last Contact ↓
Last Message + Suggested Reply
Flags
{rows_html}
"""
OUTPUT_HTML.write_text(html, encoding="utf-8")
print(f"written: {OUTPUT_HTML}")
# sync to iCloud
import shutil
icloud = Path.home() / "Library/Mobile Documents/com~apple~CloudDocs/keith_network_audit.html"
shutil.copy2(OUTPUT_HTML, icloud)
print(f"synced to iCloud: {icloud.name}")
print(f"total: {len(merged)} entries")
for k, v in counts.items():
if v:
print(f" {STATUS_LABEL[k]}: {v}")