← back to Dw Chat Analyzer
pull.py
185 lines
#!/usr/bin/env python3
"""
DW Chat Traffic Analyzer — pull.
Pulls all Zendesk Chat (Zopim) conversations for Designer Wallcoverings via the
list endpoint (GET /api/v2/chats, 40/page, paginated by next_url), extracts
traffic + lead metadata (NOT full transcripts — only a trimmed visitor-intent
snippet), and upserts into realestate.dw_chats keyed by chat id.
Token: ZENDESK_CHAT_ACCESS_TOKEN from the secrets .env. Cost: $0 (Zendesk API
is included in the plan). Idempotent (upsert on id) + resumable (cursor file).
Usage:
python3 pull.py # full pull (all ~38.9k)
python3 pull.py --max-pages 3 # bounded test
python3 pull.py --resume # continue from saved next_url cursor
"""
import os, re, sys, json, time, argparse, subprocess, urllib.request, urllib.error
TOKEN = None
BASE = "https://www.zopim.com/api/v2/chats"
PGDB = os.environ.get("PGDATABASE", "realestate")
PGHOST = os.environ.get("PGHOST", "/tmp")
HERE = os.path.dirname(os.path.abspath(__file__))
CURSOR = os.path.join(HERE, ".cursor")
TABLE = "dw_chats"
# Standalone Zopim Chat account (no Support subdomain) → link back to the Chat dashboard.
ZENDESK_LINK_BASE = os.environ.get("ZENDESK_LINK_BASE", "https://designerwallcoverings.zendesk.com/chat/agent#home")
LEAD_KW = re.compile(r"\b(quote|price|pricing|cost|buy|purchase|order|sample|swatch|"
r"yard|roll|availab|lead time|trade|designer|install|ship)\b", re.I)
def token():
global TOKEN
if TOKEN: return TOKEN
env = os.path.expanduser("~/Projects/secrets-manager/.env")
for line in open(env):
if line.startswith("ZENDESK_CHAT_ACCESS_TOKEN="):
TOKEN = line.split("=", 1)[1].strip().strip('"').strip("'"); break
if not TOKEN: sys.exit("no ZENDESK_CHAT_ACCESS_TOKEN in secrets .env")
return TOKEN
def get(url):
for attempt in range(6):
req = urllib.request.Request(url, headers={"Authorization": "Bearer " + token()})
try:
with urllib.request.urlopen(req, timeout=60) as r:
return json.load(r)
except urllib.error.HTTPError as e:
if e.code == 429: # rate limited — back off
wait = int(e.headers.get("Retry-After", 2 ** attempt))
time.sleep(min(wait, 30)); continue
raise
raise SystemExit("too many 429s")
def clean(v):
return re.sub(r"[\t\r\n]+", " ", "" if v is None else str(v)).strip()
def visitor_intent(history):
"""Concat visitor-authored messages only (skip agent), trimmed — for lead intent."""
if not isinstance(history, list): return ""
msgs = []
for h in history:
if not isinstance(h, dict): continue
if h.get("type") == "chat.msg" and (h.get("sender_type") == "visitor" or h.get("nick", "").startswith("visitor")):
m = h.get("msg")
if m: msgs.append(m)
return clean(" | ".join(msgs))[:500]
def normalize(c):
v = c.get("visitor") or {}
wp = c.get("webpath") or []
def page(i):
try: x = wp[i]; return (x.get("to") if isinstance(x, dict) else str(x)) or ""
except Exception: return ""
intent = visitor_intent(c.get("history"))
ref_terms = clean(c.get("referrer_search_terms"))
has_contact = bool(v.get("email") or v.get("phone"))
blob = intent + " " + ref_terms
is_lead = bool(has_contact or LEAD_KW.search(blob))
# Tier (DTD verdict C, 2026-08-11): HOT = contact left OR explicit purchase language
# (the call-first list); WARM = keyword-only intent (analysis bucket). Data note: WARM
# is near-empty because visitor msg/search-terms are sparse — the real signal is contact.
if has_contact or re.search(r"\b(buy|order|purchase|invoice|checkout|place an order|proceed|pay)\b", blob, re.I):
lead_tier = "HOT"
elif LEAD_KW.search(blob):
lead_tier = "WARM"
else:
lead_tier = ""
conv = c.get("conversions")
return {
"id": clean(c.get("id")),
"started_at": clean(c.get("timestamp"))[:19],
"ended_at": clean(c.get("end_timestamp"))[:19],
"duration": clean(c.get("duration")), "response_time": clean((c.get("response_time") or {}).get("avg") if isinstance(c.get("response_time"), dict) else c.get("response_time")),
"department": clean(c.get("department_name")),
"agents": clean(", ".join(c.get("agent_names") or [])),
"chat_type": clean(c.get("type")), "missed": "t" if c.get("missed") else "f",
"rating": clean(c.get("rating")), "comment": clean(c.get("comment"))[:300],
"tags": clean(", ".join(c.get("tags") or [])),
"conversions": str(len(conv) if isinstance(conv, list) else (conv or 0)),
"zendesk_ticket_id": clean(c.get("zendesk_ticket_id")),
"ref_engine": clean(c.get("referrer_search_engine")), "ref_terms": ref_terms,
"landing_page": clean(page(0))[:200], "exit_page": clean(page(-1))[:200],
"page_count": str(len(wp)),
"visitor_name": clean(v.get("name")), "visitor_email": clean(v.get("email")),
"visitor_phone": clean(v.get("phone")), "visitor_city": clean(v.get("city")),
"visitor_region": clean(v.get("region")), "visitor_country": clean(v.get("country")),
"visitor_intent": intent, "is_lead": "t" if is_lead else "f", "lead_tier": lead_tier,
"zendesk_link": ZENDESK_LINK_BASE,
}
COLS = ["id","started_at","ended_at","duration","response_time","department","agents","chat_type",
"missed","rating","comment","tags","conversions","zendesk_ticket_id","ref_engine","ref_terms",
"landing_page","exit_page","page_count","visitor_name","visitor_email","visitor_phone",
"visitor_city","visitor_region","visitor_country","visitor_intent","is_lead","lead_tier","zendesk_link"]
DDL = f"""
CREATE TABLE IF NOT EXISTS {TABLE} (
id text PRIMARY KEY, started_at timestamptz, ended_at timestamptz,
duration int, response_time int, department text, agents text, chat_type text,
missed bool, rating text, comment text, tags text, conversions int, zendesk_ticket_id text,
ref_engine text, ref_terms text, landing_page text, exit_page text, page_count int,
visitor_name text, visitor_email text, visitor_phone text, visitor_city text,
visitor_region text, visitor_country text, visitor_intent text, is_lead bool, lead_tier text,
zendesk_link text, pulled_at timestamptz DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_dwchats_started ON {TABLE}(started_at DESC);
CREATE INDEX IF NOT EXISTS idx_dwchats_lead ON {TABLE}(is_lead);
"""
def flush(records):
if not records: return
tsv = os.path.join(HERE, "_batch.tsv")
with open(tsv, "w") as f:
for r in records:
f.write("\t".join(clean(r.get(c, "")) for c in COLS) + "\n")
# empty numeric/ts/bool -> NULL via a staging text table, then cast on insert
stg_cols = ", ".join(f"{c} text" for c in COLS)
sql = f"""{DDL}
CREATE TEMP TABLE _s ({stg_cols});
\\copy _s ({','.join(COLS)}) FROM '{tsv}' WITH (FORMAT text, DELIMITER E'\\t');
INSERT INTO {TABLE} ({','.join(COLS)})
SELECT id, NULLIF(started_at,'')::timestamptz, NULLIF(ended_at,'')::timestamptz,
NULLIF(duration,'')::numeric::int, NULLIF(response_time,'')::numeric::int, department, agents, chat_type,
missed::bool, rating, comment, tags, NULLIF(conversions,'')::int, zendesk_ticket_id,
ref_engine, ref_terms, landing_page, exit_page, NULLIF(page_count,'')::int,
visitor_name, visitor_email, visitor_phone, visitor_city, visitor_region, visitor_country,
visitor_intent, is_lead::bool, lead_tier, zendesk_link
FROM _s
ON CONFLICT (id) DO UPDATE SET rating=EXCLUDED.rating, comment=EXCLUDED.comment,
is_lead=EXCLUDED.is_lead, lead_tier=EXCLUDED.lead_tier, tags=EXCLUDED.tags, pulled_at=now();
"""
p = subprocess.run(["psql","-h",PGHOST,"-d",PGDB,"-v","ON_ERROR_STOP=1"],
input=sql, text=True, capture_output=True)
if p.returncode != 0:
sys.stderr.write(p.stderr); sys.exit(1)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--max-pages", type=int, default=0) # 0 = all
ap.add_argument("--resume", action="store_true")
a = ap.parse_args()
url = BASE
if a.resume and os.path.exists(CURSOR):
url = open(CURSOR).read().strip() or BASE
total, page, buf = 0, 0, []
while url:
d = get(url); page += 1
chats = d.get("chats", [])
buf.extend(normalize(c) for c in chats)
total += len(chats)
if page % 25 == 0 or not d.get("next_url"):
flush(buf); buf = []
open(CURSOR, "w").write(d.get("next_url") or "")
print(f" page {page}: {total} chats loaded (of {d.get('count','?')})", flush=True)
url = d.get("next_url")
if a.max_pages and page >= a.max_pages: break
time.sleep(0.15) # gentle on rate limits
flush(buf)
print(f"DONE: {total} chats pulled across {page} pages -> {TABLE}")
if __name__ == "__main__":
main()