mattrguerra/meridian-outreach-vaultmainPrivate on GitHub, published here in full

census.py

Language Python15.1 KB356 lines

07-Automation/partner_pipeline/census.py

"""The XML census: refresh-xml and reconcile-deals."""
from __future__ import annotations
import json
import re
import shutil
import sys
from pathlib import Path
from .config import NON_PARTNER, guard_or_exit
from .history import load_merge_registry
from .paths import CLIENTS_DIR, OUT_DIR, TODAY, XML_MAP
from .stages import FUNNEL_DISPLAY, FUNNEL_KEYS, stage_status
from .stubs import index_existing_stubs, mirror_stub_pipeline, safe_stub_name
def norm_key(s) -> str:
"""Lowercase letters and digits only: how names are compared across Plinth,
Bitrix and the census."""
return re.sub(r"[^a-z0-9]", "", str(s or "").lower())
def _map_chunk_id(r: dict) -> str | None:
"""Census key for a client with no Bitrix contact yet: the Plinth user_id, which is
stable and unique, namespaced as pl-<id> so it can never collide with a Bitrix
contact or tracker id (those key the pipeline-mapped chunks)."""
uid = r.get("user_id")
if uid in (None, ""):
return None
return f"pl-{uid}"
def _resolve_stub(r: dict, idx: dict) -> tuple[str, dict]:
"""Return (stub_file_name, stub_frontmatter) for a plan row. Prefers the real stub
on disk (write-stubs runs first), so identity mirrors the stub per the XML contract;
falls back to the deterministic safe-name if somehow absent."""
email = (r.get("email") or "").strip().lower()
name = f"{(r.get('first_name') or '')} {(r.get('last_name') or '')}".strip().lower()
# name first: for a shared company mailbox (two different people, one email) the
# name disambiguates to the right stub; email is the fallback for the common
# unique-email case (same result there).
rec = idx["by_name"].get(name) or idx["by_email"].get(email)
if rec:
return rec["path"].name, rec.get("fm", {})
return safe_stub_name(r.get("first_name"), r.get("last_name"), r.get("email")) + ".md", {}
def _gen_map_summary(full_name, user_group, company, industry, rel_path) -> str:
s = f"{full_name} is a {user_group} client" if user_group else f"{full_name} is a partner client"
if company:
s += f" at {company}"
if industry:
s += f" in the {industry} space"
s += (". Not currently mapped to any outreach pipeline "
"(logged for accounting; pipeline placement to be populated later). "
f"Full enrichment profile: {rel_path}.")
return s
def _new_client_chunk(ET, r: dict, idx: dict):
"""Build a self-contained <client> chunk with an empty pipeline section
(<pipelines count="0" />) for a partner client not yet in any Bitrix pipeline.
Element order matches the schema-3.0 pipeline-mapped chunks."""
cid = _map_chunk_id(r)
file_name, fm = _resolve_stub(r, idx)
rel_path = f"/03-Clients/client-stubs/{file_name}"
first = (fm.get("first_name") or r.get("first_name") or "")
last = (fm.get("last_name") or r.get("last_name") or "")
full = f"{first} {last}".strip() or (r.get("email") or "")
email = (fm.get("email") or r.get("email") or "")
company = (fm.get("company") or r.get("company") or "")
title = (fm.get("title") or "")
industry = (fm.get("industry") or "")
linkedin = (fm.get("linkedin") or "")
website = (fm.get("company_website") or "")
user_group = (r.get("assigned_tier") or "")
# schema 3.0 dual identifiers, mirroring the stub. bitrix_id is the real CRM contact id
# (empty for a Plinth-sourced partner client with no Bitrix contact yet); plinth_id
# is the Plinth backend user_id, which we always have.
bitrix_id = (fm.get("bitrix_id") or "")
plinth_id = (fm.get("plinth_id") or r.get("user_id") or "")
c = ET.Element("client", {"id": cid})
def sub(tag, text=""):
e = ET.SubElement(c, tag)
if text not in (None, ""):
e.text = str(text)
return e
sub("full_name", full)
sub("first_name", first)
sub("last_name", last)
sub("email", email)
sub("bitrix_id", bitrix_id) # schema 3.0
sub("plinth_id", plinth_id) # schema 3.0
sub("company", company)
sub("title", title)
sub("industry", industry)
sub("tier") # no legacy tracker tier for Plinth-sourced partner clients
sub("user_group", user_group)
sub("linkedin", linkedin)
sub("company_website", website)
sub("summary", _gen_map_summary(full, user_group, company, industry, rel_path))
stub = ET.SubElement(c, "stub")
ET.SubElement(stub, "file_name").text = file_name
ET.SubElement(stub, "relative_path").text = rel_path
ET.SubElement(c, "pipelines", {"count": "0"}) # empty on purpose; populate later
ET.SubElement(c, "last_modified").text = TODAY
return c, email.strip().lower(), cid
def cmd_refresh_xml(args):
guard_or_exit()
import xml.etree.ElementTree as ET
if not XML_MAP.exists():
sys.exit(f"XML map not found: {XML_MAP}")
backup = OUT_DIR / f"partner-client-master-list.backup-{TODAY}.xml"
OUT_DIR.mkdir(parents=True, exist_ok=True)
shutil.copy2(XML_MAP, backup)
tree = ET.parse(XML_MAP)
root = tree.getroot()
# merge registry: an absorbed account must not exist in the census. Drop any
# absorbed pl-<id> chunk (self-healing if one was re-added by hand or sync);
# Bitrix-keyed chunks are never touched by this.
aliases, _ = load_merge_registry()
removed_merged = []
if aliases:
for el in list(root.findall("client")):
cid = el.get("id") or ""
if cid.startswith("pl-") and cid[3:] in aliases:
root.remove(el)
removed_merged.append(cid)
def _name_of(el):
f = (el.findtext("first_name") or "").strip().lower()
l = (el.findtext("last_name") or "").strip().lower()
return f"{f} {l}".strip()
# Index the existing map. Distinguish the pipeline-mapped chunks (real Bitrix
# contact / tracker ids) from the map-only chunks we key by pl-<user_id>.
# by_id : every chunk by its id (idempotency across runs for pl- chunks)
# by_email_pl : pipeline chunk emails only (never duplicate a Bitrix identity)
# seen_person : (email, name) -> pl- chunk, so two Plinth accounts that are the
# same person (same name + email) share one chunk, while two
# different people on one company mailbox each get their own.
by_id, by_email_pl, seen_person = {}, {}, {}
for client in root.findall("client"):
cid = client.get("id")
if cid:
by_id[cid] = client
e = (client.findtext("email") or "").strip().lower()
if (cid or "").startswith("pl-"):
seen_person[(e, _name_of(client))] = client
elif e:
by_email_pl[e] = client
idx = index_existing_stubs()
plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
stamped = added = skipped = 0
for r in plan["rows"]:
# Map every partner client, whether or not they sit in a pipeline yet.
if r["assigned_tier"] in NON_PARTNER:
continue
email = (r.get("email") or "").strip().lower()
name = f"{(r.get('first_name') or '')} {(r.get('last_name') or '')}".strip().lower()
cid = _map_chunk_id(r)
def _stamp(el):
lm = el.find("last_modified")
if lm is None:
lm = ET.SubElement(el, "last_modified")
lm.text = TODAY
# 1) this exact Plinth account is already mapped (prior run) -> stamp
if cid and cid in by_id:
_stamp(by_id[cid])
stamped += 1
continue
# 2) already a pipeline-mapped (Bitrix) chunk for this email -> stamp, don't duplicate
if email and email in by_email_pl:
_stamp(by_email_pl[email])
stamped += 1
continue
# 3) same person we already added under another Plinth account -> stamp, don't duplicate
if (email, name) in seen_person:
_stamp(seen_person[(email, name)])
stamped += 1
continue
# 4) otherwise this partner client is not yet in the map -> add an empty-pipeline chunk
if not cid and not email:
skipped += 1 # nothing stable to key on; leave for a human
continue
chunk, e2, cid2 = _new_client_chunk(ET, r, idx)
root.append(chunk)
if cid2:
by_id[cid2] = chunk
seen_person[(e2, name)] = chunk
added += 1
ET.indent(tree, space=" ")
tree.write(XML_MAP, encoding="UTF-8", xml_declaration=True)
total = len(root.findall("client"))
if removed_merged:
print("MERGE REGISTRY: removed absorbed chunk(s) from the census: "
+ ", ".join(removed_merged))
print(f"XML: {stamped} existing chunks stamped, {added} new empty-pipeline chunks added "
f"(id=pl-<user_id>, <pipelines count=\"0\" />). "
f"Partner clients accounted for this run: {stamped + added}"
+ (f"; {skipped} skipped (no email/user_id)" if skipped else "")
+ f". Map now holds {total} <client> chunks. Backup: {backup}")
print("REMINDER: append one LEDGER entry for this change-set.")
def _regen_summary_clause(c):
s = c.find("summary")
if s is None or not s.text:
return
ps = c.findall("pipelines/pipeline")
NUM = {1: "one", 2: "two", 3: "three"}
if ps:
parts = [f"{p.findtext('display_name')} ({p.findtext('status')}, "
f"{p.findtext('stage')})" for p in ps]
n = len(parts)
clause = (f"Currently mapped to {NUM.get(n, n)} outreach pipeline"
f"{'s' if n > 1 else ''}: "
f"{' and '.join(parts) if n <= 2 else '; '.join(parts)}.")
else:
clause = ("Not currently mapped to any outreach pipeline (logged for "
"accounting; pipeline placement to be populated later).")
s.text = re.sub(r"(Not currently mapped to any outreach pipeline[^.]*\.|"
r"Currently mapped to [^:.]+:[^.]*\.)", clause, s.text, count=1)
def cmd_reconcile_deals(args):
"""Absorb a Bitrix deals pull (JSON: {"deals":[{deal_id,cat,stage,title,created}]})
into the XML census + stubs: one block per (client, funnel) recording the newest
deal. Deals with no / ambiguous name match are reported, never guessed."""
guard_or_exit()
import xml.etree.ElementTree as ET
from collections import defaultdict
deals = []
for path in args.deals:
deals += json.loads(Path(path).read_text(encoding="utf-8")).get("deals", [])
if not XML_MAP.exists():
sys.exit(f"XML map not found: {XML_MAP}")
backup = OUT_DIR / f"partner-client-master-list.backup-{TODAY}-reconcile.xml"
OUT_DIR.mkdir(parents=True, exist_ok=True)
shutil.copy2(XML_MAP, backup)
tree = ET.parse(XML_MAP)
root = tree.getroot()
byname = defaultdict(list)
for c in root.findall("client"):
byname[norm_key(c.findtext("full_name"))].append(c)
pair = defaultdict(list)
unmatched, ambiguous = [], []
for d in deals:
if d.get("cat") not in FUNNEL_KEYS:
continue
cand = byname.get(norm_key(d.get("title")), [])
if not cand:
unmatched.append(f"{d['cat']}/{d.get('stage')}/{d.get('deal_id')} {d.get('title')}")
continue
if len(cand) > 1:
withp = [x for x in cand if x.findall("pipelines/pipeline")]
if len(withp) == 1:
cand = withp
else:
ambiguous.append(f"{d.get('title')} -> "
f"{[x.get('id') for x in cand]} (deal {d.get('deal_id')})")
continue
pair[(id(cand[0]), FUNNEL_KEYS[d["cat"]])].append((cand[0], d))
new = updated = current = 0
touched = set()
def _dt(s):
m = re.search(r"([A-Z][a-z]{2}) (\d{1,2}) (\d{4})", s or "")
MON = {x: i + 1 for i, x in enumerate(
"Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split())}
if m:
return f"{m.group(3)}-{MON[m.group(1)]:02d}-{int(m.group(2)):02d}"
return (s or "")[:10] if re.match(r"\d{4}-\d{2}-\d{2}", s or "") else ""
for (_, key), items in pair.items():
items.sort(key=lambda x: int(x[1]["deal_id"]))
c, d = items[-1]
stage, status = stage_status(d["cat"], d["stage"])
pls = c.find("pipelines")
if pls is None:
pls = ET.SubElement(c, "pipelines")
blk = None
for p in pls.findall("pipeline"):
if (p.get("name") or p.get("key")) == key:
blk = p
break
if blk is not None:
old = (blk.findtext("bitrix_deal_id") or "").strip()
if old and int(old) > int(d["deal_id"]):
current += 1
continue
changed = False
for tag, val in (("bitrix_deal_id", d["deal_id"]), ("stage", stage),
("status", status), ("display_name", FUNNEL_DISPLAY[d["cat"]])):
e = blk.find(tag)
if e is None:
e = ET.SubElement(blk, tag)
if (e.text or "").strip() != val:
e.text = val
changed = True
ed = blk.find("entered_date")
if ed is None:
ed = ET.SubElement(blk, "entered_date")
if not (ed.text or "").strip():
ed.text = _dt(d.get("created"))
changed = changed or bool(ed.text)
if changed:
updated += 1
touched.add(id(c))
else:
current += 1
else:
p = ET.SubElement(pls, "pipeline", {"name": key})
ET.SubElement(p, "client_ref", {"id": c.get("id") or ""}).text = \
c.findtext("full_name")
for tag, val in (("display_name", FUNNEL_DISPLAY[d["cat"]]),
("bitrix_category_id", d["cat"]),
("bitrix_deal_id", d["deal_id"]), ("stage", stage),
("status", status), ("entered_date", _dt(d.get("created"))),
("last_touch", ""), ("next_step", ""), ("notes", "")):
e = ET.SubElement(p, tag)
if val:
e.text = val
new += 1
touched.add(id(c))
if id(c) in touched:
pls.set("count", str(len(pls.findall("pipeline"))))
_regen_summary_clause(c)
lm = c.find("last_modified")
if lm is None:
lm = ET.SubElement(c, "last_modified")
lm.text = TODAY
fn = c.findtext("stub/file_name")
if fn:
mirror_stub_pipeline(CLIENTS_DIR / fn, key, stage, status,
_dt(d.get("created")))
ET.indent(tree, space=" ")
tree.write(XML_MAP, encoding="UTF-8", xml_declaration=True)
print(f"RECONCILE: {new} new blocks, {updated} updated, {current} already current; "
f"{len(unmatched)} deals unmatched, {len(ambiguous)} ambiguous (not attached). "
f"Backup: {backup}")
for u in unmatched[:20]:
print(" unmatched:", u)
for a in ambiguous[:20]:
print(" ambiguous:", a)
print("REMINDER: append one LEDGER entry for this change-set.")