resume-agent/backend/app/resume_document_mutations.py

"""Validated resume document edits and proposal lifecycle operations."""

from __future__ import annotations

import re
from copy import deepcopy
from datetime import UTC, datetime
from typing import Any

from .profile_summary import generated_summary
from .skill_classifier import classify_skills
from .validators import mask_phone
from .resume_document_core import (
    DocumentError,
    entry_fingerprint,
    find_bullet,
    find_entry,
    require_entry,
)

WRITABLE_ENTRY_FIELDS = {
    "school", "major", "degree", "company", "position", "project_name",
    "project_role", "start_date", "end_date_or_present", "description", "name",
    "title", "organization", "role", "award", "date", "value",
}
WRITABLE_BASICS_FIELDS = {"name", "phone", "email", "city", "portfolio_url"}
_MONTH = re.compile(r"^(?:19|20)\d{2}-(?:0[1-9]|1[0-2])$")
_PHONE = re.compile(r"^1[3-9]\d{9}$")


def _validate_entry_fields(fields: dict[str, Any]) -> None:
    for key, value in fields.items():
        if key not in WRITABLE_ENTRY_FIELDS:
            raise DocumentError("field_not_writable", f"Field '{key}' is not writable")
        if key == "start_date" and value is not None and not _MONTH.fullmatch(str(value)):
            raise DocumentError("invalid_field", "start_date must use YYYY-MM")
        if (
            key == "end_date_or_present"
            and value is not None
            and value != "present"
            and not _MONTH.fullmatch(str(value))
        ):
            raise DocumentError("invalid_field", "end_date_or_present must use YYYY-MM or present")


def _validate_basics_fields(fields: dict[str, Any]) -> None:
    for key in fields:
        if key not in WRITABLE_BASICS_FIELDS:
            raise DocumentError("field_not_writable", f"Basics field '{key}' is not writable")
    name = fields.get("name")
    if name is not None and not (0 < len(str(name).strip()) <= 64):
        raise DocumentError("invalid_field", "name must contain 1 to 64 characters")
    phone = fields.get("phone")
    if phone is not None and not _PHONE.fullmatch(str(phone)):
        raise DocumentError("invalid_field", "phone must be a valid mainland China mobile number")


def apply_update_basics(content: dict[str, Any], fields: dict[str, Any]) -> dict[str, Any]:
    _validate_basics_fields(fields)
    result = deepcopy(content)
    basics = result.setdefault("basics", {})
    for key, value in fields.items():
        if key == "phone":
            basics.pop("phone", None)
            basics["masked_phone"] = mask_phone(str(value))
            continue
        basics[key] = value.strip() if isinstance(value, str) else value
    return mark_profile_summary_stale(result)


def apply_update_skill_groups(content: dict[str, Any], skills: list[Any]) -> dict[str, Any]:
    result = deepcopy(content)
    clean: list[str] = []
    seen: set[str] = set()
    for value in skills:
        skill = str(value or "").strip()
        key = skill.casefold()
        if not skill or len(skill) > 48 or key in seen:
            continue
        seen.add(key)
        clean.append(skill)
    result["skill_groups"] = classify_skills(clean)
    return mark_profile_summary_stale(result)

def apply_update_entry(content: dict[str, Any], entry_id: str, fields: dict[str, Any]) -> dict[str, Any]:
    _validate_entry_fields(fields)
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    for key, value in fields.items():
        if value is None:
            entry.pop(key, None)
        else:
            entry[key] = value.strip() if isinstance(value, str) else value
    entry["provenance"] = "user_edited"
    return mark_profile_summary_stale(result)


def apply_update_bullet(
    content: dict[str, Any], entry_id: str, bullet_id: str, text: str
) -> dict[str, Any]:
    clean = text.strip()
    if not (0 < len(clean) <= 200):
        raise DocumentError("invalid_field", "bullet must contain 1 to 200 characters")
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    bullet = find_bullet(entry, bullet_id)
    if bullet is None:
        raise DocumentError("bullet_not_found", "Bullet not found in entry")
    bullet["text"] = clean
    entry["provenance"] = "user_edited"
    return mark_profile_summary_stale(result)


def apply_delete_entry(content: dict[str, Any], entry_id: str) -> dict[str, Any]:
    result = deepcopy(content)
    found = find_entry(result, entry_id)
    if found is None:
        raise DocumentError("entry_not_found", "Entry not found in resume")
    section, _ = found
    section["items"] = [item for item in section["items"] if item.get("id") != entry_id]
    if not section["items"]:
        result["sections"] = [item for item in result["sections"] if item.get("id") != section.get("id")]
    return mark_profile_summary_stale(result)


def apply_delete_bullet(content: dict[str, Any], entry_id: str, bullet_id: str) -> dict[str, Any]:
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    if find_bullet(entry, bullet_id) is None:
        raise DocumentError("bullet_not_found", "Bullet not found in entry")
    entry["resume_bullets"] = [
        bullet for bullet in entry.get("resume_bullets") or [] if bullet.get("id") != bullet_id
    ]
    entry["provenance"] = "user_edited"
    return mark_profile_summary_stale(result)


