Files

511 lines
20 KiB
Python

"""Deterministic output-contract checks for resume drafts.
The language model may extract application rules, but it is not trusted to
decide whether its own output follows them. This module keeps the measurable
rules independent from prose-quality validation so the same checks can run in
the pipeline, CLI, and renderer.
Character limits use NFC-normalised Unicode code points and count spaces plus
one newline between bullets; section headings are excluded. That convention
is deterministic, but an employer portal with a different counting convention
still needs a dedicated adapter.
"""
from __future__ import annotations
import re
import unicodedata
from dataclasses import dataclass
from datetime import date
from typing import Iterable
from .models import (
ConstraintKind,
DraftClaim,
DraftSection,
GenerationConfig,
JobAnalysis,
OutputMode,
QualityCategory,
QualityFinding,
QualitySeverity,
ResumeDraft,
SectionType,
)
_NON_WORD = re.compile(r"[^0-9a-z가-힣]+", flags=re.I)
_FORMAT_SPLIT = re.compile(
r"\s*(?:,|/|\||\ub610\ub294|\ud639\uc740|\bor\b)\s*", flags=re.I
)
_NUMERIC_DATE = re.compile(
r"(?<!\d)(?P<year>(?:19|20)\d{2})(?P<sep>[./-])"
r"(?P<month>\d{1,2})(?:(?P=sep)(?P<day>\d{1,2}))?(?!\d)"
)
_KOREAN_MONTH_DATE = re.compile(
r"(?<!\d)(?P<year>(?:19|20)\d{2})\s*\ub144\s*"
r"(?P<month>\d{1,2})\s*\uc6d4(?:\s*(?P<day>\d{1,2})\s*\uc77c)?"
)
_SECTION_ALIASES: dict[SectionType, frozenset[str]] = {
SectionType.SUMMARY: frozenset(
{
"summary",
"profile",
"\uc694\uc57d",
"\ud575\uc2ec\uc694\uc57d",
"\ud504\ub85c\ud544",
"\uc9c0\uc6d0\uc790\uc694\uc57d",
}
),
SectionType.CORE_COMPETENCIES: frozenset(
{
"corecompetencies",
"competencies",
"\ud575\uc2ec\uc5ed\ub7c9",
"\uc9c1\ubb34\uc5ed\ub7c9",
"\uc5ed\ub7c9",
}
),
SectionType.EXPERIENCE: frozenset(
{
"experience",
"workexperience",
"\uacbd\ub825",
"\uacbd\ub825\uc0ac\ud56d",
"\uc5c5\ubb34\uacbd\ub825",
"\uc9c1\uc7a5\uacbd\ub825",
}
),
SectionType.PROJECTS: frozenset(
{
"projects",
"project",
"\ud504\ub85c\uc81d\ud2b8",
"\uc8fc\uc694\ud504\ub85c\uc81d\ud2b8",
"\ud504\ub85c\uc81d\ud2b8\uacbd\ud5d8",
}
),
SectionType.EDUCATION: frozenset(
{"education", "\ud559\ub825", "\ud559\ub825\uc0ac\ud56d", "\uad50\uc721", "\uad50\uc721\uc0ac\ud56d"}
),
SectionType.SKILLS: frozenset(
{
"skills",
"skill",
"\uae30\uc220",
"\uae30\uc220\uc2a4\ud0dd",
"\ubcf4\uc720\uae30\uc220",
"\uc9c1\ubb34\uae30\uc220",
}
),
SectionType.CERTIFICATIONS: frozenset(
{
"certifications",
"certificates",
"\uc790\uaca9",
"\uc790\uaca9\uc99d",
"\uc790\uaca9\uc0ac\ud56d",
}
),
SectionType.AWARDS: frozenset(
{"awards", "honors", "\uc218\uc0c1", "\uc218\uc0c1\uacbd\ub825", "\uc218\uc0c1\ub0b4\uc5ed"}
),
SectionType.LANGUAGES: frozenset(
{"languages", "language", "\uc5b4\ud559", "\uc678\uad6d\uc5b4", "\uc5b4\ud559\ub2a5\ub825"}
),
SectionType.MILITARY_SERVICE: frozenset(
{"militaryservice", "\ubcd1\uc5ed", "\ubcd1\uc5ed\uc0ac\ud56d"}
),
SectionType.OTHER: frozenset({"other", "\uae30\ud0c0"}),
}
_FORMAT_ALIASES: dict[OutputMode, frozenset[str]] = {
OutputMode.MARKDOWN: frozenset(
{"md", "markdown", "textmarkdown", "\ub9c8\ud06c\ub2e4\uc6b4"}
),
OutputMode.JSON: frozenset({"json", "applicationjson"}),
OutputMode.HTML: frozenset({"html", "htm", "texthtml"}),
OutputMode.DOCX: frozenset(
{"docx", "word", "msword", "wordprocessingml", "\uc6cc\ub4dc"}
),
OutputMode.PDF: frozenset({"pdf", "applicationpdf"}),
}
@dataclass(frozen=True, slots=True)
class OutputConstraintIssue:
"""One deterministic violation or unsupported blocking requirement."""
code: str
message: str
category: QualityCategory = QualityCategory.FORMATTING
location: str | None = None
claim_id: str | None = None
evidence_ids: tuple[str, ...] = ()
suggestion: str | None = None
blocking: bool = True
class OutputConstraintError(ValueError):
"""Raised when a renderer would emit a contract-breaking document."""
def __init__(self, issues: Iterable[OutputConstraintIssue]) -> None:
self.issues = tuple(issue for issue in issues if issue.blocking)
details = "; ".join(f"{issue.code}: {issue.message}" for issue in self.issues)
super().__init__(details or "output constraint validation failed")
def _normalise_key(value: str) -> str:
return _NON_WORD.sub("", unicodedata.normalize("NFKC", value).casefold())
def _section_keys(section: DraftSection) -> frozenset[str]:
aliases = _SECTION_ALIASES.get(section.section_type, frozenset())
return frozenset(
{
_normalise_key(section.heading),
_normalise_key(section.section_type.value),
*(_normalise_key(alias) for alias in aliases),
}
)
def _matching_sections(draft: ResumeDraft, reference: str) -> list[DraftSection]:
target = _normalise_key(reference)
if not target:
return []
return [section for section in draft.sections if target in _section_keys(section)]
def count_section_characters(sections: Iterable[DraftSection]) -> int:
"""Count semantic section content using the documented portal-neutral rule."""
texts = [
unicodedata.normalize("NFC", claim.text).replace("\r\n", "\n").replace("\r", "\n")
for section in sections
for claim in sorted(section.claims, key=lambda item: (item.order, item.claim_id))
]
return len("\n".join(texts))
def _last_claim(sections: Iterable[DraftSection]) -> DraftClaim | None:
claims = [claim for section in sections for claim in section.claims]
if not claims:
return None
return max(claims, key=lambda item: (item.order, item.claim_id))
def _format_tokens(values: Iterable[str]) -> frozenset[str]:
tokens: set[str] = set()
for value in values:
for part in _FORMAT_SPLIT.split(value):
token = _normalise_key(part.removeprefix("."))
if token.endswith("\ud30c\uc77c"):
token = token[: -len("\ud30c\uc77c")]
if token:
tokens.add(token)
return frozenset(tokens)
def _mode_format_tokens(mode: OutputMode) -> frozenset[str]:
return frozenset({_normalise_key(mode.value), *_FORMAT_ALIASES[mode]})
def _date_targets(draft: ResumeDraft) -> Iterable[tuple[str, str, DraftClaim | None]]:
yield "title", draft.title, None
for section in draft.sections:
yield f"sections.{section.section_id}.heading", section.heading, None
for claim in section.claims:
yield f"claims.{claim.claim_id}.text", claim.text, claim
def _expected_date(year: int, month: int, day: int | None, pattern: str) -> str | None:
try:
if pattern == "YYYY.MM":
if day is not None:
return None
date(year, month, 1)
return f"{year:04d}.{month:02d}"
if day is None:
return None
return date(year, month, day).strftime("%Y.%m.%d")
except ValueError:
return None
def _date_issues(draft: ResumeDraft, config: GenerationConfig) -> list[OutputConstraintIssue]:
issues: list[OutputConstraintIssue] = []
for location, text, claim in _date_targets(draft):
claim_kwargs = {
"claim_id": claim.claim_id if claim else None,
"evidence_ids": tuple(claim.evidence_ids) if claim else (),
}
for match in _NUMERIC_DATE.finditer(text):
day = int(match.group("day")) if match.group("day") else None
expected = _expected_date(
int(match.group("year")), int(match.group("month")), day, config.date_format
)
if expected == match.group(0):
continue
issues.append(
OutputConstraintIssue(
code="OUTPUT.DATE_FORMAT",
message=(
f"\ub0a0\uc9dc {match.group(0)!r}\uc774(\uac00) \uc124\uc815 {config.date_format}\uc640 "
"\uc77c\uce58\ud558\uc9c0 \uc54a\uc2b5\ub2c8\ub2e4."
),
category=QualityCategory.CHRONOLOGY,
location=location,
suggestion=f"\ub0a0\uc9dc\ub97c {config.date_format} \ud615\uc2dd\uc73c\ub85c \ud1b5\uc77c\ud558\uc138\uc694.",
**claim_kwargs,
)
)
for match in _KOREAN_MONTH_DATE.finditer(text):
day = int(match.group("day")) if match.group("day") else None
expected = _expected_date(
int(match.group("year")), int(match.group("month")), day, config.date_format
)
issues.append(
OutputConstraintIssue(
code="OUTPUT.DATE_FORMAT",
message=(
f"\ub0a0\uc9dc {match.group(0)!r}\uc774(\uac00) \uc124\uc815 {config.date_format}\uc640 "
"\uc77c\uce58\ud558\uc9c0 \uc54a\uc2b5\ub2c8\ub2e4."
),
category=QualityCategory.CHRONOLOGY,
location=location,
suggestion=(
f"\ub0a0\uc9dc\ub97c {expected or config.date_format} \ud615\uc2dd\uc73c\ub85c \ud1b5\uc77c\ud558\uc138\uc694."
),
**claim_kwargs,
)
)
return issues
def _section_order_issues(
draft: ResumeDraft, config: GenerationConfig
) -> list[OutputConstraintIssue]:
rank = {section_type: index for index, section_type in enumerate(config.section_order)}
ordered = sorted(draft.sections, key=lambda item: (item.order, item.section_id))
previous: DraftSection | None = None
previous_rank = -1
for section in ordered:
current_rank = rank.get(section.section_type)
if current_rank is None:
continue
if current_rank < previous_rank and previous is not None:
return [
OutputConstraintIssue(
code="OUTPUT.SECTION_ORDER",
message=(
f"\uc139\uc158 {section.heading!r}\uc774(\uac00) \uc124\uc815\ub41c section_order\uc0c1 "
f"{previous.heading!r} \ub4a4\uc5d0 \uc62c \uc218 \uc5c6\uc2b5\ub2c8\ub2e4."
),
location=f"sections.{section.section_id}.order",
suggestion="\ucf58\ud150\uce20 \uacc4\ud68d\uacfc \ucd08\uc548\uc758 \uc139\uc158 \uc21c\uc11c\ub97c \uc124\uc815\uacfc \ub9de\ucd94\uc138\uc694.",
)
]
previous = section
previous_rank = current_rank
return []
def _posting_issues(
draft: ResumeDraft,
analysis: JobAnalysis,
output_mode: OutputMode,
) -> list[OutputConstraintIssue]:
issues: list[OutputConstraintIssue] = []
for constraint in analysis.constraints:
severity_blocking = constraint.blocking
location = f"analysis.constraints.{constraint.constraint_id}"
if constraint.kind is ConstraintKind.REQUIRED_SECTION:
references = [constraint.section] if constraint.section else list(constraint.fields)
references = [reference for reference in references if reference]
if not references:
issues.append(
OutputConstraintIssue(
code="OUTPUT.CONSTRAINT_MALFORMED",
message=(
f"\ud544\uc218 \uc139\uc158 \uc81c\uc57d {constraint.constraint_id!r}\uc5d0 section \ub610\ub294 "
"fields\uac00 \uc5c6\uc2b5\ub2c8\ub2e4."
),
location=location,
blocking=severity_blocking,
suggestion="\uacf5\uace0 \uc6d0\ubb38\uc5d0\uc11c \ud544\uc218 \uc139\uc158\uba85\uc744 \ub2e4\uc2dc \ucd94\ucd9c\ud558\uc138\uc694.",
)
)
continue
for reference in references:
if _matching_sections(draft, reference):
continue
issues.append(
OutputConstraintIssue(
code="OUTPUT.REQUIRED_SECTION",
message=(
f"\uacf5\uace0\uac00 \uc694\uad6c\ud55c \uc139\uc158 {reference!r}\uc774(\uac00) \ucd08\uc548\uc5d0 \uc5c6\uc2b5\ub2c8\ub2e4 "
f"({constraint.constraint_id})."
),
location=location,
blocking=severity_blocking,
suggestion="\uadfc\uac70\uac00 \uc788\ub294 \ud574\ub2f9 \uc139\uc158\uc744 \ucf58\ud150\uce20 \uacc4\ud68d\uc5d0 \ucd94\uac00\ud558\uc138\uc694.",
)
)
elif constraint.kind is ConstraintKind.CHARACTER_LIMIT:
# PostingConstraint validation guarantees both values, but the
# defensive guard keeps this module safe for future schema changes.
if not constraint.section or constraint.max_characters is None:
continue
sections = _matching_sections(draft, constraint.section)
if not sections:
issues.append(
OutputConstraintIssue(
code="OUTPUT.CONSTRAINT_SECTION_UNKNOWN",
message=(
f"\uae00\uc790 \uc218 \uc81c\uc57d\uc758 \uc139\uc158 {constraint.section!r}\uc744(\ub97c) "
f"\ucd08\uc548\uc5d0\uc11c \ud655\uc778\ud560 \uc218 \uc5c6\uc2b5\ub2c8\ub2e4 ({constraint.constraint_id})."
),
location=location,
blocking=severity_blocking,
suggestion="\uacf5\uace0\uc758 \uc139\uc158\uba85\uacfc \ucd08\uc548 heading\uc744 \uc77c\uce58\uc2dc\ud0a4\uc138\uc694.",
)
)
continue
actual = count_section_characters(sections)
if actual <= constraint.max_characters:
continue
claim = _last_claim(sections)
issues.append(
OutputConstraintIssue(
code="OUTPUT.CHARACTER_LIMIT",
message=(
f"{constraint.section!r} \uc139\uc158\uc774 {actual}\uc790\ub85c \ucd5c\ub300 "
f"{constraint.max_characters}\uc790\ub97c \ucd08\uacfc\ud569\ub2c8\ub2e4 "
"(NFC, \uacf5\ubc31\u00b7\uc904\ubc14\uafc8 \ud3ec\ud568)."
),
location=location,
claim_id=claim.claim_id if claim else None,
evidence_ids=tuple(claim.evidence_ids) if claim else (),
blocking=severity_blocking,
suggestion="\uc0ac\uc2e4 \uadfc\uac70\ub97c \uc720\uc9c0\ud558\uba74\uc11c \uc911\ubcf5\uacfc \uc218\uc2dd\uc5b4\ub97c \uc904\uc774\uc138\uc694.",
)
)
elif constraint.kind is ConstraintKind.FILE_FORMAT:
allowed = _format_tokens(constraint.formats)
if allowed & _mode_format_tokens(output_mode):
continue
issues.append(
OutputConstraintIssue(
code="OUTPUT.FILE_FORMAT",
message=(
f"\ucd9c\ub825 \ud615\uc2dd {output_mode.value!r}\uc774(\uac00) \uacf5\uace0 \ud5c8\uc6a9 \ud615\uc2dd "
f"{constraint.formats!r}\uc5d0 \ud3ec\ud568\ub418\uc9c0 \uc54a\uc2b5\ub2c8\ub2e4 "
f"({constraint.constraint_id})."
),
location=location,
blocking=severity_blocking,
suggestion="\uacf5\uace0\uac00 \ud5c8\uc6a9\ud55c \ud30c\uc77c \ud615\uc2dd\uc758 \uc804\uc6a9 \ub80c\ub354\ub7ec\ub97c \uc0ac\uc6a9\ud558\uc138\uc694.",
)
)
elif constraint.kind is ConstraintKind.EMPLOYER_TEMPLATE:
issues.append(
OutputConstraintIssue(
code="OUTPUT.EMPLOYER_TEMPLATE_UNVERIFIED",
message=(
f"\uc9c0\uc815 \uc591\uc2dd \uc81c\uc57d {constraint.constraint_id!r}\uc740 \ubc94\uc6a9 \ucd08\uc548\uc73c\ub85c "
"\uac80\uc99d\ud560 \uc218 \uc5c6\uc2b5\ub2c8\ub2e4."
),
location=location,
blocking=severity_blocking,
suggestion="\uae30\uad00\uc774 \uc81c\uacf5\ud55c \uc6d0\ubcf8 \uc591\uc2dd \uc804\uc6a9 \uc5b4\ub311\ud130\ub85c \uac80\uc99d\ud558\uc138\uc694.",
)
)
elif constraint.kind is ConstraintKind.OTHER:
issues.append(
OutputConstraintIssue(
code="OUTPUT.UNSUPPORTED_BLOCKING_CONSTRAINT",
message=(
f"제약 {constraint.constraint_id!r}은 결정적으로 검증할 "
"수 있는 유형으로 구조화되지 않았습니다."
),
location=location,
blocking=severity_blocking,
suggestion=(
"공고 원문에서 지원되는 제약 유형으로 다시 추출하거나 "
"전용 검증기를 연결하세요."
),
)
)
return issues
def validate_output_constraints(
draft: ResumeDraft,
config: GenerationConfig,
*,
analysis: JobAnalysis | None = None,
output_mode: OutputMode | None = None,
) -> list[OutputConstraintIssue]:
"""Return deterministic draft/output contract issues in stable order.
``max_pages`` is intentionally not estimated here. Markdown, HTML, and
JSON have no physical pagination, and guessing pages from character counts
would create a false release guarantee. A DOCX/PDF renderer must measure
the laid-out artifact and enforce ``max_pages`` in its own postflight.
"""
effective_output_mode = output_mode or config.output_mode
issues = [
*_section_order_issues(draft, config),
*_date_issues(draft, config),
]
if analysis is not None:
issues.extend(_posting_issues(draft, analysis, effective_output_mode))
return issues
def as_quality_findings(
issues: Iterable[OutputConstraintIssue],
) -> list[QualityFinding]:
"""Adapt output issues to the pipeline's repair and release-gate contract."""
return [
QualityFinding(
finding_id=f"output-constraint-{index:04d}",
code=issue.code,
severity=(QualitySeverity.ERROR if issue.blocking else QualitySeverity.WARNING),
category=issue.category,
message=issue.message,
location=issue.location,
claim_id=issue.claim_id,
evidence_ids=list(issue.evidence_ids),
suggestion=issue.suggestion,
)
for index, issue in enumerate(issues, start=1)
]
def raise_for_blocking_output_constraints(
issues: Iterable[OutputConstraintIssue],
) -> None:
blocking = [issue for issue in issues if issue.blocking]
if blocking:
raise OutputConstraintError(blocking)
__all__ = [
"OutputConstraintError",
"OutputConstraintIssue",
"as_quality_findings",
"count_section_characters",
"raise_for_blocking_output_constraints",
"validate_output_constraints",
]