510 lines
20 KiB
Python
510 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""Regenerate a cover letter using an LLM, incorporating user feedback.
|
||
Called by the dashboard server when the user clicks "Regenerate".
|
||
|
||
Args:
|
||
job_id: Notion page ID of the job
|
||
feedback: User's instructions for the revision
|
||
|
||
Outputs:
|
||
Prints JSON with {"url": "...", "file": "..."} on success
|
||
Prints JSON with {"error": "..."} on failure
|
||
"""
|
||
|
||
import json, os, sys, re, subprocess
|
||
from datetime import datetime
|
||
|
||
HERMES_DIR = os.path.expanduser("~/.hermes")
|
||
DRAFTS_DIR = os.path.join(HERMES_DIR, "radar-ref/drafts")
|
||
RESUMES_DIR = os.path.join(HERMES_DIR, "radar-ref/resumes")
|
||
COVER_LETTER_DIR = os.path.join(HERMES_DIR, "radar-ref/cover-letters")
|
||
CACHE_FILE = os.path.join(HERMES_DIR, "dashboard/jobs_cache.json")
|
||
|
||
# ── Load context ───────────────────────────────────────────────────────────
|
||
|
||
def load_job(job_id):
|
||
"""Look up a job by its Notion page ID in the cache."""
|
||
with open(CACHE_FILE) as f:
|
||
cache = json.load(f)
|
||
for job in cache.get("jobs", []):
|
||
if job["id"] == job_id:
|
||
return job
|
||
return None
|
||
|
||
def read_file_first_line_or_none(path):
|
||
if os.path.isfile(path):
|
||
with open(path) as f:
|
||
return f.read().strip()
|
||
return None
|
||
|
||
def find_first_file(dir_path):
|
||
"""Return first .pdf or .txt in a directory, or None."""
|
||
if not os.path.isdir(dir_path):
|
||
return None
|
||
for f in sorted(os.listdir(dir_path)):
|
||
if f.endswith((".pdf", ".txt", ".md")):
|
||
return os.path.join(dir_path, f)
|
||
return None
|
||
|
||
def slugify(text):
|
||
"""Turn company + role into a safe filename."""
|
||
s = text.lower().strip()
|
||
s = re.sub(r'[^\w\s-]', '', s)
|
||
s = re.sub(r'[-\s]+', '-', s)
|
||
return s[:80]
|
||
|
||
# ── LLM call ───────────────────────────────────────────────────────────────
|
||
|
||
def call_llm(system_prompt, user_prompt):
|
||
"""Call the OmniRoute cover-letter-writer model combo. Returns response text."""
|
||
payload = {
|
||
"model": "cover-letter-writer",
|
||
"stream": False,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt}
|
||
],
|
||
"temperature": 0.7,
|
||
"max_tokens": 2000
|
||
}
|
||
cmd = [
|
||
"curl", "-s",
|
||
"http://omniroute:20128/v1/chat/completions",
|
||
"-H", "Content-Type: application/json",
|
||
"-H", "Authorization: Bearer sk-1234",
|
||
"-d", json.dumps(payload)
|
||
]
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
|
||
try:
|
||
data = json.loads(result.stdout)
|
||
return data["choices"][0]["message"]["content"]
|
||
except (KeyError, json.JSONDecodeError) as e:
|
||
error_text = result.stdout[:500] if result.stdout.strip() else "Empty response"
|
||
return f"[Error: {e} — {error_text}]"
|
||
|
||
def parse_cover_letter(text, company="", role="", location=""):
|
||
"""Parse cover letter text into structured parts for HTML template."""
|
||
lines = text.strip().split('\n')
|
||
|
||
# Find subject line
|
||
subject = ""
|
||
first_idx = 0
|
||
for i, line in enumerate(lines):
|
||
s = line.strip()
|
||
if s.startswith("Subject:") or (s.startswith("**Subject:") and s.endswith("**")):
|
||
subject = s.replace("**","").replace("Subject:","").replace("**","").strip()
|
||
first_idx = i + 1
|
||
break
|
||
|
||
# Find opening paragraph (between salutation and bullets)
|
||
opening = ""
|
||
bullets = []
|
||
closing = ""
|
||
|
||
# Skip everything before "Dear Hiring Manager," — contact block + date belong
|
||
# in the template header, not in the parsed opening paragraph.
|
||
salutation_idx = None
|
||
for i, line in enumerate(lines[first_idx:], start=first_idx):
|
||
if line.strip().lower().startswith("dear ") and "hiring manager" in line.lower():
|
||
salutation_idx = i
|
||
break
|
||
|
||
parse_start = salutation_idx + 1 if salutation_idx is not None else first_idx
|
||
|
||
in_bullets = False
|
||
bridging = False
|
||
for line in lines[parse_start:]:
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
continue
|
||
|
||
# Detect bullet markers
|
||
is_bullet = (
|
||
stripped.startswith('-') or
|
||
stripped.startswith('•') or
|
||
(len(stripped) > 2 and stripped[0].isdigit() and stripped[1] in '.):')
|
||
)
|
||
|
||
# Detect salutation
|
||
if stripped.startswith("Dear ") and "Dear Hiring Manager" in stripped:
|
||
in_bullets = False
|
||
bridging = False
|
||
continue
|
||
|
||
if is_bullet:
|
||
in_bullets = True
|
||
bridging = False
|
||
# Clean bullet marker
|
||
bullet_text = stripped.lstrip('-• ').strip()
|
||
# Remove markdown bold markers per-line so the plain letter reads cleanly
|
||
bullet_text = bullet_text.replace('**', '')
|
||
# Remove leading numbering like "1. ", "2. "
|
||
if len(bullet_text) > 2 and bullet_text[0].isdigit() and bullet_text[1] == '.':
|
||
bullet_text = bullet_text[2:].strip()
|
||
bullets.append(bullet_text)
|
||
elif in_bullets and not is_bullet:
|
||
in_bullets = False
|
||
bridging = False
|
||
if not stripped.startswith("Dear ") and not stripped.startswith("Dear Hiring"):
|
||
closing += stripped + " "
|
||
elif not in_bullets and not stripped.startswith("Dear ") and not stripped.startswith("Dear Hiring"):
|
||
# Already past bullets — rest belongs to closing, not opening
|
||
if closing:
|
||
if "regards" in stripped.lower() or stripped.startswith("Sincerely") or stripped.startswith("Kind regards"):
|
||
continue
|
||
if stripped.lower() in ["sincerely,", "kind regards,", "regards,"] or stripped.startswith("Anthony Martin"):
|
||
continue
|
||
closing += stripped + " "
|
||
continue
|
||
# Still in opening section (no bullets seen yet)
|
||
if "regards" in stripped.lower() or stripped.startswith("Sincerely") or stripped.startswith("Kind regards"):
|
||
continue
|
||
if stripped.lower() in ["sincerely,", "kind regards,", "regards,"] or stripped.startswith("Anthony Martin"):
|
||
continue
|
||
# Bridge text before bullets: drop it
|
||
bridging_texts = ["my experience includes:", "my experience includes",
|
||
"these include:", "these include",
|
||
"key highlights:", "key highlights",
|
||
"relevant experience includes:", "relevant experience includes",
|
||
"relevant achievements include:", "relevant achievements include",
|
||
"my relevant experience includes:"]
|
||
lower = stripped.lower().rstrip(":")
|
||
if lower in bridging_texts or any(stripped.lower().startswith(b) for b in ["my experience ", "i bring ", "i would ", "relevant experience ", "relevant achievements ", "key achievements "]):
|
||
bridging = True
|
||
continue
|
||
if bridging:
|
||
continue
|
||
opening += stripped + " "
|
||
# If the line just added ends with a bridge phrase, strip it
|
||
opening_lower = opening.lower()
|
||
for bt in bridging_texts:
|
||
bt_stripped = bt.rstrip(":")
|
||
if opening_lower.strip().rstrip(".;: ").endswith(bt_stripped):
|
||
# Find actual position of bridge text in opening, then strip it
|
||
idx = opening_lower.rfind(bt_stripped)
|
||
if idx >= 0:
|
||
opening = opening[:idx].strip() + " "
|
||
break
|
||
|
||
if not opening and len(lines) > 3:
|
||
opening = lines[3].strip() if len(lines) > 3 else ""
|
||
|
||
# Remove markdown bold markers from opening/closing
|
||
opening = opening.replace('**', '')
|
||
closing = closing.replace('**', '')
|
||
|
||
return {
|
||
"subject": subject or f"Application for {role} at {company}",
|
||
"opening": opening.strip()[:800],
|
||
"bullets": bullets[:5],
|
||
"closing": closing.strip(),
|
||
"date": datetime.now().strftime("%d %B %Y"),
|
||
"company": company,
|
||
"role": role,
|
||
"location": location or "Perth, WA"
|
||
}
|
||
|
||
|
||
def render_cover_letter_html(structured, txt_url):
|
||
"""Render structured cover letter data as HTML using the template."""
|
||
try:
|
||
template_path = os.path.expanduser("~/.hermes/dashboard/templates/cover-letter.html")
|
||
with open(template_path) as f:
|
||
template = f.read()
|
||
|
||
# Build recipient block
|
||
recipient = [structured.get("company", ""), structured.get("location", "Perth, WA")]
|
||
|
||
# Format bullets
|
||
bullets_html = ""
|
||
for bullet in structured.get("bullets", []):
|
||
bullets_html += f"<li>{bullet}</li>\n "
|
||
if not bullets_html:
|
||
bullets_html = "<li>" + structured.get("opening", "")[:200] + "</li>"
|
||
|
||
# Replace template variables
|
||
html = template.replace("{{title}}", structured.get("subject", "Cover Letter"))
|
||
html = html.replace("{{date}}", structured.get("date", ""))
|
||
html = html.replace("{{recipient_block}}", "\n".join(recipient))
|
||
html = html.replace("{{opening}}", structured.get("opening", ""))
|
||
html = html.replace("{{bullets}}", bullets_html)
|
||
html = html.replace("{{closing}}", structured.get("closing", ""))
|
||
html = html.replace("{{txt_url}}", txt_url)
|
||
html = html.replace("{{cover_letter_docx_url}}", txt_url.replace(".txt", ".docx"))
|
||
|
||
return html
|
||
except Exception as e:
|
||
return None
|
||
def update_notion_cover_letter(page_id, url):
|
||
"""Update the 'Cover Letter' URL property on a Notion page."""
|
||
env_path = os.path.expanduser("~/.hermes/.env")
|
||
key = None
|
||
with open(env_path) as f:
|
||
for line in f:
|
||
if line.startswith("NOTION_API_KEY="):
|
||
key = line.split("=", 1)[1].strip()
|
||
break
|
||
if not key:
|
||
return False
|
||
|
||
payload = {
|
||
"properties": {
|
||
"Cover Letter": {
|
||
"type": "url",
|
||
"url": url
|
||
}
|
||
}
|
||
}
|
||
cmd = [
|
||
"curl", "-s", "-X", "PATCH",
|
||
f"https://api.notion.com/v1/pages/{page_id}",
|
||
"-H", f"Authorization: Bearer {key}",
|
||
"-H", "Content-Type: application/json",
|
||
"-H", "Notion-Version: 2022-06-28",
|
||
"-d", json.dumps(payload)
|
||
]
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
|
||
try:
|
||
data = json.loads(result.stdout)
|
||
return data.get("object") == "page"
|
||
except:
|
||
return False
|
||
|
||
# ── Main ───────────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
if len(sys.argv) < 3:
|
||
print(json.dumps({"error": "Usage: regenerate_cover_letter.py <job_id> <feedback>"}))
|
||
sys.exit(1)
|
||
|
||
job_id = sys.argv[1]
|
||
feedback = sys.argv[2]
|
||
|
||
job = load_job(job_id)
|
||
if not job:
|
||
print(json.dumps({"error": f"Job {job_id} not found in cache"}))
|
||
sys.exit(1)
|
||
|
||
company = job.get("name", "Unknown")
|
||
role = job.get("role", "Unknown")
|
||
location = job.get("location", "Perth, WA")
|
||
salary = job.get("salary")
|
||
salary_str = f"${salary:,}" if salary else "competitive"
|
||
notes = job.get("notes", "")
|
||
|
||
# Build the filename
|
||
safe_name = slugify(f"{company}-{role}")
|
||
ts = datetime.now().strftime("%Y%m%d")
|
||
filename = f"cl-{safe_name}-{ts}.txt"
|
||
filepath = os.path.join(DRAFTS_DIR, filename)
|
||
|
||
# Read resume (just the first one found)
|
||
resume_file = find_first_file(RESUMES_DIR)
|
||
resume_text = ""
|
||
if resume_file:
|
||
if resume_file.endswith(".txt"):
|
||
with open(resume_file) as f:
|
||
resume_text = f.read()
|
||
elif resume_file.lower().endswith(".pdf"):
|
||
try:
|
||
import pdfplumber
|
||
with pdfplumber.open(resume_file) as pdf:
|
||
pages = [p.extract_text() or "" for p in pdf.pages]
|
||
resume_text = "\n".join(pages)
|
||
except Exception:
|
||
resume_text = f"[Resume file: {os.path.basename(resume_file)} — pdfplumber unavailable]"
|
||
elif resume_file.lower().endswith(".docx"):
|
||
try:
|
||
from docx import Document
|
||
doc = Document(resume_file)
|
||
resume_text = "\n".join(p.text for p in doc.paragraphs)
|
||
except Exception:
|
||
resume_text = f"[Resume file: {os.path.basename(resume_file)} — python-docx unavailable]"
|
||
|
||
# Read existing cover letter if re-generating
|
||
existing_cl = ""
|
||
|
||
# System prompt for cover letter generation
|
||
system_prompt = """You are a professional cover letter writer. Write concise,
|
||
impactful cover letters for Anthony Martin, a Perth-based marketing and communications
|
||
professional.
|
||
|
||
STYLE GUIDE (from Anthony's real cover letters):
|
||
- Professional, warm, and direct — not overly formal or stiff
|
||
- Open with a personal connection or belief statement about the company/role
|
||
- Use bullet points (3-4) with specific metrics and achievements
|
||
- Reference relevant experience from this resume only
|
||
|
||
ANTHONY'S ACTUAL CAREER HISTORY (ABSOLUTE CONSTRAINT — DO NOT DEVIATE):
|
||
- Pacific Energy (Kewdale, Perth, WA): Jun 2025 – Jul 2026 — Marketing Advisor. Corporate marketing, digital content, technical communications, major events, marketing AI strategy. Led website that won Science and Sustainability Website of the Year at 2026 Australian Web Awards.
|
||
- Horizon Power (Perth, WA): 2019 – Sep 2024 — Marketing Specialist. Annual budgets >$2M, 80%+ customer satisfaction, WA EV Network marketing, Bright Horizons STEM program, regulatory compliance communications.
|
||
- Synergy (Perth, WA): 2010 – 2018 — Multiple commercial/marketing roles. SolarReturn (40% sales increase), Synergy Sellback Program (150% above target), $2M+ billing resolution, 1M+ customer communications framework.
|
||
|
||
CRITICAL RULES:
|
||
- The THREE employers above are the ONLY ones Anthony has worked for.
|
||
- NEVER mention Lotterywest, Tourism WA, Water Corporation, Department of Creative Industries, Empire Home, SVARA, Lenox Hill, Fairclough, Recruit Right, a&co Recruitment, Catholic Education WA, THE resources HUB — or any other employer — even as examples, comparisons, or "organisations including..."
|
||
- If you list employers in the opening paragraph, ONLY list: Pacific Energy, Horizon Power, Synergy.
|
||
- Treat any employer not explicitly listed here as FORBIDDEN.
|
||
|
||
KEY ACHIEVEMENTS (use these exact metrics):
|
||
- Pacific Energy 2026 Australian Web Awards — Science and Sustainability Website of the Year
|
||
- Managed annual marketing budgets >$2M at Horizon Power
|
||
- 80%+ customer satisfaction ratings
|
||
- 40% sales increase (SolarReturn at Synergy, year one)
|
||
- 150% above target (Synergy Sellback Program sales)
|
||
- $1.3M billing discrepancy resolution at Synergy
|
||
- 35% improvement in conversion rates
|
||
- 20% reduction in inquiry-to-sale timeframes
|
||
- Company-wide communications framework reaching 1M+ customers
|
||
- WA EV Network marketing leadership
|
||
- Created Bright Horizons STEM program (remote schools)
|
||
|
||
BULLET STRUCTURE:
|
||
- 3-4 bullets max
|
||
- Each bullet: action verb + context + measurable outcome
|
||
- Draw from the real achievements above — do NOT invent companies, roles, or metrics
|
||
|
||
FORMAT:
|
||
- Subject line with the role and company
|
||
- Salutation (Dear Hiring Manager,)
|
||
- Opening paragraph — why this role + company (1-2 sentences)
|
||
- 3-4 bullet point achievements (draw from actual experience)
|
||
- Closing paragraph — enthusiasm + call to action (1-2 sentences)
|
||
- Signature: Anthony Martin"""
|
||
|
||
user_prompt = f"""Write a tailored cover letter for this role:
|
||
|
||
COMPANY: {company}
|
||
ROLE: {role}
|
||
LOCATION: {location}
|
||
SALARY: {salary_str}
|
||
ADDITIONAL NOTES: {notes}
|
||
|
||
RESUME HIGHLIGHTS:
|
||
{resume_text}
|
||
|
||
FEEDBACK / REVISION INSTRUCTIONS:
|
||
{feedback}
|
||
|
||
IMPORTANT: Return ONLY the cover letter text — no commentary, no meta text. The letter must be ready to send."""
|
||
|
||
# Call LLM
|
||
cl_text = call_llm(system_prompt, user_prompt)
|
||
|
||
# Save
|
||
os.makedirs(DRAFTS_DIR, exist_ok=True)
|
||
with open(filepath, "w") as f:
|
||
f.write(cl_text)
|
||
|
||
# Build the URLs (both .txt raw and rendered HTML)
|
||
tailscale_host = "https://hermes.kangaroo-eel.ts.net"
|
||
base = f"cl-{safe_name}-{ts}"
|
||
cl_url_txt = f"{tailscale_host}/jobs/drafts/{base}.txt"
|
||
cl_url_html = f"{tailscale_host}/jobs/drafts/{base}.html"
|
||
|
||
# Try to parse structured JSON from the plain text
|
||
structured = parse_cover_letter(cl_text, company, role, location)
|
||
|
||
# Save structured JSON for template rendering
|
||
json_path = os.path.join(DRAFTS_DIR, base + ".json")
|
||
with open(json_path, "w") as f:
|
||
json.dump(structured, f, indent=2)
|
||
|
||
# Render HTML version if we have structured data
|
||
html_path = os.path.join(DRAFTS_DIR, base + ".html")
|
||
html_doc = render_cover_letter_html(structured, cl_url_txt)
|
||
if html_doc:
|
||
with open(html_path, "w") as f:
|
||
f.write(html_doc)
|
||
|
||
# Generate .docx using python-docx builder
|
||
docx_script = os.path.expanduser("~/.hermes/scripts/build_cover_letter_docx.py")
|
||
docx_result = subprocess.run(
|
||
["python3", docx_script, base],
|
||
capture_output=True, text=True, timeout=30, cwd=DRAFTS_DIR
|
||
)
|
||
if docx_result.returncode != 0:
|
||
print(f"[Warning] DOCX generation failed: {docx_result.stderr[:200]}")
|
||
|
||
# Build the URLs (both .txt raw and rendered HTML)
|
||
tailscale_host = "https://hermes.kangaroo-eel.ts.net"
|
||
cl_url_txt = f"{tailscale_host}/jobs/drafts/{base}.txt"
|
||
cl_url_html = f"{tailscale_host}/jobs/drafts/{base}.html"
|
||
cl_url_docx = f"{tailscale_host}/jobs/drafts/{base}.docx"
|
||
|
||
# Update Notion with the HTML URL (preferred view)
|
||
update_notion_cover_letter(job_id, cl_url_html)
|
||
|
||
print(json.dumps({
|
||
"url": cl_url_html,
|
||
"txt_url": cl_url_txt,
|
||
"docx_url": cl_url_docx,
|
||
"file": filepath,
|
||
"company": company,
|
||
"role": role,
|
||
"structured": structured is not None
|
||
}))
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|
||
|
||
def backfill_from_txt(txt_path):
|
||
"""Regenerate HTML/DOCX from an existing .txt file using the canonical parser."""
|
||
import subprocess
|
||
|
||
with open(txt_path, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
|
||
# Parse using canonical function
|
||
structured = parse_cover_letter(text, company="", role="", location="")
|
||
|
||
base = txt_path[:-4]
|
||
json_path = base + ".json"
|
||
html_path = base + ".html"
|
||
docx_path = base + ".docx"
|
||
|
||
# Add URLs
|
||
txt_url = f"/jobs/drafts/{os.path.basename(txt_path)}"
|
||
structured["txt_url"] = txt_url
|
||
structured["html_url"] = txt_url.replace(".txt", ".html")
|
||
structured["docx_url"] = txt_url.replace(".txt", ".docx")
|
||
|
||
# Save JSON
|
||
with open(json_path, "w", encoding="utf-8") as f:
|
||
json.dump(structured, f, indent=2)
|
||
|
||
# Render HTML using canonical function
|
||
html = render_cover_letter_html(structured, txt_url)
|
||
if html:
|
||
with open(html_path, "w", encoding="utf-8") as f:
|
||
f.write(html)
|
||
|
||
# Generate DOCX
|
||
docx_script = os.path.expanduser("~/.hermes/scripts/build_cover_letter_docx.py")
|
||
subprocess.run(["python3", docx_script, json_path], capture_output=True, timeout=60)
|
||
|
||
return os.path.exists(html_path) and os.path.exists(docx_path)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
|
||
# Check for backfill mode
|
||
if len(sys.argv) == 2 and sys.argv[1].endswith(".txt"):
|
||
txt_file = sys.argv[1]
|
||
print(f"Backfilling {txt_file}...")
|
||
success = backfill_from_txt(txt_file)
|
||
print(f" {'✓' if success else '✗'} {os.path.basename(txt_file)}")
|
||
sys.exit(0 if success else 1)
|
||
|
||
# Normal mode: generate from Notion job
|
||
if len(sys.argv) < 2:
|
||
print("Usage: python3 regenerate_cover_letter.py <notion_job_id> [feedback]")
|
||
print(" or: python3 regenerate_cover_letter.py <path/to/cover-letter.txt>")
|
||
sys.exit(1)
|
||
|
||
job_id = sys.argv[1]
|
||
feedback = sys.argv[2] if len(sys.argv) > 2 else ""
|
||
main(job_id, feedback)
|