#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re from collections import Counter, defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parent.parent RESUME_DIR = ROOT / "resumes" OUT_PATH = ROOT / "resume_library_db.json" VAULT_PATH = ROOT / "resume_vault.json" EXCLUDED_FILENAMES = { "darsh_shah_resume_template.md", "resume_completion_log.md", } THEME_KEYWORDS = { "leadership": ["led", "managed", "manager", "cross-functional", "stakeholder", "team"], "product": ["product", "roadmap", "prd", "user", "launch", "feature", "retention"], "finance": ["revenue", "arr", "ebitda", "forecast", "fundraise", "investment", "p&l"], "analytics": ["analysis", "dashboard", "a/b", "experiment", "data", "metric", "insight"], "technical": ["python", "sql", "ml", "engineering", "api", "automation", "platform"], "operations": ["process", "sla", "support", "roll-out", "adoption", "training"], "growth": ["gtm", "growth", "acquisition", "conversion", "sign-ups", "retention"], } WEAK_VERBS = { "assisted", "collaborated", "contributed", "helped", "participated", "supported", "worked", } TRY_HARD_TERMS = { "advanced": "Sounds inflated unless the technical level is essential.", "c-suite": "Prestige signal; use sparingly and prefer the decision or artifact.", "comprehensive": "Often filler; replace with the concrete scope.", "cutting-edge": "Buzzword; replace with the actual technology or outcome.", "dynamic": "Generic resume language.", "enablement": "Can be useful, but often sounds corporate; make the user/outcome concrete.", "exceptional": "Unsubstantiated praise.", "leadership-ready": "Sounds dressed up; name the audience or decision plainly.", "orchestrating": "Can sound theatrical; use 'led' or name the implementation work.", "proven track record": "Generic resume language.", "spearheaded": "Often try-hard; use only when leadership is the point.", "strategic": "Often overused; make the strategy/decision specific.", "transformation": "Grand noun; use only for truly large-scale change.", "world-class": "Unsubstantiated praise.", } CLUSTER_RULES = [ ( "epic-ai-credits-pricing", ["epic", "ai", "credits"], ["unit economics", "api costs", "cache hit", "gross margin", "pricing"], ), ( "epic-contract-to-cash-revenue-leakage", ["epic"], ["contract-to-cash", "royalty leakage", "revenue leakage", "workday", "underpaid", "release-term"], ), ( "epic-market-intelligence-tam", ["epic"], ["38,000", "market opportunity", "tam", "china", "mobile", "market share"], ), ( "epic-royalty-guardrails", ["epic"], ["royalty guardrails", "approval guardrails", "exemption", "minimum guarantee", "royalty rates"], ), ( "pass-pricing-tiers", ["pass+"], ["pricing tiers", "three b2b saas pricing", "price increase", "ebitda"], ), ( "pass-reverse-free-trial", ["pass+"], ["reverse free trial", "reverse-free-trial", "free-trial", "paid conversion"], ), ( "sharechat-salesforce-workflow", ["sharechat"], ["salesforce", "approval rules", "source of truth", "google sheets", "crm"], ), ( "sharechat-series-g-revenue-planning", ["sharechat"], ["series g", "$255m", "investor diligence", "fundraise"], ), ( "sharechat-revenue-forecasting", ["sharechat"], ["revenue forecast", "weekly sales", "business reviews", "within 5%", "forecasting dashboard"], ), ( "sharechat-live-audio-unit-economics", ["sharechat"], ["live audio", "virtual gifting", "creator payouts", "gmv", "social audio"], ), ( "rainshine-acquisition-integration-planning", ["rainshine"], ["acquisition", "post-close", "integration targets", "content ip", "$75m"], ), ( "rainshine-shared-services-cost-savings", ["rainshine"], ["shared-services", "shared services", "cost reductions", "cost savings", "$20m-$30m"], ), ( "epic-ai-enablement", ["epic"], ["claude code", "schema-aware", "streamlit", "github actions", "docker"], ), ( "cadence-transcription-app", ["cadence"], ["whisper", "transcription", "spoken notes", "voice-to-text"], ), ] def section_of(line: str, current: str | None) -> str | None: l = line.strip().lower().lstrip("#").strip() if not l: return current if "education" == l or l.startswith("education "): return "education" if l == "experience" or l.startswith("experience ") or l == "professional experience": return "experience" if l.startswith("additional") or "technical skills" in l: return "additional" return current def metrics_in(text: str) -> list[str]: patterns = [ r"\$[\d,.]+(?:[MBK]|M|B|K)?", r"\b\d+(?:\.\d+)?%", r"\b\d+(?:\.\d+)?x\b", r"\b\d+(?:\.\d+)?[MBK]\b", r"\b\d[\d,]*(?:\+\b)?", ] hits: list[str] = [] for pat in patterns: hits.extend(re.findall(pat, text, flags=re.IGNORECASE)) seen = [] for h in hits: if h not in seen: seen.append(h) return seen def themes_in(text: str) -> list[str]: l = text.lower() themes: list[str] = [] for theme, kws in THEME_KEYWORDS.items(): if any(kw in l for kw in kws): themes.append(theme) return themes def keywords_in(text: str) -> list[str]: tokens = re.findall(r"[A-Za-z][A-Za-z0-9+/.-]*", text) stop = { "the", "and", "for", "with", "from", "into", "through", "across", "among", "within", "this", "that", "was", "were", "are", "is", "to", "of", "in", "on", "by", "or", "a", "an", "as", "at", "it", "its", "their", "my", "our", "your", "using", "use", "led", "built", "created", "developed", "improved", "launched", "managed", } freq = Counter(t for t in tokens if len(t) > 2 and t.lower() not in stop) return [t for t, _ in freq.most_common(8)] def slugify(text: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") return slug[:80] or "uncategorized" def context_text(context_window: list[str]) -> str: return " ".join(context_window).lower() def cluster_id_for(text: str, context_window: list[str]) -> str: haystack = f"{context_text(context_window)} {text.lower()}" for cluster_id, required, any_terms in CLUSTER_RULES: if all(term in haystack for term in required) and any(term in haystack for term in any_terms): return cluster_id company = "general" if "epic" in haystack: company = "epic" elif "pass+" in haystack: company = "pass" elif "sharechat" in haystack: company = "sharechat" elif "rainshine" in haystack: company = "rainshine" elif "kellogg" in haystack: company = "kellogg" kws = keywords_in(text)[:4] return f"{company}-{slugify('-'.join(kws))}" def opening_verb(text: str) -> str: match = re.match(r"^([A-Za-z][A-Za-z-]*)", text.strip()) return match.group(1) if match else "" def word_count(text: str) -> int: return len(re.findall(r"\S+", text)) def starts_with_metric(text: str) -> bool: return bool(re.match(r"^\s*(?:\$?\d|~?\$)", text)) def bullet_shape(text: str) -> str: l = text.lower() if any(term in l for term in ["claude", "streamlit", "whisper", "github actions", "docker"]): return "ai_enablement" if any(term in l for term in ["workflow", "pipeline", "dashboard", "data model", "salesforce", "automation", "source of truth"]): return "system_or_workflow" if any(term in l for term in ["pricing", "margin", "revenue", "ebitda", "gross margin", "royalty", "deal"]): return "commercial_or_financial_outcome" if any(term in l for term in ["strategy", "planning", "board", "executive", "market opportunity", "tam"]): return "strategy_or_decision_support" if any(term in l for term in ["led", "trained", "adoption", "rollout", "implementation"]): return "leadership_or_rollout" return "general_achievement" def critique_for(text: str) -> dict: l = text.lower() flags = [] hints = [] words = word_count(text) verb = opening_verb(text) if words > 35: flags.append("long_over_35_words") hints.append("Trim to 25-35 words or split into one achievement and one supporting detail.") if words < 12: flags.append("possibly_too_thin") hints.append("Add scale, audience, or outcome if this bullet needs to earn resume space.") if verb.lower() in WEAK_VERBS: flags.append("weak_opening_verb") hints.append("Replace the opening verb with stronger ownership language if truthful.") if starts_with_metric(text): flags.append("starts_with_metric") hints.append("Lead with the metric only if the number is the strongest differentiator.") matched_try_hard = [term for term in TRY_HARD_TERMS if term in l] if matched_try_hard: flags.append("try_hard_language") hints.extend(TRY_HARD_TERMS[term] for term in matched_try_hard) if sum(1 for term in ["built", "created", "developed", "implemented"] if l.startswith(term)) and any( term in l for term in ["workflow", "pipeline", "dashboard", "data model", "system", "automation"] ): flags.append("implementation_heavy") hints.append("Check nearby bullets; if several are system-building bullets, reframe one around the decision or business result.") if len(re.findall(r"\b(?:and|,)\b", text)) > 7: flags.append("possibly_overstuffed") hints.append("Choose the 1-2 details that make this bullet distinct; avoid preserving every true input.") return { "flags": sorted(set(flags)), "hints": list(dict.fromkeys(hints)), } def quality_score(bullet: dict) -> int: critique = bullet.get("critique", {}) flags = critique.get("flags", []) words = bullet.get("word_count", word_count(bullet["text"])) score = 0 score += min(len(bullet.get("metrics", [])), 4) * 3 score += min(len(bullet.get("themes", [])), 4) if 20 <= words <= 35: score += 4 elif words > 45: score -= 4 score -= len(flags) * 2 if bullet.get("source_resume", "").startswith("anthropic_analytics/"): score += 2 return score def load_approval_lookup() -> dict[str, dict[str, str]]: if not VAULT_PATH.exists(): return {} data = json.loads(VAULT_PATH.read_text(encoding="utf-8")) lookup: dict[str, dict[str, str]] = {} for achievement in data.get("achievements", []): achievement_id = str(achievement.get("id", "")) canonical = str(achievement.get("canonical_text", "")).strip() if achievement.get("status") == "approved" and canonical: lookup[canonical] = { "approval_status": "approved", "achievement_id": achievement_id, } for history in achievement.get("history", []): text = str(history.get("text", "")).strip() if text: lookup[text] = { "approval_status": "superseded", "achievement_id": achievement_id, } return lookup def build_bullet( bullet_text: str, path: Path, context_window: list[str], *, company: str | None, role: str | None, approval_lookup: dict[str, dict[str, str]], ) -> dict: critique = critique_for(bullet_text) approval = approval_lookup.get( bullet_text, {"approval_status": "unreviewed", "achievement_id": None}, ) return { "text": bullet_text, "cluster_id": cluster_id_for(bullet_text, context_window), "word_count": word_count(bullet_text), "opening_verb": opening_verb(bullet_text), "bullet_shape": bullet_shape(bullet_text), "critique": critique, "themes": themes_in(bullet_text), "metrics": metrics_in(bullet_text), "keywords": keywords_in(bullet_text), "source_resume": path.relative_to(RESUME_DIR).as_posix(), "section": "Professional Experience", "company": company, "role": role, **approval, "context": context_window[-2:], } def build_clusters(bullets: list[dict]) -> list[dict]: grouped: dict[str, list[dict]] = defaultdict(list) for bullet in bullets: grouped[bullet["cluster_id"]].append(bullet) clusters = [] for cluster_id, variants in sorted(grouped.items()): variant_metrics = [] variant_themes = [] for variant in variants: variant_metrics.extend(variant.get("metrics", [])) variant_themes.extend(variant.get("themes", [])) flag_counts = Counter( flag for variant in variants for flag in variant.get("critique", {}).get("flags", []) ) shape_counts = Counter(variant.get("bullet_shape", "general_achievement") for variant in variants) canonical = max(variants, key=quality_score) clusters.append( { "cluster_id": cluster_id, "canonical_text": canonical["text"], "canonical_source": canonical["source_resume"], "themes": sorted(set(variant_themes)), "metrics": sorted(set(variant_metrics)), "bullet_shapes": dict(sorted(shape_counts.items())), "critique_flags": dict(sorted(flag_counts.items())), "variant_count": len(variants), "source_resumes": sorted({v["source_resume"] for v in variants}), "variants": variants, } ) return clusters def build_critique_summary(bullets: list[dict], clusters: list[dict]) -> dict: flag_counts = Counter( flag for bullet in bullets for flag in bullet.get("critique", {}).get("flags", []) ) shape_counts = Counter(bullet.get("bullet_shape", "general_achievement") for bullet in bullets) flagged_clusters = [ { "cluster_id": cluster["cluster_id"], "variant_count": cluster["variant_count"], "critique_flags": cluster["critique_flags"], "canonical_source": cluster["canonical_source"], "canonical_text": cluster["canonical_text"], } for cluster in clusters if cluster.get("critique_flags") ] flagged_clusters.sort( key=lambda c: (sum(c["critique_flags"].values()), c["variant_count"]), reverse=True, ) return { "flag_counts": dict(flag_counts.most_common()), "bullet_shape_counts": dict(shape_counts.most_common()), "top_flagged_clusters": flagged_clusters[:15], "guidance": [ "Use flagged bullets as critique candidates, not automatic defects.", "Prefer canonical variants with fewer flags and stronger target-role fit.", "When a cluster has many variants, compare variants before rewriting from scratch.", ], } def looks_like_bullet_continuation(line: str) -> bool: if not line: return False if re.match(r"^\d{4}\s*-\s*\d{4}\b", line): return False if re.match(r"^(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Sept|Oct|Nov|Dec)[A-Za-z-]*\b", line): return False if re.match(r"^[a-z0-9(]", line): return True return False def company_from_heading(line: str) -> str | None: if not line.startswith("### "): return None parts = [part.strip() for part in line[4:].split("|")] if len(parts) < 2: return None return re.sub(r"\s+\([^)]*\)\s*$", "", parts[1]).strip() or None def role_from_line(line: str) -> str | None: match = re.match(r"^\*\*(.+?)\*\*", line) return match.group(1).strip() if match else None def parse_resume(path: Path, approval_lookup: dict[str, dict[str, str]]) -> dict: text = path.read_text(encoding="utf-8") lines = [ln.rstrip() for ln in text.splitlines()] section = None bullets = [] skills = {"technical": []} education = [] context_window: list[str] = [] last_bullet: dict | None = None current_company: str | None = None current_role: str | None = None for raw in lines: line = raw.strip() section = section_of(line, section) if not line or line.startswith("_Source:") or line.startswith("# "): continue if section == "education": education.append(line) if "technical skills:" in line.lower(): _, _, rhs = line.partition(":") skills["technical"] = [s.strip() for s in rhs.split(",") if s.strip()] if section == "experience": heading_company = company_from_heading(line) if heading_company: current_company = heading_company current_role = None heading_role = role_from_line(line) if heading_role: current_role = heading_role if line.startswith("-"): bullet_text = line.lstrip("-").strip() last_bullet = build_bullet( bullet_text, path, context_window, company=current_company, role=current_role, approval_lookup=approval_lookup, ) bullets.append(last_bullet) elif last_bullet and looks_like_bullet_continuation(line): merged = f"{last_bullet['text']} {line}".strip() bullets[-1] = build_bullet( merged, path, context_window, company=current_company, role=current_role, approval_lookup=approval_lookup, ) last_bullet = bullets[-1] else: last_bullet = None context_window.append(line) context_window = context_window[-6:] return { "resume_file": path.name, "education": education[:10], "bullets": bullets, "skills": skills, } def build_database() -> dict: approval_lookup = load_approval_lookup() resumes = sorted( p for p in RESUME_DIR.rglob("*.md") if p.name not in EXCLUDED_FILENAMES and not any(part.startswith(".") for part in p.relative_to(RESUME_DIR).parts) ) parsed = [parse_resume(p, approval_lookup) for p in resumes] all_bullets = [] skills_tech = [] for entry in parsed: all_bullets.extend(entry["bullets"]) skills_tech.extend(entry["skills"].get("technical", [])) clusters = build_clusters(all_bullets) roles = sorted( { (bullet.get("company"), bullet.get("role")) for bullet in all_bullets if bullet.get("company") or bullet.get("role") }, key=lambda item: ((item[0] or "").lower(), (item[1] or "").lower()), ) return { "schema_version": 2, "corpus_scope": "all_resume_markdown_including_unapproved_drafts", "resumes_dir": str(RESUME_DIR.relative_to(ROOT)), "resume_count": len(parsed), "roles": [{"company": company, "role": role} for company, role in roles], "bullets": all_bullets, "clusters": clusters, "critique_summary": build_critique_summary(all_bullets, clusters), "skills": { "technical": sorted({s for s in skills_tech}), }, "education": [ {"source_resume": entry["resume_file"], "lines": entry["education"]} for entry in parsed ], "user_preferences": { "section_order_hint": ["education", "experience", "additional"], "bullet_style": "dash", "source_format": "markdown_extracted_from_docx_pdf", }, } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Build the broad resume search corpus from Markdown sources." ) parser.add_argument( "--check", action="store_true", help="Return nonzero when the existing output differs; do not write.", ) parser.add_argument( "--output", default=str(OUT_PATH), help="Output JSON path (default: resume_library_db.json)", ) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) output_path = Path(args.output).expanduser() if not output_path.is_absolute(): output_path = (ROOT / output_path).resolve() db = build_database() payload = json.dumps(db, indent=2) + "\n" if args.check: if not output_path.exists() or output_path.read_text(encoding="utf-8") != payload: print(f"Stale: {output_path}") return 1 print(f"Current: {output_path}") return 0 output_path.write_text(payload, encoding="utf-8") display_path = ( output_path.relative_to(ROOT) if output_path.is_relative_to(ROOT) else output_path ) print(f"Wrote {display_path}") print(f"Resumes: {db['resume_count']}") print(f"Bullets indexed: {len(db['bullets'])}") return 0 if __name__ == "__main__": raise SystemExit(main())