batch_web.py

skills/client-enrich/scripts/batch_web.py

1. #!/usr/bin/env python3
2. """
3. batch_web.py: write-back for web-search signals (client-enrich, no-credit mode).
5. For stubs where company LinkedIn posts were unavailable, this fills the Signals
6. section from web-search findings while preserving Person (from frontmatter) and
7. Company (minimal) content. Uses enrich_writeback.py so the vault contract holds.
9. Usage:
10. python3 batch_web.py --batch checkin_batch.json --signals web_signals.json \
11. --stubdir "/path/03-Clients/Check-In" --script "/path/enrich_writeback.py" [--date YYYY-MM-DD]
13. web_signals.json: { "Stub File.md": {"signals": ["- 2026: ...", ...],
14. "sources": ["- https://...", ...]}, ... }
15. signals/sources are pre-formatted markdown bullet lines. If a company yielded
16. nothing, pass {"signals": [], "sources": []} and a [TBD] line is written.
17. """
18. import argparse, json, os, subprocess, sys, datetime, tempfile
20. def main():
21. ap = argparse.ArgumentParser()
22. ap.add_argument("--batch", required=True)
23. ap.add_argument("--signals", required=True)
24. ap.add_argument("--stubdir", required=True)
25. ap.add_argument("--script", required=True)
26. ap.add_argument("--date", default=datetime.date.today().isoformat())
27. args = ap.parse_args()
29. batch = {r["file"]: r for r in json.load(open(args.batch, encoding="utf-8"))}
30. sig = json.load(open(args.signals, encoding="utf-8"))
32. rows = []
33. for fname, data in sig.items():
34. r = batch.get(fname, {"file": fname})
35. person = []
36. if r.get("title"): person.append(f"- Title: {r['title']}")
37. if r.get("seniority"): person.append(f"- Seniority: {r['seniority']}")
38. loc = ", ".join([x for x in [r.get("city"), r.get("state")] if x])
39. if loc: person.append(f"- Location: {loc}")
40. if r.get("linkedin"): person.append(f"- LinkedIn: {r['linkedin']}")
41. person = person or ["[TBD]"]
43. company = []
44. if r.get("company"): company.append(f"- {r['company']}")
45. if r.get("company_li"): company.append(f"- LinkedIn: {r['company_li']}")
46. company.append("- [TBD: firmographics (revenue/headcount/funding) not pulled, Apollo skipped this run]")
48. signals = list(data.get("signals") or [])
49. sources = list(data.get("sources") or [])
50. if not signals:
51. signals = ["[TBD: no recent public news found via web search]"]
52. sources.append(f"- Web search, {args.date}")
53. sources.append("- Stub frontmatter (prior enrichment)")
55. payload = {"frontmatter": {}, "person": person, "company": company,
56. "signals": signals, "sources": sources}
57. with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as tf:
58. json.dump(payload, tf); pj = tf.name
59. res = subprocess.run([sys.executable, args.script, "--stub",
60. os.path.join(args.stubdir, fname), "--payload", pj,
61. "--date", args.date], capture_output=True, text=True)
62. os.unlink(pj)
63. rows.append((fname, len([s for s in signals if not s.startswith("[TBD")]),
64. res.returncode == 0, res.stderr.strip()[:120]))
66. ok = sum(1 for _,_,k,_ in rows if k)
67. print(f"WROTE {ok}/{len(rows)} stubs OK")
68. for f, n, k, err in rows:
69. print(f" {f[:34]:36} | {n} sig | {'OK' if k else 'FAIL: '+err}")
71. if __name__ == "__main__":
72. main()