#!/usr/bin/env python3
"""Build Network Audit HTML from Google Sheet + DM scan CSV."""
import csv
import json
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"
ACTIVITY_LOG = Path(__file__).parent / "activity_log.json"
# load sent handles to embed in HTML (so GitHub Pages shows sent state)
_sent_handles = []
if ACTIVITY_LOG.exists():
try:
_log = json.loads(ACTIVITY_LOG.read_text())
if _log and isinstance(_log[0], dict):
_sent_handles = list({e["handle"] for e in _log if "handle" in e})
else:
_sent_handles = _log
except Exception:
pass
SENT_HANDLES_JS = json.dumps(_sent_handles)
# ── 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,
})
# ── Skip list ─────────────────────────────────────────────────────────────────
SKIP_FILE = Path(__file__).parent / "skip_list.txt"
skip_handles = set()
if SKIP_FILE.exists():
for line in SKIP_FILE.read_text().splitlines():
h = line.strip().lower()
if h:
skip_handles.add(h)
# ── 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()
if h in skip_handles:
continue
dm_by_handle[h] = {
"display_name": row.get("display_name", ""),
"focus_hint": row.get("focus_hint", ""),
"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
# ── Family / friends exclude list (not leads, won't appear in audit) ──────────
EXCLUDE_HANDLES = {
"@the.ecosystem.movement",
"@finnwheatley",
"@aliseabass",
"@aprilvaccaroart",
"@alexislumpkin",
"@nature_in_lex",
"@drpeterrohde",
"@allymacpole",
"@cristallelove",
}
# dedupe sheet_rows by handle: keep the entry with the longest non-handle-like name
_seen_handles = {}
for _r in sheet_rows:
_h = _r["handle"].lower().strip()
_name = (_r.get("name", "") or "").strip()
# score the name: real names (with space) > short names > handle-as-name
_score = (10 if " " in _name else (5 if _name and not _name.startswith("@") else 0)) + len(_name)
if _h not in _seen_handles or _score > _seen_handles[_h][1]:
_seen_handles[_h] = (_r, _score)
sheet_rows = [v[0] for v in _seen_handles.values()]
merged = []
for r in sheet_rows:
if r["handle"].lower().strip() in EXCLUDE_HANDLES:
continue
dm = dm_by_handle.get(r["handle"])
if not dm:
# no IG thread = nothing actionable; skip stub rows that only have a sheet checkbox
continue
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"
# ── User-set buckets (EXCLUDED / PARTNER) ────────────────────────────────────
import json as _json
_buckets_path = Path(__file__).parent / "buckets.json"
USER_BUCKETS = {}
if _buckets_path.exists():
try:
USER_BUCKETS = _json.loads(_buckets_path.read_text())
except Exception:
USER_BUCKETS = {}
# ── Auto-pause computation from activity_log ─────────────────────────────────
from collections import Counter
from datetime import timedelta as _timedelta
_log_path = Path(__file__).parent / "activity_log.json"
_bump_count = Counter()
_last_bump_date = {}
if _log_path.exists():
try:
_entries = _json.loads(_log_path.read_text())
# sort by timestamp ascending so we can dedupe consecutive ones
def _parse_ts(_e):
try: return datetime.strptime(_e["ts"], "%Y-%m-%d %H:%M")
except Exception: return datetime.min
_entries_sorted = sorted(_entries, key=_parse_ts)
_prev_bump = {} # handle -> last bump ts seen
for _e in _entries_sorted:
_h = _e.get("handle", "").strip().lower()
if not _h: continue
if _e.get("type") != "bump_up": continue
_ts = _parse_ts(_e)
if _ts == datetime.min: continue
# dedupe: skip entries within 5 min of the previous one for same handle
if _h in _prev_bump and (_ts - _prev_bump[_h]).total_seconds() < 300:
continue
_prev_bump[_h] = _ts
_bump_count[_h] += 1
if _h not in _last_bump_date or _ts > _last_bump_date[_h]:
_last_bump_date[_h] = _ts
except Exception:
pass
# ── Load engagers.json (new prospects from likes/comments on last 3 posts) ──
ENGAGERS_DATA = {"posts": [], "engagers": []}
_engagers_path = Path(__file__).parent / "engagers.json"
if _engagers_path.exists():
try:
ENGAGERS_DATA = _json.loads(_engagers_path.read_text())
except Exception:
pass
ENGAGERS_POSTS_BY_ID = {p["id"]: p for p in ENGAGERS_DATA.get("posts", [])}
# ── Load new_followers.json (welcome targets from overnight scrape) ──────────
NEW_FOLLOWERS_DATA = {"followers": []}
_nf_path = Path(__file__).parent / "new_followers.json"
if _nf_path.exists():
try:
NEW_FOLLOWERS_DATA = _json.loads(_nf_path.read_text())
except Exception:
pass
def _keith_message_dates(summary: str):
"""Set of dates (date objects) where keith sent a message in the convo summary."""
out = set()
if not summary: return out
for entry in summary.split(" | "):
m = re.match(r"\[([^\]]+)\]\s*([^:]+):", entry)
if not m: continue
sender = m.group(2).strip().lower()
if sender != "keith": continue
date_str = m.group(1).strip()
for fmt in ("%d.%m.%Y", "%Y-%m-%d %H:%M", "%d.%m.%y"):
try:
out.add(datetime.strptime(date_str, fmt).date())
break
except Exception:
pass
return out
def _latest_other_date(summary: str):
"""Parse conversation_summary, return most recent OTHER (non-keith) message date."""
if not summary: return None
latest = None
for entry in summary.split(" | "):
m = re.match(r"\[([^\]]+)\]\s*([^:]+):", entry)
if not m: continue
date_str = m.group(1).strip()
sender = m.group(2).strip().lower()
if sender == "keith": continue
for fmt in ("%d.%m.%Y", "%Y-%m-%d %H:%M", "%d.%m.%y"):
try:
ts = datetime.strptime(date_str, fmt)
if latest is None or ts > latest:
latest = ts
break
except Exception:
pass
return latest
PAUSE_DAYS = 180 # 6 months
def compute_bucket(handle: str, last_msg_date_str: str, last_sender: str, summary: str = "") -> str:
"""Determine bucket: LEAD (default), EXCLUDED, PARTNER, LONG_PAUSE, EXHAUSTED."""
h = handle.lower().strip()
user = USER_BUCKETS.get(h)
if user and user.get("bucket") in ("EXCLUDED", "PARTNER"):
return user["bucket"]
# validate bump count: only count activity_log entries that have a matching
# keith message on the same date in the conversation_summary
keith_dates = _keith_message_dates(summary)
raw_bumps = _bump_count.get(h, 0)
last_bump_raw = _last_bump_date.get(h)
if raw_bumps == 0 or not last_bump_raw or last_bump_raw.date() not in keith_dates:
# the most recent activity_log bump is phantom (test click, etc)
# we conservatively require the LAST bump to be confirmed before pausing
return "LEAD"
bumps = raw_bumps # could be more granular but keeping simple
if bumps < 2:
return "LEAD"
last_bump = last_bump_raw
# Did the prospect engage since last bump? Check ANY OTHER message in convo,
# not just last_sender (since keith may have replied AFTER the prospect).
latest_other = _latest_other_date(summary)
if latest_other and latest_other.date() >= last_bump.date():
return "LEAD"
if bumps >= 3:
return "EXHAUSTED"
if datetime.now() < last_bump + _timedelta(days=PAUSE_DAYS):
return "LONG_PAUSE"
return "LEAD" # pause expired, eligible again
def esc(s: str) -> str:
return s.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
def render_convo_tail(summary: str, n: int = 5) -> str:
"""Parse "[DATE] sender: text | ..." and render last n messages as HTML."""
if not summary:
return ""
parts = [p.strip() for p in summary.split(" | ") if p.strip()]
parts = parts[-n:]
out = []
for p in parts:
m = re.match(r"\[([^\]]*)\]\s*([^:]+):\s*(.*)", p, re.DOTALL)
if not m:
out.append(f'
{esc(p)}
')
continue
date, sender, text = m.group(1), m.group(2).strip(), m.group(3).strip()
cls = "convo-keith" if sender.lower() == "keith" else "convo-other"
sender_label = "keith" if sender.lower() == "keith" else esc(sender)
out.append(
f'
'
f'{esc(date)} · {sender_label}'
f'{esc(text)}'
f'
'
)
return "".join(out)
# ── Build rows ────────────────────────────────────────────────────────────────
counts = {k: 0 for k in STATUS_LABEL}
bucket_counts = {"LEAD_REPLY": 0, "LEAD_BUMP": 0, "PARTNER": 0, "LONG_PAUSE": 0, "EXCLUDED": 0, "EXHAUSTED": 0, "ENGAGERS": 0, "NEW_FOLLOWERS": 0}
rows_html = ""
for r in merged:
st = status(r)
color = STATUS_COLOR[st]
label = STATUS_LABEL[st]
dm = r["dm"]
_row_bucket = compute_bucket(r["handle"], dm.get("last_msg_date", "") if dm else "", dm.get("last_sender", "") if dm else "", dm.get("conversation_summary", "") if dm else "")
_next_bump_iso = ""
if _row_bucket == "LONG_PAUSE":
_lb = _last_bump_date.get(r["handle"].lower().strip())
if _lb:
_next_bump_iso = (_lb + _timedelta(days=PAUSE_DAYS)).strftime("%Y-%m-%d")
counts[st] += 1
if _row_bucket == "PARTNER":
bucket_counts["PARTNER"] += 1
elif _row_bucket == "LONG_PAUSE":
bucket_counts["LONG_PAUSE"] += 1
elif _row_bucket == "EXCLUDED":
bucket_counts["EXCLUDED"] += 1
elif _row_bucket == "EXHAUSTED":
bucket_counts["EXHAUSTED"] += 1
elif st == "needs_reply":
bucket_counts["LEAD_REPLY"] += 1
elif st == "bump_up":
bucket_counts["LEAD_BUMP"] += 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 ────────────────────────
_handle_clean = r["handle"].lstrip("@").lower()
_handle_alphas = re.sub(r"[^a-z]", "", _handle_clean)
# candidates in priority order: IG display_name first (has real names),
# then the sheet name. Reject any token that equals the handle.
_candidates = []
if dm and dm.get("display_name"):
_candidates.append(dm["display_name"])
if r.get("name"):
_candidates.append(r["name"])
_first = ""
for _src in _candidates:
for _tok in re.split(r"[\s,._\-]+", _src.strip()):
_clean = re.sub(r"[^A-Za-z]", "", _tok)
if len(_clean) >= 2 and _clean.lower() != _handle_alphas:
_first = _clean[0].upper() + _clean[1:].lower()
break
if _first:
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. \U0001fAF6\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 \U0001fAF6\U0001f64f\U0001fab7"
t_sm = f"{_greet}, just circling back, how are you doing? Curious what drew you to follow Shaking Medicine \U0001fAF6\U0001f64f\U0001fab7"
t_content = f"{_greet}, just circling back, how are you doing? Curious if anything from my content that I shared recently stood out to you \U0001fAF6\U0001f64f\U0001fab7"
_focus = dm.get("focus_hint", "") if dm else ""
_focus_block = (f'
👁 focus on: {esc(_focus)}
'
if _focus and dm and dm.get("follow_up_type") == "NEEDS_REPLY" else "")
_convo_block = (f'📜 show last 5 messages
{date_display}{date_raw[:10] if date_raw != '—' else '—'}
{preview_html}
{suggestion_html}
{flag_reached}
{flag_followed}
{flag_note}
"""
# ── Render engagers as additional rows ────────────────────────────────────────
# deconflict: if person also showed up as new-follower, only show them as follower
_NEW_FOLLOWER_HANDLES = {f.get("handle", "").lower().strip() for f in NEW_FOLLOWERS_DATA.get("followers", [])}
for _eng in ENGAGERS_DATA.get("engagers", []):
handle = _eng.get("handle", "")
name = _eng.get("name", handle.lstrip("@"))
if not handle: continue
h_lc = handle.lower().strip()
# deconflict: skip if also a new-follower (follower bucket wins)
if h_lc in _NEW_FOLLOWER_HANDLES: continue
# final safety: skip if in EXCLUDE_HANDLES or USER_BUCKETS EXCLUDED/PARTNER
if h_lc in EXCLUDE_HANDLES: continue
if USER_BUCKETS.get(h_lc, {}).get("bucket") in ("EXCLUDED", "PARTNER"):
continue
bucket_counts["ENGAGERS"] += 1
post = ENGAGERS_POSTS_BY_ID.get(_eng.get("post_id", ""), {})
descriptor = post.get("descriptor", "your recent post")
post_url = post.get("url", "")
action = _eng.get("action", "engaged")
when = _eng.get("ts", "")
handle_raw = handle.lstrip("@")
ig_url = f"https://www.instagram.com/{handle_raw}/"
dm_url = f"https://ig.me/m/{handle_raw}"
# extract first name for template (same logic as main loop)
_src_name = name.strip()
_tokens = re.split(r"[\s,._\-]+", _src_name)
_first_e = ""
for _tok in _tokens:
_clean = re.sub(r"[^A-Za-z]", "", _tok)
if len(_clean) >= 2 and _clean.lower() != handle_raw.lower():
_first_e = _clean[0].upper() + _clean[1:].lower()
break
_greet_e = f"Hey {_first_e}" if _first_e else "Hey there"
template = f"{_greet_e}! 🤗 Thanks so much for the love on our {descriptor}. 💙🙏 I'm wondering, what resonated for you the most?"
comment_blurb = ""
if action == "commented" and _eng.get("comment"):
comment_blurb = f' "{esc(_eng["comment"][:100])}"'
rows_html += f"""
"""
# ── Render new followers as additional rows ──────────────────────────────────
for _nf in NEW_FOLLOWERS_DATA.get("followers", []):
handle = _nf.get("handle", "")
name = _nf.get("name", handle.lstrip("@"))
if not handle: continue
h_lc = handle.lower().strip()
if h_lc in EXCLUDE_HANDLES: continue
if USER_BUCKETS.get(h_lc, {}).get("bucket") in ("EXCLUDED", "PARTNER"):
continue
bucket_counts["NEW_FOLLOWERS"] += 1
handle_raw = handle.lstrip("@")
ig_url = f"https://www.instagram.com/{handle_raw}/"
dm_url = f"https://ig.me/m/{handle_raw}"
when = _nf.get("followed_at", "")
# first name extraction
_src_name = name.strip()
_tokens = re.split(r"[\s,._\-]+", _src_name)
_first_n = ""
for _tok in _tokens:
_clean = re.sub(r"[^A-Za-z]", "", _tok)
if len(_clean) >= 2 and _clean.lower() != handle_raw.lower():
_first_n = _clean[0].upper() + _clean[1:].lower()
break
_greet_n = f"Hey {_first_n}" if _first_n else "Hey there"
welcome = f"{_greet_n}! 🤗 Thanks so much for following along. Curious, what brought you to Shaking Medicine?"
welcome_floki = (
f"{_greet_n}! 🤗 Thanks so much for following! I suppose you found me through that Shaking Medicine reel we made with Floki. "
"I" + chr(39) + "m curious, what resonated the most for you?"
)
welcome_casual = (
f"{_greet_n}! 🌿 So glad you" + chr(39) + "re here. "
"Curious what drew you to explore shaking medicine?"
)
rows_html += f"""