resume-agent/backend/app/enrichment_modules.py

"""创建后丰富模块规格表（PRD §10.2 优先级队列）。

声明式表驱动：FSM 与路由层只读本表，不硬编码模块逻辑。
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from .models import JobType

#: records 中合法的记录键
RECORD_KEYS = (
    "education",
    "work_experience",
    "internship_experience",
    "project_experience",
    "campus_experience",
    "competition",
)


@dataclass(frozen=True, slots=True)
class ModuleSpec:
    name: str  # 队列元素唯一名
    kind: str  # "record_chat" | "record_form" | "tags"
    record_type: str | None  # records 目标键；None = 需用户先选类型或运行时决定
    multi: bool  # 多段模块（确认后发 AddAnother）
    core_fields: tuple[str, ...]  # 核心字段（缺一不进确认）
    optional_fields: tuple[str, ...]
    components: tuple[str, ...]  # 须进 STAGE_COMPONENTS 白名单
    prompt: str  # 模块开放式提问话术
    skippable: bool = True


_CONFIRM = ("RecordFields", "ExperienceConfirmCard")
_CONFIRM_MULTI = ("RecordFields", "ExperienceConfirmCard", "AddAnother")


def _record(
    name: str,
    record_type: str | None,
    multi: bool,
    prompt: str,
    picker: bool = False,
    kind: str = "record_fields",
) -> ModuleSpec:
    components = (("ChoiceChips",) if picker else ()) + (_CONFIRM_MULTI if multi else _CONFIRM)
    return ModuleSpec(
        name=name,
        kind=kind,
        record_type=record_type,
        multi=multi,
        core_fields=(),
        optional_fields=("description",),
        components=components,
        prompt=prompt,
    )


ENRICHMENT_MODULES: dict[str, ModuleSpec] = {
    spec.name: spec
    for spec in (
        _record(
            "internship",
            "internship_experience",
            True,
            "补充一段实习经历，包括公司、职位和时间。",
        ),
        _record(
            "more_work",
            "work_experience",
            True,
            "还有其他工作经历吗？填写公司、职位和时间。",
        ),
        _record(
            "project",
            "project_experience",
            True,
            "填写一个你做过的项目，包括名称、你的角色和时间。",
        ),
        _record(
            "education",
            "education",
            True,
            "补充一段教育经历，包括学校、专业、学历和时间。",
        ),
        _record(
            "campus_experience",
            "campus_experience",
            True,
            "补充一段校园经历，包括组织、角色和时间。",
        ),

        ModuleSpec(
            name="competition",
            kind="record_form",
            record_type="competition",
            multi=True,
            core_fields=("name", "award", "date"),
            optional_fields=("description",),
            components=("CompetitionFields", "ExperienceConfirmCard", "AddAnother"),
            prompt="有竞赛获奖经历吗？填写竞赛名称、奖项和获奖月份。",
        ),
        ModuleSpec(
            name="skills",
            kind="tags",
            record_type=None,
            multi=False,
            core_fields=(),
            optional_fields=("skills",),
            components=("TagsInput",),
            prompt="列一下你的技能，逐个添加，可以留空。",
        ),
        ModuleSpec(
            name="certificates",
            kind="tags",
            record_type=None,
            multi=False,
            core_fields=(),
            optional_fields=("certificates",),
            components=("TagsInput",),
            prompt="列一下你的证书，可以留空。",
        ),
    )
}

ENRICHMENT_QUEUES: dict[JobType, tuple[str, ...]] = {
    JobType.CAMPUS: (
        "internship",
        "project",
        "competition",
        "skills",
        "certificates",
    ),
    JobType.SOCIAL: (
        "more_work",
        "project",
        "education",
        "skills",
        "certificates",
    ),
    JobType.INTERNSHIP: (
        "campus_experience",
        "project",
        "competition",
        "skills",
        "certificates",
    ),
}


def module_by_name(name: str) -> ModuleSpec:
    return ENRICHMENT_MODULES[name]


PICKER_OPTIONS: dict[str, tuple[tuple[str, str], ...]] = {}

TAG_SEQUENCE = ("skills", "certificates")
TAG_TITLES = {"skills": "你的技能", "certificates": "你的证书"}

_DEFAULT_SKILLS = ("沟通协调", "问题解决", "团队协作")
_SKILL_RULES = (
    (("前端", "frontend", "web"), ("HTML", "CSS", "JavaScript", "TypeScript", "Vue", "React", "Git")),
    (("java",), ("Java", "Spring Boot", "MySQL", "Redis", "Git")),
    (("后端", "backend"), ("Python", "Java", "MySQL", "Redis", "Docker", "Git")),
    (("数据", "算法", "ai", "人工智能"), ("Python", "SQL", "Pandas", "机器学习", "Git")),
    (("产品",), ("需求分析", "原型设计", "数据分析", "项目管理")),
)


_DESCRIPTION_SKILLS = (
    (("python",), "Python"),
    (("fastapi",), "FastAPI"),
    (("django",), "Django"),
    (("flask",), "Flask"),
    (("java",), "Java"),
    (("spring boot", "springboot", "spring"), "Spring Boot"),
    (("mysql",), "MySQL"),
    (("postgres", "postgresql"), "PostgreSQL"),
    (("redis",), "Redis"),
    (("docker",), "Docker"),
    (("kubernetes", "k8s"), "Kubernetes"),
    (("vue",), "Vue"),
    (("react",), "React"),
    (("typescript",), "TypeScript"),
    (("javascript",), "JavaScript"),
    (("sql",), "SQL"),
    (("pandas",), "Pandas"),
    (("机器学习", "machine learning"), "机器学习"),
)


def skill_suggestions(
    target_position: str | None, profile: dict[str, Any] | None = None
) -> list[str]:
    """Suggest skills from target role and facts already supplied by the user."""
    normalized = (target_position or "").strip().lower()
    base_suggestions: list[str] = []
    for keywords, rule_suggestions in _SKILL_RULES:
        if any(keyword in normalized for keyword in keywords):
            base_suggestions = list(rule_suggestions)
            break
    if not base_suggestions:
        base_suggestions = list(_DEFAULT_SKILLS)

    profile = profile or {}
    supplied = _profile_text(profile).lower()
    for aliases, skill in _DESCRIPTION_SKILLS:
        if any(alias in supplied for alias in aliases):
            base_suggestions.append(skill)

    existing = {
        str(skill).strip().casefold()
        for skill in ((profile.get("tags") or {}).get("skills") or [])
        if str(skill).strip()
    }
    result: list[str] = []
    seen: set[str] = set()
    for skill in base_suggestions:
        key = skill.casefold()
        if key not in seen and key not in existing:
            result.append(skill)
            seen.add(key)
    return result[:8]


def _profile_text(profile: dict[str, Any]) -> str:
    """Collect user-entered descriptions only; never infer skills from resume examples."""
    values: list[str] = []
    entries: list[Any] = [profile.get("anchor")]
    entries.extend(profile.get("experiences") or [])
    for records in (profile.get("records") or {}).values():
        entries.extend(records or [])
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        description = entry.get("description")
        if description:
            values.append(str(description))
        values.extend(str(item) for item in (entry.get("highlights") or []) if item)
    return "\n".join(values)