def set_pending_proposal(
    content: dict[str, Any],
    entry_id: str,
    optimized_description: str,
    *,
    source: str,
    changes: list[str] | None = None,
    generation_source: str | None = None,
    fallback_reason: str | None = None,
    missing_facts: list[str] | None = None,
    unconfirmed_suggestions: list[str] | None = None,
    optional_enhancements: list[str] | None = None,
    validation_warnings: list[str] | None = None,
    star: dict[str, Any] | None = None,
    omitted_fact_ids: list[str] | None = None,
) -> dict[str, Any]:
    clean = str(optimized_description).strip()
    if not clean:
        raise DocumentError("nothing_to_expand", "Expander produced no optimized description")
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    proposal = {
        "optimized_description": clean,
        "changes": [str(item).strip() for item in changes or [] if str(item).strip()][:5],
        "source": source,
        "based_on": entry_fingerprint(entry),
        "created_at": datetime.now(UTC).isoformat(),
    }
    if generation_source:
        proposal["generation_source"] = generation_source
    if fallback_reason:
        proposal["fallback_reason"] = fallback_reason
    for key, values in (
        ("missing_facts", missing_facts),
        ("unconfirmed_suggestions", unconfirmed_suggestions),
        ("optional_enhancements", optional_enhancements),
        ("validation_warnings", validation_warnings),
    ):
        cleaned = [str(item).strip() for item in values or [] if str(item).strip()]
        if cleaned:
            proposal[key] = list(dict.fromkeys(cleaned))[:8]
    if isinstance(star, dict) and star:
        proposal["star"] = deepcopy(star)
    if omitted_fact_ids:
        proposal["omitted_fact_ids"] = [
            str(item).strip() for item in omitted_fact_ids if str(item).strip()
        ][:24]

    entry["pending_proposal"] = proposal
    return result


def confirm_proposal(content: dict[str, Any], entry_id: str) -> dict[str, Any]:
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    proposal = entry.get("pending_proposal")
    if not isinstance(proposal, dict):
        raise DocumentError("optimize_not_pending", "No pending proposal for entry")

    if proposal.get("based_on") != entry_fingerprint(entry):
        raise DocumentError("proposal_stale", "Entry changed after proposal was created")
    entry["previous_version"] = {
        "description": entry.get("description"),
        "provenance": entry.get("provenance", "user_provided"),
    }
    entry["description"] = str(proposal["optimized_description"]).strip()
    entry["provenance"] = proposal.get("source", "ai_expanded")
    entry.pop("pending_proposal", None)
    return mark_profile_summary_stale(result)


def reject_proposal(content: dict[str, Any], entry_id: str) -> dict[str, Any]:
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    if "pending_proposal" not in entry:
        raise DocumentError("optimize_not_pending", "No pending proposal for entry")
    entry.pop("pending_proposal", None)
    return result


def undo_entry(content: dict[str, Any], entry_id: str) -> dict[str, Any]:
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    previous = entry.get("previous_version")
    if not isinstance(previous, dict):
        raise DocumentError("nothing_to_undo", "No previous version stored for entry")
    if previous.get("description") is None:
        entry.pop("description", None)
    else:
        entry["description"] = previous["description"]
    entry["provenance"] = previous.get("provenance", "user_edited")
    entry.pop("previous_version", None)
    return mark_profile_summary_stale(result)


def _summary(content: dict[str, Any]) -> dict[str, Any] | None:
    value = content.get("profile_summary")
    return value if isinstance(value, dict) else None


def mark_profile_summary_stale(content: dict[str, Any]) -> dict[str, Any]:
    result = deepcopy(content)
    summary = _summary(result)
    if summary and str(summary.get("content") or "").strip():
        summary["stale"] = True
    return result


def set_generated_profile_summary(
    content: dict[str, Any], summary_text: str, *, replace_stale: bool = False
) -> dict[str, Any]:
    result = deepcopy(content)
    summary = _summary(result)
    if summary and not (replace_stale and summary.get("stale") is True):
        return result
    result["profile_summary"] = generated_summary(summary_text)
    return result



def set_profile_summary_proposal(content: dict[str, Any], summary_text: str) -> dict[str, Any]:
    result = deepcopy(content)
    summary = _summary(result)
    if summary is None:
        summary = {"content": "", "source": "ai_generated", "generated_at": None, "stale": False}
        result["profile_summary"] = summary
    proposal = generated_summary(summary_text)
    summary["pending_proposal"] = {
        "content": proposal["content"],
        "source": "ai_generated",
        "generated_at": proposal["generated_at"],
    }
    return result


def confirm_profile_summary_proposal(content: dict[str, Any]) -> dict[str, Any]:
    result = deepcopy(content)
    summary = _summary(result)
    proposal = summary.get("pending_proposal") if summary else None
    if not isinstance(proposal, dict) or not str(proposal.get("content") or "").strip():
        raise DocumentError("profile_summary_not_pending", "No pending profile summary proposal")
    summary.update(generated_summary(str(proposal["content"])))
    summary.pop("pending_proposal", None)
    return result


def reject_profile_summary_proposal(content: dict[str, Any]) -> dict[str, Any]:
    result = deepcopy(content)
    summary = _summary(result)
    if not summary or "pending_proposal" not in summary:
        raise DocumentError("profile_summary_not_pending", "No pending profile summary proposal")
    summary.pop("pending_proposal", None)
    return result


def apply_update_profile_summary(content: dict[str, Any], summary_text: str) -> dict[str, Any]:
    result = deepcopy(content)
    proposal = generated_summary(summary_text)
    result["profile_summary"] = {
        "content": proposal["content"],
        "source": "user_edited",
        "generated_at": proposal["generated_at"],
        "stale": False,
    }
    return result


def set_entry_gap_report(
    content: dict[str, Any], entry_id: str, gaps: list[dict[str, Any]]
) -> dict[str, Any]:
    """Persist the latest gap analysis so the conversion panel survives other run changes."""
    result = deepcopy(content)
    entry = require_entry(result, entry_id)
    entry["gap_report"] = {
        "gaps": [dict(gap) for gap in gaps],
        "based_on": entry_fingerprint(entry),
    }
    return result