patch_research.py

skills/client-enrich/scripts/patch_research.py

1. #!/usr/bin/env python3
2. """
3. patch_research.py: surgically patch individual ## Research subsections of a stub,
4. preserving the ones you don't touch. Complements enrich_writeback.py (which rebuilds
5. the whole block). Honors the contract: only ## Research + frontmatter change; never
6. Human Notes.
8. Patch JSON (all optional):
9. "frontmatter": {k:v} fill EMPTY/[TBD] frontmatter keys only (unless --force)
10. "person": [lines] REPLACE the Person / LinkedIn subsection
11. "company": [lines] REPLACE the Company subsection
12. "signals": [lines] REPLACE the Recent News & Signals subsection
13. "signals_append": [lines] APPEND to signals (removes a lone [TBD] placeholder first)
14. "sources_append": [lines] APPEND to Sources, skipping lines already there
15. Subsections not named are left byte-for-byte intact.
17. Usage: python3 patch_research.py --stub PATH --patch patch.json [--force] [--date YYYY-MM-DD]
18. """
20. import argparse
21. import datetime
22. import json
23. import re
24. import sys
26. from _stub import fm_set, split_frontmatter
28. HEADINGS = [
29. "### Company",
30. "### Person / LinkedIn",
31. "### Recent News & Signals",
32. "### Sources",
33. ]
34. KEYMAP = {
35. "### Company": "company",
36. "### Person / LinkedIn": "person",
37. "### Recent News & Signals": "signals",
38. "### Sources": "sources",
39. }
40. FM_KEYS = [
41. "title",
42. "seniority",
43. "city",
44. "state",
45. "linkedin",
46. "company",
47. "company_linkedin",
48. "industry",
49. "company_website",
50. ]
53. def parse_research(body):
54. m = re.search(r"(?m)^## Research\s*$", body)
55. if not m:
56. sys.exit("ERROR: no ## Research")
57. start = m.start()
58. nxt = re.search(r"(?m)^## (?!Research).*$", body[start + 1 :])
59. if not nxt:
60. sys.exit("ERROR: no section after Research")
61. end = start + 1 + nxt.start()
62. block = body[start:end]
63. # comment line right after ## Research (optional)
64. lines = block.splitlines()
65. comment = ""
66. idx = 1
67. if idx < len(lines) and lines[idx].strip().startswith("<!--"):
68. comment = lines[idx]
69. idx += 1
70. subs = {}
71. cur = None
72. for l in lines[idx:]:
73. if l.strip() in HEADINGS:
74. cur = l.strip()
75. subs[cur] = []
76. elif cur is not None:
77. subs[cur].append(l)
78. # trim trailing blank lines per sub
79. for k in subs:
80. while subs[k] and subs[k][-1].strip() == "":
81. subs[k].pop()
82. return start, end, comment, subs
85. def main():
86. ap = argparse.ArgumentParser()
87. ap.add_argument("--stub", required=True)
88. ap.add_argument("--patch", required=True)
89. ap.add_argument("--force", action="store_true")
90. ap.add_argument("--date", default=datetime.date.today().isoformat())
91. a = ap.parse_args()
92. text = open(a.stub, encoding="utf-8").read()
93. patch = json.load(open(a.patch, encoding="utf-8"))
94. fm, body = split_frontmatter(text)
95. human = body.split("## Human Notes", 1)
96. human_snap = ("## Human Notes" + human[1]) if len(human) > 1 else None
98. changed = []
99. for k, v in (patch.get("frontmatter") or {}).items():
100. if k in FM_KEYS and str(v).strip():
101. fm, did = fm_set(fm, k, str(v).strip(), a.force, literal=False)
102. if did:
103. changed.append(k)
104. # Stamp the file's existing date key (vault stubs use last_modified); never add a second.
105. for _k in ("last_modified", "updated"):
106. if re.search(rf"(?m)^{_k}:", fm):
107. fm = re.sub(rf"(?m)^{_k}:.*$", f"{_k}: '{a.date}'", fm, count=1)
108. break
110. start, end, comment, subs = parse_research(body)
112. def setsub(h, lines):
113. subs[h] = list(lines)
115. if "person" in patch:
116. setsub("### Person / LinkedIn", patch["person"])
117. if "company" in patch:
118. setsub("### Company", patch["company"])
119. if "signals" in patch:
120. setsub("### Recent News & Signals", patch["signals"])
121. if "signals_append" in patch:
122. cur = [
123. l
124. for l in subs.get("### Recent News & Signals", [])
125. if "[TBD" not in l and l.strip()
126. ]
127. setsub("### Recent News & Signals", cur + patch["signals_append"])
128. if "sources_append" in patch:
129. cur = subs.get("### Sources", [])
130. have = set(x.strip() for x in cur)
131. setsub(
132. "### Sources",
133. cur + [s for s in patch["sources_append"] if s.strip() not in have],
134. )
136. out = ["## Research"]
137. if comment:
138. out.append(comment)
139. for h in HEADINGS:
140. out.append(h)
141. ls = subs.get(h, [])
142. out.extend(ls if ls else (["[TBD]"] if h != "### Sources" else []))
143. newblock = "\n".join(out) + "\n"
144. body = body[:start] + newblock + "\n" + body[end:]
146. if human_snap is not None:
147. after = body.split("## Human Notes", 1)
148. if ("## Human Notes" + after[1] if len(after) > 1 else None) != human_snap:
149. sys.exit("ABORT: Human Notes would change")
150. open(a.stub, "w", encoding="utf-8").write(f"---\n{fm}\n---\n{body}")
151. touched = set(patch)
152. if "signals_append" in patch:
153. touched.add("signals")
154. if "sources_append" in patch:
155. touched.add("sources")
156. patched = ",".join(KEYMAP[h] for h in HEADINGS if KEYMAP[h] in touched)
157. print(
158. f"OK {a.stub.split('/')[-1]} | fm:{','.join(changed) or '-'} | patched:{patched}"
159. )
162. if __name__ == "__main__":
163. main()