"""Domain models for an evidence-grounded Korean resume generation harness. The models deliberately keep source evidence, job requirements, generated claims, and quality findings as separate concepts. That separation makes unsupported claims and accidental use of sensitive personal data detectable before rendering. """ from __future__ import annotations import calendar import hashlib import json import re import unicodedata from datetime import date, datetime, timezone from enum import StrEnum from typing import Annotated, Literal, Self from pydantic import ( AwareDatetime, BaseModel, ConfigDict, Field, StringConstraints, computed_field, field_validator, model_validator, ) from .records import EmploymentType, ResumeRecords, StructuredRecord Identifier = Annotated[ str, StringConstraints( strip_whitespace=True, min_length=1, max_length=128, pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$", ), ] NonEmptyText = Annotated[ str, StringConstraints(strip_whitespace=True, min_length=1, max_length=20_000) ] ShortText = Annotated[ str, StringConstraints(strip_whitespace=True, min_length=1, max_length=300) ] def _utc_now() -> datetime: return datetime.now(timezone.utc) def _duplicates(values: list[str]) -> list[str]: """Return duplicates in stable order, comparing identifiers literally.""" seen: set[str] = set() duplicates: list[str] = [] for value in values: if value in seen and value not in duplicates: duplicates.append(value) seen.add(value) return duplicates def _normalised_duplicates(values: list[str]) -> list[str]: seen: set[str] = set() duplicates: list[str] = [] for value in values: normalised = value.casefold() if normalised in seen and normalised not in duplicates: duplicates.append(normalised) seen.add(normalised) return duplicates class DomainModel(BaseModel): """Strict base used by all externally exchanged harness data.""" model_config = ConfigDict( extra="forbid", str_strip_whitespace=True, validate_assignment=True, ) class OutputMode(StrEnum): MARKDOWN = "markdown" JSON = "json" HTML = "html" DOCX = "docx" PDF = "pdf" class ResumeMode(StrEnum): PRIVATE_MODERN = "private_modern" PUBLIC_BLIND = "public_blind" EMPLOYER_FORM = "employer_form" class EvidenceCategory(StrEnum): CAREER = "career" PROJECT = "project" EDUCATION = "education" SKILL = "skill" CERTIFICATION = "certification" AWARD = "award" PUBLICATION = "publication" LANGUAGE = "language" VOLUNTEER = "volunteer" MILITARY_SERVICE = "military_service" OTHER = "other" class EvidenceSource(StrEnum): USER_STATEMENT = "user_statement" DOCUMENT = "document" PORTFOLIO = "portfolio" CERTIFICATE = "certificate" EMPLOYMENT_RECORD = "employment_record" PUBLIC_URL = "public_url" IMPORTED_RESUME = "imported_resume" class VerificationStatus(StrEnum): UNVERIFIED = "unverified" SELF_REPORTED = "self_reported" DOCUMENT_VERIFIED = "document_verified" EXTERNALLY_VERIFIED = "externally_verified" class SensitiveDataCategory(StrEnum): PHOTO = "photo" BIRTH_DATE = "birth_date" GENDER = "gender" FULL_ADDRESS = "full_address" MARITAL_STATUS = "marital_status" FAMILY_DETAILS = "family_details" RELIGION = "religion" DISABILITY = "disability" HEALTH = "health" MILITARY_DETAILS = "military_details" COMPENSATION = "compensation" POLITICAL_OPINION = "political_opinion" PROPERTY = "property" NATIONAL_ID = "national_id" BANK_ACCOUNT = "bank_account" PROHIBITED_SENSITIVE_CATEGORIES = frozenset( { SensitiveDataCategory.NATIONAL_ID, SensitiveDataCategory.BANK_ACCOUNT, SensitiveDataCategory.HEALTH, SensitiveDataCategory.POLITICAL_OPINION, SensitiveDataCategory.PROPERTY, } ) _KOREAN_RESIDENT_ID_PATTERN = re.compile(r"(? bool: """Detect an identity token without rejecting ordinary Korean morphology.""" if not identity: return False escaped = re.escape(identity) if re.search(r"[가-힣]", identity): compact_identity = re.sub(r"\s+", "", identity) flexible_identity = r"\s*".join( re.escape(character) for character in compact_identity ) return ( re.search( rf"(? set[str]: normalised = unicodedata.normalize("NFKC", text).casefold() anchors: set[str] = set() korean_suffixes = ( "에서는", "으로는", "에게서", "께서는", "에서", "으로", "에게", "께서", "부터", "까지", "처럼", "보다", "이나", "이나마", "은", "는", "이", "가", "을", "를", "의", "에", "와", "과", "도", ) for token in _SEMANTIC_TOKEN_PATTERN.findall(normalised): variants = {token} if re.fullmatch(r"[가-힣]+", token): for suffix in korean_suffixes: if token.endswith(suffix) and len(token) - len(suffix) >= 2: variants.add(token[: -len(suffix)]) break anchors.update( variant for variant in variants if variant not in _SEMANTIC_STOPWORDS and len(variant) >= 2 ) return anchors def _high_signal_tokens(text: str) -> set[str]: """Extract exact technology/credential-like tokens and typed quantities.""" normalised = unicodedata.normalize("NFKC", text).casefold() ascii_tokens = { token.casefold() for token in _HIGH_SIGNAL_ASCII_PATTERN.findall(normalised) if token.casefold() not in _SEMANTIC_STOPWORDS } quantities = { re.sub(r"\s+", "", token) for token in _HIGH_SIGNAL_QUANTITY_PATTERN.findall(normalised) } return ascii_tokens | quantities def _source_quote_occurs(source: str, quote: str) -> bool: """Match a contiguous quote without accepting ASCII token substrings.""" normalised_source = re.sub( r"\s+", " ", unicodedata.normalize("NFKC", source) ).casefold() normalised_quote = re.sub( r"\s+", " ", unicodedata.normalize("NFKC", quote) ).strip().casefold() escaped = re.escape(normalised_quote).replace(r"\ ", r"\s+") prefix = r"(? list[tuple[int, int]]: """Locate a phrase while tolerating Korean layout whitespace differences.""" normalised_text = unicodedata.normalize("NFKC", text).casefold() compact_phrase = re.sub( r"\s+", "", unicodedata.normalize("NFKC", phrase).casefold() ) if not compact_phrase: return [] pattern = r"\s*".join(re.escape(character) for character in compact_phrase) if compact_phrase[0].isascii() and compact_phrase[0].isalnum(): pattern = r"(? int: if first[1] < second[0]: return second[0] - first[1] if second[1] < first[0]: return first[0] - second[1] return 0 def _value_is_closest_to_scope( scope_spans: list[tuple[int, int]], selected_spans: list[tuple[int, int]], alternative_spans: list[tuple[int, int]], ) -> bool: """Bind a typed value to its local subject, not another nearby subject.""" if not scope_spans or not selected_spans: return False selected_distance = min( _span_distance(scope, value) for scope in scope_spans for value in selected_spans ) if not alternative_spans: return True alternative_distance = min( _span_distance(scope, value) for scope in scope_spans for value in alternative_spans ) # A tie is ambiguous and therefore cannot support a blocking constraint. return selected_distance < alternative_distance def _classification_marker_is_local( classification_quote: str, source_quote: str, marker: re.Pattern[str], ) -> bool: """Ensure a required/preferred heading does not cross another section.""" normalised = unicodedata.normalize("NFKC", classification_quote).casefold() source_spans = _phrase_spans(normalised, source_quote) marker_spans = [match.span() for match in marker.finditer(normalised)] for source_span in source_spans: for marker_span in marker_spans: if marker_span[1] > source_span[0]: continue between = normalised[marker_span[1] : source_span[0]] if _CLASSIFICATION_SECTION_BOUNDARY_PATTERN.search(between) is None: return True return False def _posting_blocking_constraint_clauses( raw_text: str, ) -> list[tuple[str, frozenset[str]]]: """Return explicit clauses and the constraint kinds each clause requires.""" normalised = unicodedata.normalize("NFKC", raw_text) detected: list[tuple[str, frozenset[str]]] = [] clauses = re.split(r"[\n.;。]+", normalised) for clause in clauses: clause = clause.strip() if not clause: continue non_blocking = _NON_BLOCKING_CONSTRAINT_MARKER_PATTERN.search(clause) explicit_blocking = re.search( r"(?:필수|반드시|이내|이하|미만|금지|불가|로만)", clause, ) if non_blocking is not None and explicit_blocking is None: continue expected_kinds = frozenset( kind for kind, pattern in _EXPLICIT_BLOCKING_SUBMISSION_PATTERNS if pattern.search(clause) ) if expected_kinds: detected.append((clause, expected_kinds)) return detected def _has_sufficient_source_anchors(text: str, quote: str) -> bool: claimed = _semantic_anchors(text) quoted = _semantic_anchors(quote) if not claimed or not quoted: return False matched = claimed & quoted minimum = 1 if len(claimed) <= 2 else max(2, (len(claimed) + 1) // 2) return len(matched) >= minimum _EVIDENCE_SENSITIVE_PATTERNS: tuple[ tuple[SensitiveDataCategory, tuple[re.Pattern[str], ...]], ... ] = ( ( SensitiveDataCategory.PHOTO, ( re.compile(r"(?:증명|반명함|여권|프로필)\s*사진"), re.compile(r"사진\s*(?:첨부|부착|제출)"), ), ), ( SensitiveDataCategory.BIRTH_DATE, ( re.compile(r"(?:생년월일|출생일?)\s*[:：]?"), re.compile(r"(? Self: if self.day is not None and self.month is None: raise ValueError("day requires month") if self.month is not None and self.day is not None: try: date(self.year, self.month, self.day) except ValueError as exc: raise ValueError("invalid calendar date") from exc return self @property def precision(self) -> Literal["year", "month", "day"]: if self.day is not None: return "day" if self.month is not None: return "month" return "year" def earliest(self) -> date: return date(self.year, self.month or 1, self.day or 1) def latest(self) -> date: month = self.month or 12 day = self.day or calendar.monthrange(self.year, month)[1] return date(self.year, month, day) def format_ko(self) -> str: if self.day is not None: return f"{self.year}.{self.month:02d}.{self.day:02d}" if self.month is not None: return f"{self.year}.{self.month:02d}" return str(self.year) class DateRange(DomainModel): start: ResumeDate end: ResumeDate | None = None ongoing: bool = False @model_validator(mode="after") def validate_range(self) -> Self: if self.ongoing and self.end is not None: raise ValueError("ongoing date range cannot have an end date") if self.end is not None and self.end.latest() < self.start.earliest(): raise ValueError("end date must not be earlier than start date") return self class ContactInfo(DomainModel): email: str | None = Field(default=None, max_length=254) phone: str | None = Field(default=None, max_length=30) city: str | None = Field(default=None, max_length=100) links: list[str] = Field(default_factory=list, max_length=10) @field_validator("email") @classmethod def validate_email(cls, value: str | None) -> str | None: if value is None: return value if not re.fullmatch(r"[^\s@]+@[^\s@]+\.[^\s@]+", value): raise ValueError("invalid email address") return value @field_validator("phone") @classmethod def validate_phone(cls, value: str | None) -> str | None: if value is None: return value compact = re.sub(r"[\s().-]", "", value) if not re.fullmatch(r"\+?\d{8,15}", compact): raise ValueError("phone must contain 8 to 15 digits") return value @field_validator("city") @classmethod def require_coarse_region(cls, value: str | None) -> str | None: if value is None: return value compact = re.sub(r"\s+", " ", value).strip() korean_top_regions = { "서울", "서울특별시", "부산", "부산광역시", "대구", "대구광역시", "인천", "인천광역시", "광주", "광주광역시", "대전", "대전광역시", "울산", "울산광역시", "세종", "세종특별자치시", "경기", "경기도", "강원", "강원특별자치도", "충북", "충청북도", "충남", "충청남도", "전북", "전북특별자치도", "전남", "전라남도", "경북", "경상북도", "경남", "경상남도", "제주", "제주특별자치도", } if compact in korean_top_regions: return compact if ( re.search(r"\d|번지|아파트|빌딩|오피스텔|우편번호", compact) or re.search( r"(?:^|\s)[가-힣]{2,}(?:시|구|군|읍|면|동|리)(?:\s|$)", compact, ) or len(compact) > 50 ): raise ValueError("city must be a coarse city/province-level region") return compact @field_validator("links") @classmethod def validate_links(cls, values: list[str]) -> list[str]: for value in values: if not re.fullmatch(r"https?://[^\s]+", value): raise ValueError("contact links must be absolute HTTP(S) URLs") if _normalised_duplicates(values): raise ValueError("contact links must be unique") return values @model_validator(mode="after") def require_contact_channel(self) -> Self: if self.email is None and self.phone is None and not self.links: raise ValueError("at least one contact channel is required") return self class SensitiveDataConsent(DomainModel): """Explicit, purpose-bound permission for one sensitive data category.""" consent_id: Identifier category: SensitiveDataCategory purpose: ShortText granted: bool = True granted_at: AwareDatetime expires_at: AwareDatetime | None = None revoked_at: AwareDatetime | None = None @model_validator(mode="after") def validate_consent_timeline(self) -> Self: if self.expires_at is not None and self.expires_at <= self.granted_at: raise ValueError("consent expiry must be later than grant time") if self.revoked_at is not None and self.revoked_at < self.granted_at: raise ValueError("consent cannot be revoked before it is granted") return self def is_active_at(self, instant: datetime) -> bool: if instant.tzinfo is None or instant.utcoffset() is None: raise ValueError("consent checks require a timezone-aware datetime") return ( self.granted and self.granted_at <= instant and (self.expires_at is None or instant < self.expires_at) and (self.revoked_at is None or instant < self.revoked_at) ) class EvidenceItem(DomainModel): """Atomic candidate fact that may support one or more generated claims.""" evidence_id: Identifier category: EvidenceCategory content: NonEmptyText source: EvidenceSource source_reference: str | None = Field(default=None, max_length=2_000) date_range: DateRange | None = None verification_status: VerificationStatus = VerificationStatus.UNVERIFIED metrics: dict[str, str | int | float] = Field(default_factory=dict, max_length=30) keywords: list[ShortText] = Field(default_factory=list, max_length=50) sensitive_category: SensitiveDataCategory | None = None consent_id: Identifier | None = None confidential: bool = False @field_validator("content") @classmethod def reject_resident_registration_number(cls, value: str) -> str: if _KOREAN_RESIDENT_ID_PATTERN.search(value): raise ValueError("Korean resident registration numbers are prohibited") return value @field_validator("keywords") @classmethod def unique_keywords(cls, values: list[str]) -> list[str]: if _normalised_duplicates(values): raise ValueError("evidence keywords must be unique") return values @model_validator(mode="after") def validate_sensitive_data_reference(self) -> Self: auxiliary_values = [ self.source_reference or "", *(str(key) for key in self.metrics), *(str(value) for value in self.metrics.values()), *self.keywords, ] auxiliary_text = " ".join(auxiliary_values) sensitive_scan_text = f"{self.content} {auxiliary_text}" if ( _KOREAN_RESIDENT_ID_PATTERN.search(sensitive_scan_text) or _EVIDENCE_BANK_PATTERN.search(sensitive_scan_text) or _EVIDENCE_PASSPORT_PATTERN.search(sensitive_scan_text) ): raise ValueError( "national ID, passport, and bank account values are prohibited " "in all evidence fields" ) if _EVIDENCE_EMAIL_PATTERN.search(auxiliary_text) or _EVIDENCE_PHONE_PATTERN.search( auxiliary_text ): raise ValueError( "contact details belong in ContactInfo and cannot enter evidence metadata" ) if any(_EVIDENCE_SECRET_KEY_PATTERN.search(str(key)) for key in self.metrics): raise ValueError("authentication secrets cannot enter evidence metrics") if _EVIDENCE_SECRET_VALUE_PATTERN.search( f"{self.content} {auxiliary_text}" ): raise ValueError("authentication secret values cannot enter evidence") if _EVIDENCE_HEALTH_TERM_PATTERN.search(sensitive_scan_text): raise ValueError("health data must never enter evidence metadata") if _EVIDENCE_BANK_PATTERN.search(self.content): raise ValueError("bank account values must never enter candidate evidence") if _EVIDENCE_EMAIL_PATTERN.search(self.content) or _EVIDENCE_PHONE_PATTERN.search( self.content ): raise ValueError( "contact details belong in ContactInfo and cannot enter evidence content" ) detected_categories = { category for category, patterns in _EVIDENCE_SENSITIVE_PATTERNS if any(pattern.search(sensitive_scan_text) for pattern in patterns) } prohibited_detected = detected_categories & PROHIBITED_SENSITIVE_CATEGORIES if prohibited_detected: raise ValueError( "prohibited health, political opinion, property, national ID, " "or bank account data must never enter evidence" ) if len(detected_categories) > 1: raise ValueError( "evidence contains multiple sensitive categories; split or remove it" ) if detected_categories and self.sensitive_category not in detected_categories: detected = next(iter(detected_categories)).value raise ValueError( f"detected sensitive content requires category {detected!r} and consent" ) if self.sensitive_category in PROHIBITED_SENSITIVE_CATEGORIES: raise ValueError( "prohibited health, political opinion, property, national IDs, " "and bank accounts must never enter a resume" ) if self.sensitive_category is not None and self.consent_id is None: raise ValueError("sensitive evidence requires an explicit consent_id") if self.sensitive_category is None and self.consent_id is not None: raise ValueError("consent_id is only valid for sensitive evidence") return self @property def statement(self) -> str: """Readable compatibility name for the factual content.""" return self.content class CandidateFact(EvidenceItem): """Semantic alias retained for callers that refer to candidate facts.""" class CandidateProfile(DomainModel): candidate_id: Identifier name: ShortText name_en: str | None = Field(default=None, max_length=200) contact: ContactInfo headline: str | None = Field(default=None, max_length=300) summary: str | None = Field(default=None, max_length=2_000) facts: list[EvidenceItem] = Field(min_length=1, max_length=1_000) records: ResumeRecords = Field(default_factory=ResumeRecords) consents: list[SensitiveDataConsent] = Field(default_factory=list, max_length=100) locale: Literal["ko-KR"] = "ko-KR" updated_at: AwareDatetime = Field(default_factory=_utc_now) @model_validator(mode="after") def validate_evidence_and_consents(self) -> Self: duplicate_evidence = _duplicates([fact.evidence_id for fact in self.facts]) if duplicate_evidence: raise ValueError(f"duplicate evidence_id values: {duplicate_evidence}") duplicate_consents = _duplicates([item.consent_id for item in self.consents]) if duplicate_consents: raise ValueError(f"duplicate consent_id values: {duplicate_consents}") consent_by_id = {item.consent_id: item for item in self.consents} self.records.assert_evidence_integrity( {fact.evidence_id: fact.category.value for fact in self.facts} ) evidence_texts: dict[str, str] = {} for fact in self.facts: date_values: list[str] = [] if fact.date_range is not None: date_values.append(fact.date_range.start.format_ko()) if fact.date_range.end is not None: date_values.append(fact.date_range.end.format_ko()) evidence_texts[fact.evidence_id] = " ".join( [ fact.content, *fact.keywords, *(str(key) for key in fact.metrics), *(str(value) for value in fact.metrics.values()), *date_values, ] ) self.records.assert_value_grounding(evidence_texts) for fact in self.facts: identity_values = [self.name, self.name_en or ""] transmitted_fact_text = " ".join( [ fact.content, *(str(key) for key in fact.metrics), *(str(value) for value in fact.metrics.values()), *fact.keywords, ] ) if any( len(value) >= 2 and _contains_identity_echo(transmitted_fact_text, value) for value in identity_values if value ): raise ValueError( f"evidence {fact.evidence_id!r} contains candidate identity; " "keep identity separate from facts" ) if fact.sensitive_category is None: continue consent = consent_by_id.get(fact.consent_id) if consent is None: raise ValueError( f"evidence {fact.evidence_id!r} references unknown consent_id" ) if consent.category != fact.sensitive_category: raise ValueError( f"evidence {fact.evidence_id!r} and consent category differ" ) if not consent.is_active_at(self.updated_at): raise ValueError( f"evidence {fact.evidence_id!r} does not have active consent" ) return self @property def evidence_by_id(self) -> dict[str, EvidenceItem]: return {fact.evidence_id: fact for fact in self.facts} @property def structured_record_by_evidence_id(self) -> dict[str, StructuredRecord]: """Index typed records by the evidence item that proves each record.""" return { evidence_id: record for record in self.records.all_records() for evidence_id in record.evidence_ids } class JobPosting(DomainModel): posting_id: Identifier company_name: ShortText title: ShortText raw_text: NonEmptyText source_url: str | None = Field(default=None, max_length=2_000) location: str | None = Field(default=None, max_length=200) employment_type: EmploymentType | None = None posted_on: date | None = None closes_on: date | None = None collected_at: AwareDatetime = Field(default_factory=_utc_now) @field_validator("source_url") @classmethod def validate_source_url(cls, value: str | None) -> str | None: if value is not None and not re.fullmatch(r"https?://[^\s]+", value): raise ValueError("source_url must be an absolute HTTP(S) URL") return value @model_validator(mode="after") def validate_posting_dates(self) -> Self: if ( self.posted_on is not None and self.closes_on is not None and self.closes_on < self.posted_on ): raise ValueError("job closing date must not precede posting date") return self class RequirementKind(StrEnum): REQUIRED = "required" PREFERRED = "preferred" RESPONSIBILITY = "responsibility" CONTEXT = "context" class RequirementCategory(StrEnum): EXPERIENCE = "experience" SKILL = "skill" EDUCATION = "education" CERTIFICATION = "certification" DOMAIN = "domain" LANGUAGE = "language" BEHAVIOUR = "behaviour" OTHER = "other" class JobRequirement(DomainModel): requirement_id: Identifier text: NonEmptyText kind: RequirementKind category: RequirementCategory priority: int = Field(default=3, ge=1, le=5) source_quote: NonEmptyText classification_quote: str | None = Field(default=None, max_length=2_000) keywords: list[ShortText] = Field(default_factory=list, max_length=50) @field_validator("keywords") @classmethod def validate_keywords(cls, values: list[str]) -> list[str]: if _normalised_duplicates(values): raise ValueError("requirement keywords must be unique") return values @model_validator(mode="after") def require_classification_provenance(self) -> Self: if self.kind not in {RequirementKind.REQUIRED, RequirementKind.PREFERRED}: return self evidence = self.classification_quote or self.source_quote marker = ( _REQUIRED_MARKER_PATTERN if self.kind is RequirementKind.REQUIRED else _PREFERRED_MARKER_PATTERN ) if marker.search(evidence) is None: raise ValueError( f"{self.kind.value} requirement needs a matching " "classification_quote from the posting" ) opposite_marker = ( _PREFERRED_MARKER_PATTERN if self.kind is RequirementKind.REQUIRED else _REQUIRED_MARKER_PATTERN ) if opposite_marker.search(evidence) is not None: raise ValueError( f"{self.kind.value} classification_quote contains an opposing marker" ) if self.classification_quote is not None and not _source_quote_occurs( self.classification_quote, self.source_quote ): raise ValueError( "classification_quote must be one contiguous posting excerpt " "that also contains source_quote" ) if ( self.classification_quote is not None and not _classification_marker_is_local( self.classification_quote, self.source_quote, marker ) ): raise ValueError( "classification_quote marker and source_quote must be in the " "same posting section" ) return self def _requirement_anchors(requirement: JobRequirement) -> set[str]: return _semantic_anchors( " ".join( [ requirement.text, requirement.source_quote, *requirement.keywords, ] ) ) def _claim_mentions_requirement( claim_text: str, requirement: JobRequirement ) -> bool: """Conservatively validate a claim-to-requirement scoring link.""" return _text_supports_requirement(claim_text, requirement, direct=True) def _text_supports_requirement( text: str, requirement: JobRequirement, *, direct: bool ) -> bool: """Reject a coincidental shared noun while preserving short tech skills.""" requirement_anchors = _requirement_anchors(requirement) text_anchors = _semantic_anchors(text) matched_anchors = requirement_anchors & text_anchors if not matched_anchors: return False requirement_signals = _high_signal_tokens( " ".join([requirement.text, *requirement.keywords]) ) if requirement_signals: if not requirement_signals <= _high_signal_tokens(text): return False core_anchors = _semantic_anchors( " ".join([requirement.text, *requirement.keywords]) ) signal_anchors = _semantic_anchors(" ".join(requirement_signals)) specific_anchors = { anchor for anchor in core_anchors - signal_anchors if anchor not in _TECH_REQUIREMENT_CONTEXT_TERMS and re.fullmatch(r"\d+(?:\.\d+)?", anchor) is None } if not specific_anchors <= text_anchors: return False # A fully matched explicit technology token is sufficient for a short # technology requirement (for example Python, Kafka, C++, or CI/CD), # even when the posting models it as an experience/responsibility. return True if not direct or len(requirement_anchors) == 1: return True if any( len(anchor) >= 4 and re.fullmatch(r"[가-힣]+", anchor) for anchor in matched_anchors ): # A distinctive Korean domain term such as "데이터베이스" or # "모니터링" can stand alone; short generic nouns such as "고객" # cannot validate a composite DIRECT requirement. return True return len(matched_anchors) >= 2 def _evidence_supports_requirement( evidence: EvidenceItem, requirement: JobRequirement, *, direct: bool ) -> bool: evidence_text = " ".join( [ evidence.content, *evidence.keywords, *(str(key) for key in evidence.metrics), *(str(value) for value in evidence.metrics.values()), ] ) return _text_supports_requirement(evidence_text, requirement, direct=direct) class ConstraintKind(StrEnum): BLIND_FIELD = "blind_field" REDACTION = "redaction" REQUIRED_SECTION = "required_section" CHARACTER_LIMIT = "character_limit" FILE_FORMAT = "file_format" EMPLOYER_TEMPLATE = "employer_template" OTHER = "other" class PostingConstraint(DomainModel): """One application rule extracted verbatim from a job posting. ``fields`` intentionally uses posting vocabulary rather than a universal privacy enum because Korean public institutions differ on items such as school names, employer names, and identifying email domains. """ constraint_id: Identifier kind: ConstraintKind description: NonEmptyText source_quote: NonEmptyText fields: list[ShortText] = Field(default_factory=list, max_length=100) section: str | None = Field(default=None, max_length=300) max_characters: int | None = Field(default=None, ge=1, le=100_000) formats: list[ShortText] = Field(default_factory=list, max_length=20) blocking: bool = True @model_validator(mode="after") def validate_typed_payload(self) -> Self: if _normalised_duplicates(self.fields): raise ValueError("posting constraint fields must be unique") if _normalised_duplicates(self.formats): raise ValueError("posting constraint formats must be unique") if self.kind in {ConstraintKind.BLIND_FIELD, ConstraintKind.REDACTION}: if not self.fields: raise ValueError("blind and redaction constraints require fields") if self.kind is ConstraintKind.CHARACTER_LIMIT: if self.max_characters is None or self.section is None: raise ValueError( "character limit constraints require section and max_characters" ) if self.kind is ConstraintKind.FILE_FORMAT and not self.formats: raise ValueError("file format constraints require formats") return self def _constraint_payload_is_grounded(constraint: PostingConstraint) -> bool: """Verify typed constraint values against the quoted posting text.""" quote = unicodedata.normalize("NFKC", constraint.source_quote).casefold() quote_anchors = _semantic_anchors(quote) if constraint.kind is ConstraintKind.CHARACTER_LIMIT: assert constraint.max_characters is not None assert constraint.section is not None number_matches = list(re.finditer(r"(? list[str]: if _normalised_duplicates(values): raise ValueError("analysis keywords must be unique") return values @model_validator(mode="after") def validate_requirement_ids(self) -> Self: duplicates = _duplicates( [requirement.requirement_id for requirement in self.requirements] ) if duplicates: raise ValueError(f"duplicate requirement_id values: {duplicates}") duplicate_constraints = _duplicates( [constraint.constraint_id for constraint in self.constraints] ) if duplicate_constraints: raise ValueError( f"duplicate constraint_id values: {duplicate_constraints}" ) return self def assert_matches_posting(self, posting: JobPosting) -> Self: if posting.posting_id != self.posting_id: raise ValueError("job analysis references a different posting") missing_quotes = [ requirement.requirement_id for requirement in self.requirements if not _source_quote_occurs(posting.raw_text, requirement.source_quote) ] missing_constraint_quotes = [ constraint.constraint_id for constraint in self.constraints if not _source_quote_occurs(posting.raw_text, constraint.source_quote) ] missing_classification_quotes = [ requirement.requirement_id for requirement in self.requirements if requirement.classification_quote is not None and not _source_quote_occurs( posting.raw_text, requirement.classification_quote ) ] expected_constraint_kinds: dict[str, frozenset[ConstraintKind]] = { "character_limit": frozenset({ConstraintKind.CHARACTER_LIMIT}), "file_format": frozenset({ConstraintKind.FILE_FORMAT}), "employer_template": frozenset({ConstraintKind.EMPLOYER_TEMPLATE}), "required_section": frozenset({ConstraintKind.REQUIRED_SECTION}), "privacy": frozenset( {ConstraintKind.BLIND_FIELD, ConstraintKind.REDACTION} ), } uncovered_blocking_constraints: list[str] = [] for clause_index, (clause, expected_kinds) in enumerate( _posting_blocking_constraint_clauses(posting.raw_text), start=1 ): for expected_kind in sorted(expected_kinds): accepted_kinds = expected_constraint_kinds[expected_kind] if not any( constraint.blocking and constraint.kind in accepted_kinds and _source_quote_occurs(clause, constraint.source_quote) for constraint in self.constraints ): uncovered_blocking_constraints.append( f"clause-{clause_index}:{expected_kind}" ) ungrounded_requirements = [ requirement.requirement_id for requirement in self.requirements if ( not _has_sufficient_source_anchors( requirement.text, requirement.source_quote ) or not _high_signal_tokens(requirement.text) <= _high_signal_tokens(requirement.source_quote) ) ] ungrounded_constraints = [ constraint.constraint_id for constraint in self.constraints if not ( _semantic_anchors(constraint.description) & _semantic_anchors(constraint.source_quote) ) ] ungrounded_constraint_payloads = [ constraint.constraint_id for constraint in self.constraints if not _constraint_payload_is_grounded(constraint) ] misclassified_requirements = [] for requirement in self.requirements: classification = ( requirement.classification_quote or requirement.source_quote ) combined = f"{classification}\n{requirement.source_quote}" required_marker = _REQUIRED_MARKER_PATTERN.search(combined) is not None preferred_marker = _PREFERRED_MARKER_PATTERN.search(combined) is not None if ( requirement.kind is RequirementKind.REQUIRED and preferred_marker ) or ( requirement.kind is RequirementKind.PREFERRED and required_marker ): misclassified_requirements.append(requirement.requirement_id) if ( missing_quotes or missing_constraint_quotes or missing_classification_quotes ): raise ValueError( "job analysis contains source quotes absent from posting: " f"requirements={missing_quotes}, constraints={missing_constraint_quotes}, " f"classifications={missing_classification_quotes}" ) if ( ungrounded_requirements or ungrounded_constraints or ungrounded_constraint_payloads ): raise ValueError( "job analysis text or typed constraint value lacks a meaningful " "anchor in its source quote: " f"requirements={ungrounded_requirements}, " f"constraints={ungrounded_constraints}, " f"constraint_payloads={ungrounded_constraint_payloads}" ) if uncovered_blocking_constraints: raise ValueError( "job analysis omitted an explicit blocking submission constraint: " f"{uncovered_blocking_constraints}" ) if misclassified_requirements: raise ValueError( "job analysis changed an explicit required/preferred marker: " f"requirements={misclassified_requirements}" ) return self class EvidenceMatchType(StrEnum): DIRECT = "direct" TRANSFERABLE = "transferable" PARTIAL = "partial" GAP = "gap" class EvidenceMatch(DomainModel): requirement_id: Identifier evidence_ids: list[Identifier] = Field(default_factory=list, max_length=100) match_type: EvidenceMatchType relevance_score: float = Field(ge=0.0, le=1.0) rationale: str | None = Field(default=None, max_length=2_000) gap_reason: str | None = Field(default=None, max_length=2_000) @model_validator(mode="after") def validate_match(self) -> Self: duplicates = _duplicates(self.evidence_ids) if duplicates: raise ValueError(f"duplicate evidence references: {duplicates}") if self.match_type is EvidenceMatchType.GAP: if self.evidence_ids: raise ValueError("gap matches cannot reference evidence") if self.relevance_score != 0: raise ValueError("gap matches must have relevance_score 0") if not self.gap_reason: raise ValueError("gap matches require gap_reason") else: if not self.evidence_ids: raise ValueError("non-gap matches require evidence") if self.relevance_score <= 0: raise ValueError("non-gap matches require a positive relevance score") if not self.rationale: raise ValueError("non-gap matches require a rationale") return self class EvidenceMap(DomainModel): map_id: Identifier posting_id: Identifier analysis_id: Identifier matches: list[EvidenceMatch] = Field(min_length=1, max_length=500) generated_at: AwareDatetime = Field(default_factory=_utc_now) @model_validator(mode="after") def validate_unique_requirements(self) -> Self: duplicates = _duplicates([match.requirement_id for match in self.matches]) if duplicates: raise ValueError(f"requirement mapped more than once: {duplicates}") return self def assert_referential_integrity( self, profile: CandidateProfile, analysis: JobAnalysis ) -> Self: errors: list[str] = [] if self.analysis_id != analysis.analysis_id: errors.append("map analysis_id does not match analysis") if self.posting_id != analysis.posting_id: errors.append("map posting_id does not match analysis") requirement_by_id = { requirement.requirement_id: requirement for requirement in analysis.requirements } known_requirements = set(requirement_by_id) mapped_requirements = {match.requirement_id for match in self.matches} missing_requirements = sorted(known_requirements - mapped_requirements) unknown_requirements = sorted(mapped_requirements - known_requirements) if missing_requirements: errors.append(f"requirements without a mapping: {missing_requirements}") if unknown_requirements: errors.append(f"unknown requirement references: {unknown_requirements}") known_evidence = set(profile.evidence_by_id) referenced_evidence = { evidence_id for match in self.matches for evidence_id in match.evidence_ids } unknown_evidence = sorted(referenced_evidence - known_evidence) if unknown_evidence: errors.append(f"unknown evidence references: {unknown_evidence}") for match in self.matches: if match.match_type is EvidenceMatchType.GAP: continue requirement = requirement_by_id.get(match.requirement_id) if requirement is None: continue for evidence_id in match.evidence_ids: fact = profile.evidence_by_id.get(evidence_id) if fact is None: continue if not _evidence_supports_requirement( fact, requirement, direct=match.match_type is EvidenceMatchType.DIRECT, ): errors.append( f"evidence match {match.requirement_id!r}/{evidence_id!r} " "lacks a semantic anchor" ) if errors: raise ValueError("; ".join(errors)) return self class ClaimKind(StrEnum): FACTUAL = "factual" POSITIONING = "positioning" class DraftClaim(DomainModel): claim_id: Identifier text: NonEmptyText kind: ClaimKind = ClaimKind.FACTUAL evidence_ids: list[Identifier] = Field(default_factory=list, max_length=100) requirement_ids: list[Identifier] = Field(default_factory=list, max_length=100) sensitive_categories: set[SensitiveDataCategory] = Field(default_factory=set) order: int = Field(default=0, ge=0) @model_validator(mode="after") def validate_grounding(self) -> Self: if not self.evidence_ids: raise ValueError("every draft claim requires supporting evidence") duplicate_evidence = _duplicates(self.evidence_ids) if duplicate_evidence: raise ValueError(f"duplicate claim evidence: {duplicate_evidence}") duplicate_requirements = _duplicates(self.requirement_ids) if duplicate_requirements: raise ValueError(f"duplicate claim requirements: {duplicate_requirements}") if self.sensitive_categories & PROHIBITED_SENSITIVE_CATEGORIES: raise ValueError("draft claims cannot contain prohibited sensitive data") if _KOREAN_RESIDENT_ID_PATTERN.search(self.text): raise ValueError("Korean resident registration numbers are prohibited") return self class SectionType(StrEnum): SUMMARY = "summary" CORE_COMPETENCIES = "core_competencies" EXPERIENCE = "experience" PROJECTS = "projects" EDUCATION = "education" SKILLS = "skills" CERTIFICATIONS = "certifications" AWARDS = "awards" LANGUAGES = "languages" MILITARY_SERVICE = "military_service" OTHER = "other" class PlannedSection(DomainModel): """Bounded section instruction passed to the drafting prompt.""" section_id: Identifier section_type: SectionType heading: ShortText evidence_ids: list[Identifier] = Field(default_factory=list, max_length=500) requirement_ids: list[Identifier] = Field(default_factory=list, max_length=500) bullet_budget: int = Field(ge=1, le=30) order: int = Field(ge=0) @model_validator(mode="after") def validate_references(self) -> Self: duplicate_evidence = _duplicates(self.evidence_ids) if duplicate_evidence: raise ValueError(f"duplicate planned evidence: {duplicate_evidence}") duplicate_requirements = _duplicates(self.requirement_ids) if duplicate_requirements: raise ValueError( f"duplicate planned requirements: {duplicate_requirements}" ) return self class ContentPlan(DomainModel): """Evidence-bounded content plan between matching and prose drafting.""" plan_id: Identifier candidate_id: Identifier posting_id: Identifier | None = None mode: ResumeMode = ResumeMode.PRIVATE_MODERN sections: list[PlannedSection] = Field(min_length=1, max_length=50) created_at: AwareDatetime = Field(default_factory=_utc_now) @model_validator(mode="after") def validate_sections(self) -> Self: duplicate_ids = _duplicates([section.section_id for section in self.sections]) if duplicate_ids: raise ValueError(f"duplicate planned section_id values: {duplicate_ids}") duplicate_orders = _duplicates([str(section.order) for section in self.sections]) if duplicate_orders: raise ValueError(f"duplicate planned section orders: {duplicate_orders}") if self.mode is ResumeMode.PUBLIC_BLIND and any( section.section_type is SectionType.MILITARY_SERVICE for section in self.sections ): raise ValueError("public blind plans cannot include military details") return self def assert_referential_integrity( self, profile: CandidateProfile, analysis: JobAnalysis | None = None, ) -> Self: errors: list[str] = [] if self.candidate_id != profile.candidate_id: errors.append("plan candidate_id does not match profile") if analysis is not None and self.posting_id != analysis.posting_id: errors.append("plan posting_id does not match analysis") known_evidence = set(profile.evidence_by_id) requirement_by_id = ( {item.requirement_id: item for item in analysis.requirements} if analysis is not None else {} ) known_requirements = set(requirement_by_id) for section in self.sections: missing_evidence = sorted(set(section.evidence_ids) - known_evidence) if missing_evidence: errors.append( f"planned section {section.section_id!r} references unknown " f"evidence {missing_evidence}" ) if analysis is not None: missing_requirements = sorted( set(section.requirement_ids) - known_requirements ) if missing_requirements: errors.append( f"planned section {section.section_id!r} references unknown " f"requirements {missing_requirements}" ) if errors: raise ValueError("; ".join(errors)) return self def assert_matches_evidence_map(self, evidence_map: EvidenceMap) -> Self: """Ensure selected evidence/requirement pairs were actually mapped.""" errors: list[str] = [] if self.posting_id != evidence_map.posting_id: errors.append("content plan posting_id does not match evidence map") matches_by_requirement = { match.requirement_id: match for match in evidence_map.matches } for section in self.sections: section_requirements = set(section.requirement_ids) for requirement_id in section_requirements: match = matches_by_requirement.get(requirement_id) if match is None: errors.append( f"planned section {section.section_id!r} uses unmapped " f"requirement {requirement_id!r}" ) elif match.match_type is EvidenceMatchType.GAP: errors.append( f"planned section {section.section_id!r} uses gap " f"requirement {requirement_id!r}" ) elif not set(section.evidence_ids) & set(match.evidence_ids): errors.append( f"planned section {section.section_id!r} has no evidence " f"mapped to requirement {requirement_id!r}" ) for evidence_id in section.evidence_ids: supporting_requirements = { requirement_id for requirement_id in section_requirements if requirement_id in matches_by_requirement and evidence_id in matches_by_requirement[requirement_id].evidence_ids } if not supporting_requirements: errors.append( f"planned section {section.section_id!r} uses evidence " f"{evidence_id!r} outside mapped requirement pairs" ) if errors: raise ValueError("; ".join(errors)) return self class DraftSection(DomainModel): section_id: Identifier section_type: SectionType heading: ShortText claims: list[DraftClaim] = Field(min_length=1, max_length=500) order: int = Field(ge=0) @model_validator(mode="after") def validate_claims(self) -> Self: duplicate_ids = _duplicates([claim.claim_id for claim in self.claims]) if duplicate_ids: raise ValueError(f"duplicate claim_id values: {duplicate_ids}") duplicate_orders = _duplicates([str(claim.order) for claim in self.claims]) if duplicate_orders: raise ValueError(f"duplicate claim order values: {duplicate_orders}") return self class ResumeDraft(DomainModel): draft_id: Identifier candidate_id: Identifier posting_id: Identifier | None = None title: ShortText mode: ResumeMode = ResumeMode.PRIVATE_MODERN sections: list[DraftSection] = Field(min_length=1, max_length=50) generated_at: AwareDatetime = Field(default_factory=_utc_now) @model_validator(mode="after") def validate_structure_and_mode(self) -> Self: duplicate_sections = _duplicates( [section.section_id for section in self.sections] ) if duplicate_sections: raise ValueError(f"duplicate section_id values: {duplicate_sections}") duplicate_orders = _duplicates([str(section.order) for section in self.sections]) if duplicate_orders: raise ValueError(f"duplicate section order values: {duplicate_orders}") all_claim_ids = [ claim.claim_id for section in self.sections for claim in section.claims ] duplicate_claims = _duplicates(all_claim_ids) if duplicate_claims: raise ValueError(f"claim_id values must be globally unique: {duplicate_claims}") if self.mode is ResumeMode.PUBLIC_BLIND: sensitive = { category for section in self.sections for claim in section.claims for category in claim.sensitive_categories } if sensitive: raise ValueError("public blind resume drafts cannot contain sensitive data") return self def assert_referential_integrity( self, profile: CandidateProfile, analysis: JobAnalysis | None = None, ) -> Self: errors: list[str] = [] if self.candidate_id != profile.candidate_id: errors.append("draft candidate_id does not match profile") if analysis is not None and self.posting_id != analysis.posting_id: errors.append("draft posting_id does not match analysis") known_evidence = set(profile.evidence_by_id) requirement_by_id = ( {item.requirement_id: item for item in analysis.requirements} if analysis is not None else {} ) known_requirements = set(requirement_by_id) for section in self.sections: for claim in section.claims: missing_evidence = sorted(set(claim.evidence_ids) - known_evidence) if missing_evidence: errors.append( f"claim {claim.claim_id!r} references unknown evidence " f"{missing_evidence}" ) if analysis is not None: missing_requirements = sorted( set(claim.requirement_ids) - known_requirements ) if missing_requirements: errors.append( f"claim {claim.claim_id!r} references unknown requirements " f"{missing_requirements}" ) for category in claim.sensitive_categories: supporting_facts = [ profile.evidence_by_id[evidence_id] for evidence_id in claim.evidence_ids if evidence_id in profile.evidence_by_id ] if not any( fact.sensitive_category == category for fact in supporting_facts ): errors.append( f"claim {claim.claim_id!r} marks unsupported sensitive " f"category {category.value!r}" ) if errors: raise ValueError("; ".join(errors)) return self def assert_matches_plan(self, plan: ContentPlan) -> Self: """Verify that drafting did not escape the evidence-bounded plan.""" errors: list[str] = [] if self.candidate_id != plan.candidate_id: errors.append("draft candidate_id does not match content plan") if self.posting_id != plan.posting_id: errors.append("draft posting_id does not match content plan") if self.mode is not plan.mode: errors.append("draft mode does not match content plan") planned_by_id = {section.section_id: section for section in plan.sections} drafted_by_id = {section.section_id: section for section in self.sections} missing_sections = sorted(set(planned_by_id) - set(drafted_by_id)) extra_sections = sorted(set(drafted_by_id) - set(planned_by_id)) if missing_sections: errors.append(f"planned sections missing from draft: {missing_sections}") if extra_sections: errors.append(f"unplanned draft sections: {extra_sections}") for section_id in sorted(set(planned_by_id) & set(drafted_by_id)): planned = planned_by_id[section_id] drafted = drafted_by_id[section_id] if drafted.section_type is not planned.section_type: errors.append(f"section {section_id!r} changed planned type") if drafted.order != planned.order: errors.append(f"section {section_id!r} changed planned order") if len(drafted.claims) > planned.bullet_budget: errors.append(f"section {section_id!r} exceeds bullet budget") allowed_evidence = set(planned.evidence_ids) allowed_requirements = set(planned.requirement_ids) for claim in drafted.claims: if not set(claim.evidence_ids) <= allowed_evidence: errors.append( f"claim {claim.claim_id!r} uses unplanned evidence" ) if not set(claim.requirement_ids) <= allowed_requirements: errors.append( f"claim {claim.claim_id!r} uses unplanned requirements" ) if errors: raise ValueError("; ".join(errors)) return self def assert_matches_evidence_map(self, evidence_map: EvidenceMap) -> Self: """Require every claim requirement to share a mapped evidence item.""" errors: list[str] = [] if self.posting_id != evidence_map.posting_id: errors.append("draft posting_id does not match evidence map") matches = {match.requirement_id: match for match in evidence_map.matches} for section in self.sections: for claim in section.claims: claim_evidence = set(claim.evidence_ids) for requirement_id in claim.requirement_ids: match = matches.get(requirement_id) if match is None: errors.append( f"claim {claim.claim_id!r} uses unmapped requirement " f"{requirement_id!r}" ) elif match.match_type is EvidenceMatchType.GAP: errors.append( f"claim {claim.claim_id!r} uses gap requirement " f"{requirement_id!r}" ) elif not claim_evidence & set(match.evidence_ids): errors.append( f"claim {claim.claim_id!r} has no evidence mapped to " f"requirement {requirement_id!r}" ) if errors: raise ValueError("; ".join(errors)) return self def fingerprint(self) -> str: """Return a canonical SHA-256 binding for quality/audit artifacts.""" payload = json.dumps( self.model_dump(mode="json", exclude={"generated_at"}), ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(payload).hexdigest() class QualitySeverity(StrEnum): INFO = "info" WARNING = "warning" ERROR = "error" CRITICAL = "critical" class QualityCategory(StrEnum): EVIDENCE = "evidence" JOB_ALIGNMENT = "job_alignment" COMPLETENESS = "completeness" CONSISTENCY = "consistency" CHRONOLOGY = "chronology" KOREAN_LANGUAGE = "korean_language" READABILITY = "readability" FORMATTING = "formatting" PRIVACY = "privacy" BIAS = "bias" class QualityFinding(DomainModel): finding_id: Identifier code: Identifier severity: QualitySeverity category: QualityCategory message: NonEmptyText location: str | None = Field(default=None, max_length=500) claim_id: Identifier | None = None evidence_ids: list[Identifier] = Field(default_factory=list, max_length=100) suggestion: str | None = Field(default=None, max_length=2_000) @field_validator("evidence_ids") @classmethod def validate_evidence_ids(cls, values: list[str]) -> list[str]: if _duplicates(values): raise ValueError("quality finding evidence references must be unique") return values @computed_field @property def blocking(self) -> bool: return self.severity in {QualitySeverity.ERROR, QualitySeverity.CRITICAL} class QualityReport(DomainModel): report_id: Identifier draft_id: Identifier draft_fingerprint: Annotated[ str, StringConstraints(pattern=r"^[a-f0-9]{64}$") ] | None = None evaluation_fingerprint: Annotated[ str, StringConstraints(pattern=r"^[a-f0-9]{64}$") ] | None = None overall_score: float = Field(default=0.0, ge=0, le=100) evidence_coverage: float = Field(default=0.0, ge=0, le=1) requirement_coverage: float = Field(default=0.0, ge=0, le=1) category_scores: dict[QualityCategory, float] = Field(default_factory=dict) findings: list[QualityFinding] = Field(default_factory=list, max_length=1_000) minimum_score: float = Field(default=90, ge=0, le=100) minimum_evidence_coverage: float = Field(default=1.0, ge=0, le=1) minimum_requirement_coverage: float = Field(default=0.80, ge=0, le=1) evaluated_at: AwareDatetime = Field(default_factory=_utc_now) @field_validator("category_scores") @classmethod def validate_category_scores( cls, values: dict[QualityCategory, float] ) -> dict[QualityCategory, float]: invalid = [score for score in values.values() if not 0 <= score <= 100] if invalid: raise ValueError("all category scores must be between 0 and 100") return values @model_validator(mode="after") def validate_findings(self) -> Self: duplicates = _duplicates([finding.finding_id for finding in self.findings]) if duplicates: raise ValueError(f"duplicate finding_id values: {duplicates}") return self @computed_field @property def passed(self) -> bool: return ( self.overall_score >= self.minimum_score and self.evidence_coverage >= self.minimum_evidence_coverage and self.requirement_coverage >= self.minimum_requirement_coverage and not any(finding.blocking for finding in self.findings) ) @computed_field @property def blocking_count(self) -> int: return sum(finding.blocking for finding in self.findings) class GenerationConfig(DomainModel): output_mode: OutputMode = OutputMode.MARKDOWN resume_mode: ResumeMode = ResumeMode.PRIVATE_MODERN locale: Literal["ko-KR"] = "ko-KR" as_of_date: date = Field(default_factory=date.today) max_pages: int = Field(default=2, ge=1, le=5) strict_evidence: bool = True include_photo: bool = False allowed_sensitive_categories: set[SensitiveDataCategory] = Field( default_factory=set ) employer_required_sensitive_categories: set[SensitiveDataCategory] = Field( default_factory=set ) minimum_quality_score: float = Field(default=90, ge=0, le=100) minimum_evidence_coverage: float = Field(default=1.0, ge=0, le=1) minimum_requirement_coverage: float = Field(default=0.80, ge=0, le=1) date_format: Literal["YYYY.MM", "YYYY.MM.DD"] = "YYYY.MM" section_order: list[SectionType] = Field( default_factory=lambda: [ SectionType.SUMMARY, SectionType.CORE_COMPETENCIES, SectionType.EXPERIENCE, SectionType.PROJECTS, SectionType.EDUCATION, SectionType.SKILLS, SectionType.CERTIFICATIONS, ] ) @model_validator(mode="after") def validate_privacy_configuration(self) -> Self: configured_sensitive = ( self.allowed_sensitive_categories | self.employer_required_sensitive_categories ) if configured_sensitive & PROHIBITED_SENSITIVE_CATEGORIES: raise ValueError( "prohibited health, political opinion, property, national IDs, " "and bank accounts can never be enabled" ) if self.include_photo and SensitiveDataCategory.PHOTO not in ( self.allowed_sensitive_categories ): raise ValueError("include_photo requires PHOTO in allowed sensitive data") if self.resume_mode is ResumeMode.PUBLIC_BLIND: if self.include_photo or configured_sensitive: raise ValueError("public blind mode forbids photo and all sensitive data") elif self.resume_mode is not ResumeMode.EMPLOYER_FORM: if configured_sensitive: raise ValueError( "sensitive data can only be enabled for an employer_form" ) else: unrequested = ( self.allowed_sensitive_categories - self.employer_required_sensitive_categories ) if unrequested: values = sorted(category.value for category in unrequested) raise ValueError( "sensitive data requires a recorded employer requirement: " f"{values}" ) if len(self.section_order) != len(set(self.section_order)): raise ValueError("section_order values must be unique") return self def assert_profile_compatible(self, profile: CandidateProfile) -> Self: """Ensure configured sensitive fields have active candidate consent.""" instant = datetime.combine( self.as_of_date, datetime.min.time(), tzinfo=timezone.utc ) active_categories = { consent.category for consent in profile.consents if consent.is_active_at(instant) } missing = self.allowed_sensitive_categories - active_categories if missing: values = sorted(category.value for category in missing) raise ValueError(f"no active consent for sensitive categories: {values}") return self __all__ = [ "CandidateFact", "CandidateProfile", "ClaimKind", "ContentPlan", "ConstraintKind", "ContactInfo", "DateRange", "DraftClaim", "DraftSection", "EmploymentType", "EvidenceCategory", "EvidenceItem", "EvidenceMap", "EvidenceMatch", "EvidenceMatchType", "EvidenceSource", "GenerationConfig", "JobAnalysis", "JobPosting", "JobRequirement", "OutputMode", "PlannedSection", "PostingConstraint", "QualityCategory", "QualityFinding", "QualityReport", "QualitySeverity", "RequirementCategory", "RequirementKind", "ResumeDate", "ResumeDraft", "ResumeMode", "ResumeRecords", "SectionType", "SensitiveDataCategory", "SensitiveDataConsent", "VerificationStatus", ]