matchmaking-v2/research/poc/browser-apply/parse_resume.py

"""
parse_resume.py — turn a résumé PDF into a structured profile using a VISION model.

This reuses the proven pattern from the resume-builder service
(growqr-demo/resume-builder/app/services/parse_service.py): render each PDF page to a PNG,
send the images to gpt-4o-mini with JSON mode, merge the pages. Vision (not text extraction)
because it preserves layout — columns, headers, contact blocks — that plain text destroys.

The output `profile` dict is the SINGLE SOURCE OF TRUTH the apply agent uses to fill the form,
so identity and experience come from the real résumé, never from guesses.

Standalone usage:
    python parse_resume.py                 # parses ./Resume.pdf and prints the profile
Importable:
    from parse_resume import parse_resume_to_profile
    profile = await parse_resume_to_profile("Resume.pdf")
"""
from __future__ import annotations

import asyncio
import base64
import json
import os
from datetime import date
from typing import Any

import fitz  # PyMuPDF
from dotenv import load_dotenv
from openai import AsyncOpenAI

load_dotenv()

VISION_MODEL = "gpt-4o-mini"

SYSTEM_PROMPT = """You are a precise résumé parser. You are given an image of ONE page of a
candidate's résumé. Extract ONLY what is actually present — never invent or infer values that
are not visibly written. Return a JSON object with exactly these keys:

{
  "full_name": string | null,
  "email": string | null,
  "phone": string | null,
  "location": string | null,          // city/region as written
  "linkedin_url": string | null,
  "github_url": string | null,
  "portfolio_url": string | null,
  "current_title": string | null,     // most recent / current role title
  "skills": [string],                  // technical skills, as written
  "experience": [                       // every role visible on THIS page
    {"company": string, "title": string, "start_date": "YYYY-MM" | null,
     "end_date": "YYYY-MM" | null, "is_current": boolean}
  ],
  "education": [
    {"institution": string, "degree": string, "field_of_study": string | null,
     "end_date": "YYYY-MM" | null}
  ]
}

If a field is not on this page, use null (or [] for lists). Output ONLY the JSON object."""


def _pdf_to_png_b64(pdf_path: str, dpi: int = 200) -> list[str]:
    """Render each PDF page to a base64-encoded PNG (same approach as resume-builder)."""
    doc = fitz.open(pdf_path)
    images: list[str] = []
    for page in doc:
        pix = page.get_pixmap(dpi=dpi)
        images.append(base64.b64encode(pix.tobytes("png")).decode("utf-8"))
    doc.close()
    return images


async def _parse_page(client: AsyncOpenAI, img_b64: str, page_num: int, total: int) -> dict[str, Any]:
    resp = await client.chat.completions.create(
        model=VISION_MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": f"Résumé page {page_num} of {total}."},
                    {"type": "image_url",
                     "image_url": {"url": f"data:image/png;base64,{img_b64}", "detail": "high"}},
                ],
            },
        ],
        temperature=0.1,
        max_tokens=4000,
        response_format={"type": "json_object"},
    )
    try:
        return json.loads(resp.choices[0].message.content or "{}")
    except json.JSONDecodeError:
        return {}


def _first(*vals):
    for v in vals:
        if v:
            return v
    return None


def _years_experience(experience: list[dict]) -> int | None:
    """Rough total experience: earliest start date -> today (or latest end)."""
    starts = [e.get("start_date") for e in experience if e.get("start_date")]
    if not starts:
        return None
    earliest = min(starts)  # "YYYY-MM" strings sort chronologically
    try:
        y, m = (int(x) for x in earliest.split("-")[:2])
    except (ValueError, IndexError):
        return None
    today = date.today()
    months = (today.year - y) * 12 + (today.month - m)
    return max(0, round(months / 12))


def _merge(pages: list[dict[str, Any]]) -> dict[str, Any]:
    """Merge per-page results into one profile; first non-empty wins for scalars."""
    skills: list[str] = []
    experience: list[dict] = []
    education: list[dict] = []
    for p in pages:
        for s in p.get("skills") or []:
            if s and s not in skills:
                skills.append(s)
        experience.extend(p.get("experience") or [])
        education.extend(p.get("education") or [])

    profile = {
        "full_name": _first(*[p.get("full_name") for p in pages]),
        "email": _first(*[p.get("email") for p in pages]),
        "phone": _first(*[p.get("phone") for p in pages]),
        "location": _first(*[p.get("location") for p in pages]),
        "linkedin_url": _first(*[p.get("linkedin_url") for p in pages]),
        "github_url": _first(*[p.get("github_url") for p in pages]),
        "portfolio_url": _first(*[p.get("portfolio_url") for p in pages]),
        "current_title": _first(*[p.get("current_title") for p in pages]),
        "skills": skills,
        "experience": experience,
        "education": education,
    }
    profile["years_experience"] = _years_experience(experience)
    return profile


async def parse_resume_to_profile(pdf_path: str = "Resume.pdf") -> dict[str, Any]:
    if not os.path.exists(pdf_path):
        raise FileNotFoundError(f"résumé not found: {pdf_path}")
    if not os.environ.get("OPENAI_API_KEY"):
        raise RuntimeError("OPENAI_API_KEY not set (put it in .env)")

    images = _pdf_to_png_b64(pdf_path)
    client = AsyncOpenAI()
    pages = await asyncio.gather(
        *[_parse_page(client, img, i + 1, len(images)) for i, img in enumerate(images)]
    )
    return _merge(list(pages))


if __name__ == "__main__":
    profile = asyncio.run(parse_resume_to_profile())
    print(json.dumps(profile, indent=2, ensure_ascii=False))