batch_enrich.py

skills/client-enrich/scripts/batch_enrich.py

1. #!/usr/bin/env python3
2. """
3. batch_enrich.py: deterministic batch write-back for client-enrich (posts + frontmatter mode).
5. Takes a list of stubs and a flat list of LinkedIn company posts, groups posts by
6. company slug, auto-selects the most outreach-relevant ones, builds a Research
7. payload per stub (Person from existing frontmatter, Company minimal, Signals from
8. posts, Sources), and calls enrich_writeback.py for each stub so the vault contract
9. is enforced (writes only under ## Research, never Human Notes).
11. Usage:
12. python3 batch_enrich.py --batch checkin_batch.json --posts posts.json \
13. --stubdir "/path/to/03-Clients/Check-In" --script "/path/to/enrich_writeback.py" [--date YYYY-MM-DD]
15. checkin_batch.json: [{file,email,linkedin,company,company_li,slug,title,seniority,city,state}, ...]
16. posts.json: [{targetUrl, content, date, type, authorName, authorType, repostedByName, linkedinUrl}, ...]
17. """
18. import argparse, json, os, re, subprocess, sys, datetime, tempfile
20. REL_KEYWORDS = [
21. "launch", "launching", "introducing", "new product", "now available", "now at",
22. "hiring", "we're hiring", "join our team", "join us", "open role", "opening",
23. "opened", "grand opening", "expansion", "expanding", "expand", "new location",
24. "event", "festival", "conference", "expo", "trade show", "tradeshow", "booth",
25. "sponsor", "sponsoring", "award", "awarded", "won", "ranked", "honored",
26. "milestone", "anniversary", "celebrat", "partnership", "partnered", "partner with",
27. "funding", "raised", "series a", "series b", "series c", "acquisition", "acquire",
28. "store", "retail", "ribbon", "exhibit", "summit", "kickoff", "unveil", "debut",
29. ]
31. def slug_of(url):
32. m = re.search(r'/company/([^/?]+)', url or "")
33. return m.group(1).lower() if m else ""
35. def short_date(iso):
36. try:
37. return datetime.datetime.fromisoformat(iso.replace("Z", "+00:00")).strftime("%b %Y")
38. except Exception:
39. return (iso or "")[:7]
41. def one_line(text, limit=190):
42. t = re.sub(r'\s+', ' ', (text or "").strip())
43. if len(t) > limit:
44. cut = t[:limit]
45. # try to end on sentence/word boundary
46. for sep in (". ", "! ", "? "):
47. i = cut.rfind(sep)
48. if i > 60:
49. return cut[:i+1].strip()
50. return cut.rsplit(" ", 1)[0].strip() + "…"
51. return t
53. def score(post, company_name):
54. c = (post.get("content") or "").lower()
55. s = 0
56. for kw in REL_KEYWORDS:
57. if kw in c:
58. s += 3
59. # prefer company-authored over reposts / individual profiles
60. if (post.get("authorType") == "company"):
61. s += 4
62. if post.get("type") == "post":
63. s += 1
64. if post.get("repostedByName"):
65. s -= 1
66. # mild recency tiebreak handled by sort on date
67. if not c.strip():
68. s -= 5
69. return s
71. def select_posts(posts, company_name, n=3):
72. ranked = sorted(posts, key=lambda p: (score(p, company_name), p.get("date") or ""), reverse=True)
73. out = [p for p in ranked if (p.get("content") or "").strip()][:n]
74. return out
76. def main():
77. ap = argparse.ArgumentParser()
78. ap.add_argument("--batch", required=True)
79. ap.add_argument("--posts", required=True)
80. ap.add_argument("--stubdir", required=True)
81. ap.add_argument("--script", required=True)
82. ap.add_argument("--date", default=datetime.date.today().isoformat())
83. args = ap.parse_args()
85. batch = json.load(open(args.batch, encoding="utf-8"))
86. raw = json.load(open(args.posts, encoding="utf-8"))
87. # accept: flat list of normalized dicts, raw harvestapi items, or {items:[...]}
88. if isinstance(raw, dict) and "items" in raw:
89. raw = raw["items"]
90. posts = []
91. for p in raw:
92. if "targetUrl" in p: # already normalized
93. posts.append(p); continue
94. posts.append({
95. "targetUrl": (p.get("query") or {}).get("targetUrl", ""),
96. "content": p.get("content", ""),
97. "date": (p.get("postedAt") or {}).get("date", ""),
98. "type": p.get("type", ""),
99. "authorName": (p.get("author") or {}).get("name", ""),
100. "authorType": (p.get("author") or {}).get("type", ""),
101. "repostedByName": (p.get("repostedBy") or {}).get("name", ""),
102. "linkedinUrl": p.get("linkedinUrl", ""),
103. })
105. by_slug = {}
106. for p in posts:
107. s = slug_of(p.get("targetUrl"))
108. if s:
109. by_slug.setdefault(s, []).append(p)
111. summary = []
112. for r in batch:
113. stub_path = os.path.join(args.stubdir, r["file"])
114. # Person from existing frontmatter (free, no API)
115. person = []
116. if r.get("title"): person.append(f"- Title: {r['title']}")
117. if r.get("seniority"): person.append(f"- Seniority: {r['seniority']}")
118. loc = ", ".join([x for x in [r.get("city"), r.get("state")] if x])
119. if loc: person.append(f"- Location: {loc}")
120. if r.get("linkedin"): person.append(f"- LinkedIn: {r['linkedin']}")
121. person = person or ["[TBD]"]
123. # Company minimal (Apollo skipped this run)
124. company = []
125. if r.get("company"): company.append(f"- {r['company']}")
126. if r.get("company_li"): company.append(f"- LinkedIn: {r['company_li']}")
127. company.append("- [TBD: firmographics (revenue/headcount/funding) not pulled, Apollo skipped this run]")
129. # Signals from posts
130. slug = r.get("slug", "")
131. chosen = select_posts(by_slug.get(slug, []), r.get("company", ""), n=3) if slug else []
132. signals, sources = [], []
133. for p in chosen:
134. signals.append(f"- {short_date(p.get('date'))}: {one_line(p.get('content'))}")
135. if p.get("linkedinUrl"):
136. sources.append(f"- LinkedIn post ({short_date(p.get('date'))}): {p['linkedinUrl']}")
137. if not signals:
138. signals = ["[TBD: no recent company LinkedIn posts found in last 3 months]"]
139. sources.append(f"- Company LinkedIn posts via Apify harvestapi/linkedin-company-posts, {args.date}")
140. sources.append(f"- Stub frontmatter (prior enrichment)")
142. payload = {"frontmatter": {}, "person": person, "company": company,
143. "signals": signals, "sources": sources}
144. with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as tf:
145. json.dump(payload, tf)
146. pjson = tf.name
147. res = subprocess.run([sys.executable, args.script, "--stub", stub_path,
148. "--payload", pjson, "--date", args.date],
149. capture_output=True, text=True)
150. os.unlink(pjson)
151. ok = res.returncode == 0
152. summary.append((r["file"], len([s for s in signals if not s.startswith("[TBD")]), ok,
153. res.stderr.strip()[:120]))
155. good = sum(1 for _,_,ok,_ in summary if ok)
156. withsig = sum(1 for _,n,_,_ in summary if n > 0)
157. print(f"WROTE {good}/{len(summary)} stubs OK; {withsig} have >=1 real signal.")
158. print("\nstub | signals | status")
159. for f, n, ok, err in summary:
160. print(f" {f[:34]:36} | {n} | {'OK' if ok else 'FAIL: '+err}")
162. if __name__ == "__main__":
163. main()