generated from kgod/ai-review-template
229 lines
9.3 KiB
Python
229 lines
9.3 KiB
Python
"""Resume document v3 identity, compatibility, lookup, and fingerprint helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from .skill_classifier import classify_skills
|
|
|
|
SCHEMA_VERSION = 3
|
|
META_KEYS = {
|
|
"pending_proposal",
|
|
"previous_version",
|
|
"gap_report",
|
|
# OfferPai row/paragraph identifiers are synchronization metadata, not
|
|
# user-visible resume content and must not make proposals stale by
|
|
# themselves.
|
|
"offerpai_record_id",
|
|
"offerpai_description_ids",
|
|
}
|
|
ITEM_KEY_FIELDS: dict[str, tuple[str, ...]] = {
|
|
"education": ("school", "start_date"),
|
|
"work_experience": ("company", "position", "start_date"),
|
|
"internship_experience": ("company", "position", "start_date"),
|
|
"project_experience": ("project_name", "start_date"),
|
|
"campus_experience": ("organization", "role", "start_date"),
|
|
"competition": ("name", "award", "date"),
|
|
"additional_experience": ("title",),
|
|
"skills": ("value",),
|
|
"certificates": ("value",),
|
|
}
|
|
|
|
|
|
class DocumentError(Exception):
|
|
def __init__(self, code: str, message: str) -> None:
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
|
|
|
|
def new_id(prefix: str) -> str:
|
|
return f"{prefix}_{uuid4().hex[:12]}"
|
|
|
|
|
|
def bullet_text(bullet: Any) -> str:
|
|
return str(bullet.get("text", "")) if isinstance(bullet, dict) else str(bullet)
|
|
|
|
|
|
def _item_key(kind: str, item: dict[str, Any], index: int) -> str:
|
|
parts = [str(item.get(field) or "") for field in ITEM_KEY_FIELDS.get(kind, ())]
|
|
key = "|".join(parts).strip("|")
|
|
return f"{kind}:{key}" if key else f"{kind}:idx:{index}"
|
|
|
|
|
|
def normalize_document(document: dict[str, Any]) -> dict[str, Any]:
|
|
"""Return the v3 resume shape while preserving legacy content."""
|
|
result = deepcopy(document)
|
|
result["schema_version"] = SCHEMA_VERSION
|
|
result["basics"] = result.get("basics") if isinstance(result.get("basics"), dict) else {}
|
|
result["target"] = result.get("target") if isinstance(result.get("target"), dict) else {}
|
|
sections = result.get("sections") if isinstance(result.get("sections"), list) else []
|
|
skill_groups = result.get("skill_groups") if isinstance(result.get("skill_groups"), list) else []
|
|
legacy_skill_sections = [section for section in sections if section.get("kind") == "skills"]
|
|
if legacy_skill_sections and not skill_groups:
|
|
values = [
|
|
str(item.get("value")).strip()
|
|
for section in legacy_skill_sections
|
|
for item in section.get("items", [])
|
|
if isinstance(item, dict) and str(item.get("value") or "").strip()
|
|
]
|
|
if values:
|
|
skill_groups = classify_skills(values)
|
|
result["sections"] = [section for section in sections if section.get("kind") != "skills"]
|
|
result["skill_groups"] = skill_groups
|
|
return result
|
|
|
|
|
|
def merge_ids(old: dict[str, Any] | None, new: dict[str, Any]) -> dict[str, Any]:
|
|
result = normalize_document(new)
|
|
old_normalized = normalize_document(old or {})
|
|
old_sections = {section.get("kind"): section for section in old_normalized.get("sections", [])}
|
|
for section in result.get("sections", []):
|
|
kind = str(section.get("kind"))
|
|
old_section = old_sections.get(kind) or {}
|
|
section["id"] = old_section.get("id") or new_id("sec")
|
|
pool: dict[str, list[dict[str, Any]]] = {}
|
|
for index, item in enumerate(old_section.get("items", [])):
|
|
pool.setdefault(_item_key(kind, item, index), []).append(item)
|
|
for index, item in enumerate(section.get("items", [])):
|
|
candidates = pool.get(_item_key(kind, item, index)) or []
|
|
old_item = candidates.pop(0) if candidates else None
|
|
item["id"] = (old_item or {}).get("id") or new_id("entry")
|
|
item["provenance"] = (
|
|
(old_item or {}).get("provenance")
|
|
or item.get("provenance")
|
|
or "user_provided"
|
|
)
|
|
_merge_bullets(old_item or {}, item)
|
|
for meta in META_KEYS:
|
|
if old_item and meta in old_item:
|
|
item[meta] = deepcopy(old_item[meta])
|
|
return result
|
|
|
|
|
|
|
|
def merge_profile_refresh(old: dict[str, Any], regenerated: dict[str, Any]) -> dict[str, Any]:
|
|
"""Merge a profile refresh without discarding imported or confirmed content."""
|
|
old_normalized = normalize_document(old)
|
|
refreshed = merge_ids(old_normalized, regenerated)
|
|
old_sections = {
|
|
str(section.get("kind")): section
|
|
for section in old_normalized.get("sections", [])
|
|
if isinstance(section, dict)
|
|
}
|
|
refreshed_by_kind = {
|
|
str(section.get("kind")): section
|
|
for section in refreshed.get("sections", [])
|
|
if isinstance(section, dict)
|
|
}
|
|
|
|
# Retain a section when profile collection has no representation for it.
|
|
for kind, old_section in old_sections.items():
|
|
if kind not in refreshed_by_kind:
|
|
refreshed["sections"].append(deepcopy(old_section))
|
|
refreshed_by_kind[kind] = refreshed["sections"][-1]
|
|
|
|
for kind, section in refreshed_by_kind.items():
|
|
old_section = old_sections.get(kind)
|
|
if old_section is None:
|
|
continue
|
|
pool: dict[str, list[dict[str, Any]]] = {}
|
|
for index, old_item in enumerate(old_section.get("items", [])):
|
|
if isinstance(old_item, dict):
|
|
pool.setdefault(_item_key(kind, old_item, index), []).append(old_item)
|
|
new_items: list[dict[str, Any]] = []
|
|
for index, item in enumerate(section.get("items", [])):
|
|
candidates = pool.get(_item_key(kind, item, index)) or []
|
|
old_item = candidates.pop(0) if candidates else None
|
|
if old_item is None:
|
|
new_items.append(item)
|
|
continue
|
|
# Preview-side confirmed wording and edits remain authoritative.
|
|
preserved = deepcopy(old_item)
|
|
preserved["id"] = item.get("id") or old_item.get("id") or new_id("entry")
|
|
new_items.append(preserved)
|
|
# Imported and manually edited entries that were not regenerated must
|
|
# precede newly collected records instead of disappearing.
|
|
preserved_unmatched = [
|
|
item for candidates in pool.values() for item in candidates
|
|
]
|
|
section["items"] = [*preserved_unmatched, *new_items]
|
|
|
|
if isinstance(old_normalized.get("profile_summary"), dict):
|
|
refreshed["profile_summary"] = deepcopy(old_normalized["profile_summary"])
|
|
refreshed["profile_summary"]["stale"] = True
|
|
return refreshed
|
|
|
|
def _merge_bullets(old_item: dict[str, Any], item: dict[str, Any]) -> None:
|
|
bullets = item.get("resume_bullets")
|
|
if not bullets:
|
|
return
|
|
pool: dict[str, list[dict[str, Any]]] = {}
|
|
for old_bullet in old_item.get("resume_bullets") or []:
|
|
pool.setdefault(bullet_text(old_bullet), []).append(old_bullet)
|
|
normalized = []
|
|
for bullet in bullets:
|
|
text = bullet_text(bullet)
|
|
candidates = pool.get(text) or []
|
|
old_bullet = candidates.pop(0) if candidates else None
|
|
normalized.append({"id": (old_bullet or {}).get("id") or new_id("b"), "text": text})
|
|
item["resume_bullets"] = normalized
|
|
|
|
|
|
def find_section(content: dict[str, Any], section_id: str) -> dict[str, Any] | None:
|
|
return next((section for section in content.get("sections", []) if section.get("id") == section_id), None)
|
|
|
|
|
|
def find_entry(
|
|
content: dict[str, Any], entry_id: str
|
|
) -> tuple[dict[str, Any], dict[str, Any]] | None:
|
|
for section in content.get("sections", []):
|
|
for item in section.get("items", []):
|
|
if item.get("id") == entry_id:
|
|
return section, item
|
|
return None
|
|
|
|
|
|
def find_bullet(entry: dict[str, Any], bullet_id: str) -> dict[str, Any] | None:
|
|
return next(
|
|
(bullet for bullet in entry.get("resume_bullets") or [] if bullet.get("id") == bullet_id),
|
|
None,
|
|
)
|
|
|
|
|
|
def entry_fingerprint(entry: dict[str, Any]) -> str:
|
|
material = {key: value for key, value in entry.items() if key not in META_KEYS and key != "id"}
|
|
if "resume_bullets" in material:
|
|
material["resume_bullets"] = [bullet_text(bullet) for bullet in material["resume_bullets"]]
|
|
blob = json.dumps(material, ensure_ascii=False, sort_keys=True)
|
|
return hashlib.sha1(blob.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def require_entry(content: dict[str, Any], entry_id: str) -> dict[str, Any]:
|
|
found = find_entry(content, entry_id)
|
|
if found is None:
|
|
raise DocumentError("entry_not_found", "Entry not found in resume")
|
|
return found[1]
|
|
|
|
|
|
def gap_report_is_stale(entry: dict[str, Any]) -> bool:
|
|
"""Return whether a persisted gap report predates the entry content."""
|
|
report = entry.get("gap_report")
|
|
if not isinstance(report, dict) or not report.get("based_on"):
|
|
return False
|
|
return report["based_on"] != entry_fingerprint(entry)
|
|
|
|
def attach_gap_report_staleness(content: dict[str, Any]) -> dict[str, Any]:
|
|
"""Return an outbound-only content copy with a derived gap-report stale marker."""
|
|
result = deepcopy(content)
|
|
for section in result.get("sections", []):
|
|
for item in section.get("items", []):
|
|
report = item.get("gap_report")
|
|
if isinstance(report, dict):
|
|
report["stale"] = gap_report_is_stale(item)
|
|
return result
|