添加简历诊断功能

2026-04-07 20:15:43 +08:00
parent 602f226377
commit 8ffcb351a6
10 changed files with 1004 additions and 10 deletions
@@ -0,0 +1,427 @@
 # 简历诊断功能 - 技术方案
 ## 一、功能概述
 对用户已有简历的**描述文本**进行 AI 诊断，找出问题、给出改进建议和 AI 改写版本，生成诊断报告。
 ### 诊断范围（只诊断描述文本）
 | 来源表 | 字段 | module_type |
 |--------|------|-------------|
 | bg_user_resume | summary（个人概述） | summary |
 | bg_user_resume_education | description | education |
 | bg_user_resume_work | description | work |
 | bg_user_resume_internship | description | internship |
 | bg_user_resume_project | description | project |
 | bg_user_resume_competition | description | competition |
 ### 执行策略：先分后合
 ```
 第一阶段：所有模块记录并行 AI 诊断（asyncio.gather）
 第二阶段：汇总 issues → 代码算评级 → AI 生成整体评价 → 写入数据库
 ```
 AI 模型：`LLM.DOUBAO_SEED_PRO`，暂不接入功能权限校验。
 ---
 ## 二、问题分类与诊断判断思路
 ### 2.1 紧急修复（urgent）
 #### typo — 错别字 / 语病
 **判断思路：**
 - 检查文本中是否存在明显的**错别字**（如"功则"应为"功能"、"负责"写成"付责"）
 - 检查是否存在**语病**：主谓搭配不当、语序混乱、成分残缺
 - 检查**标点符号**使用错误：中英文标点混用、缺少句号、逗号过多造成长句
 - 检查**用词不当**：近义词误用、口语化表达出现在正式简历中
 **AI 判断信号：**
 - 同音字替换（的/得/地 混用）
 - 句子读不通顺，需要反复阅读才能理解
 - 专业术语拼写错误
 - 一句话中出现多个逗号，缺少句号断句
 ---
 ### 2.2 重点优化（important）
 #### no_result — 缺少成果
 **判断思路：**
 - 描述只写了"做了什么"（任务/职责），没有写"做出了什么结果"
 - 典型的**流水账式描述**：只有动作没有产出
 - 缺少对业务/团队/项目的**实际贡献**和**影响**
 **AI 判断信号：**
 - 只有动词+宾语（"负责XX系统开发"、"参与XX项目"），没有后续的结果说明
 - 全是过程描述，找不到"提升了"、"优化了"、"实现了"、"完成了"等结果性表述
 - 对比 STAR 法则：有 Situation + Task + Action，但缺 Result
 **正面示例（有成果）：**
 > 负责用户中心系统重构，将接口响应时间从 800ms 降至 200ms，用户投诉率下降 60%
 **反面示例（缺成果）：**
 > 负责用户中心系统重构，使用 Spring Boot + Redis 实现了新的架构
 #### no_quantify — 缺少量化
 **判断思路：**
 - 有成果描述但**没有具体数字**支撑
 - 使用了模糊表达："大幅提升"、"显著改善"、"大量用户"，但没有具体数值
 - 缺少以下任何维度的量化：人数、金额、百分比、时间周期、覆盖范围、处理规模
 **AI 判断信号：**
 - 出现"大幅"、"显著"、"有效"、"极大"等模糊程度副词，但没有跟随数字
 - 提到了结果但只是定性描述，没有定量数据
 - 可以合理推断应该有数据但未提供（如"提升了性能"没说提升多少）
 **正面示例（有量化）：**
 > 优化数据库查询，将平均响应时间从 2s 降至 200ms，日处理订单量从 5 万提升至 20 万
 **反面示例（缺量化）：**
 > 优化数据库查询，显著提升了系统性能，改善了用户体验
 #### weak_relevance — 岗位相关性弱
 > **前置条件：仅在 `target_position` 有值时才判断，未填目标岗位则跳过此维度，计数为 0。**
 **判断思路：**
 - 需要结合 `target_position`（目标岗位）进行判断
 - 描述内容与目标岗位的**核心职责**关联度低
 - 花大量篇幅描述与目标岗位**无关的技能或经历**
 - 对于目标岗位来说，这段描述**无法体现匹配度**
 **AI 判断信号：**
 - 目标岗位是"Java后端工程师"，但描述中全是前端或运营内容
 - 描述的技能/工具与目标岗位的 JD 常见要求差距大
 - 可转移技能存在但未被强调，反而突出了无关内容
 ---
 ### 2.3 表达提升（expression）
 #### not_concise — 表述不精炼
 **判断思路：**
 - 句子**偏长**（单句超过 50 字），信息密度低
 - 存在**赘词和重复表达**："进行了开发"可简化为"开发了"
 - **信息堆叠**：一句话塞了太多内容，应拆分为多个要点
 - 使用了**空泛的修饰词**："充分"、"积极"、"认真"等无实质信息
 **AI 判断信号：**
 - 单个描述段落超过 80 字但核心信息只有一个
 - 出现"进行了"、"完成了对...的"、"负责了...的工作"等冗余句式
 - 同一段落中重复表达相似的意思
 - 可以删去一半文字而不损失关键信息
 **正面示例（精炼）：**
 > 设计并实现分布式缓存方案，QPS 从 1000 提升至 8000，缓存命中率 95%
 **反面示例（不精炼）：**
 > 在项目中，我积极主动地参与了分布式缓存方案的设计与实现工作，通过对缓存策略的深入研究和反复优化，最终成功地将系统的 QPS 从原来的 1000 提升到了 8000
 #### format_inconsistent — 格式不统一
 **判断思路：**
 - 同一份简历中**时间格式不统一**（有的写"2023.06"，有的写"2023年6月"，有的写"2023/06"）
 - **标点风格不统一**：有的段落用分号结尾，有的用句号，有的不加标点
 - **数字写法不统一**：有的用阿拉伯数字，有的用中文数字
 - **项目符号不统一**：有的用"•"，有的用"-"，有的用"1."
 - **人称不统一**：有的用"我"，有的用第三人称，有的省略主语
 **AI 判断信号：**
 - 同一段描述中出现两种以上的格式风格
 - 与模块上下文中的时间格式不一致
 - 段落之间的排版结构差异明显
 ---
 ## 三、综合评级规则
 由代码硬算，不依赖 AI 判断：
 | 评级 | 条件 | 评语 |
 |------|------|------|
 | A（优秀） | urgent=0, important<=1, expression<=1 | 您的简历相当出彩，在求职市场中格外抢眼，能清晰展现您的优势与经历，已经超越绝大多数候选人了。 |
 | B（良好） | urgent=0, important 2-3, expression<=2 | 简历已经很棒了，但还有提升的潜力。再调整一下细节，会有更具有竞争力！ |
 | C（一般） | urgent=1, 或 important 3-4 | 你的简历还有打磨空间，多推敲细节、补充些具体内容，整体会更出彩。 |
 | D（待提升） | urgent>=2, 或 (important>=4 且 has_weak_relevance) | 您的简历目前还有较大提升空间，建议尽快补充关键经历、完善内容表达，并优化整体结构。 |
 判断优先级：从 D → C → B → A 依次判断，命中即返回。
 ---
 ## 四、数据库表设计
 ### bg_resume_diagnosis_report
 | 字段 | 类型 | 说明 |
 |------|------|------|
 | id | BigInteger | 主键，雪花ID |
 | resume_id | BigInteger | 关联 bg_user_resume.id |
 | user_id | BigInteger | 用户ID |
 | grade | VARCHAR(1) | 评级 A/B/C/D |
 | summary | TEXT | AI 生成的整体评价 |
 | urgent_total | Integer | 紧急修复总数 |
 | important_total | Integer | 重点优化总数 |
 | expression_total | Integer | 表达提升总数 |
 | create_time | DateTime | 创建时间 |
 | update_time | DateTime | 更新时间 |
 ### bg_resume_diagnosis_issue
 | 字段 | 类型 | 说明 |
 |------|------|------|
 | id | BigInteger | 主键，雪花ID |
 | report_id | BigInteger | 关联 report.id |
 | resume_id | BigInteger | 关联 bg_user_resume.id |
 | user_id | BigInteger | 用户ID |
 | module_type | VARCHAR(32) | summary/education/work/internship/project/competition |
 | module_record_id | BigInteger | 模块记录ID（summary 时为 resume_id） |
 | finding | TEXT | 诊断发现 |
 | importance | TEXT | 为什么重要 |
 | suggestion | TEXT | 改进建议 |
 | urgent_issues | JSON | {"typo": 0} |
 | important_issues | JSON | {"no_result": 0, "no_quantify": 0, "weak_relevance": 0} |
 | expression_issues | JSON | {"not_concise": 0, "format_inconsistent": 0} |
 | optimized_content | JSON | AI改写后的内容。子表模块（education/work/internship/project/competition）与原 description 格式一致 `[{id, text}]`，保持原 id 不变只改写 text；summary 模块为纯文本字符串 |
 | status | Integer | 0=待处理 1=已处理 |
 | user_feedback | Integer | 0=未评价 1=符合 2=不符合 |
 | create_time | DateTime | 创建时间 |
 | update_time | DateTime | 更新时间 |
 ### 建表 SQL
 ```sql
 CREATE TABLE `bg_resume_diagnosis_report` (
  `id` bigint NOT NULL COMMENT '主键，雪花ID',
  `resume_id` bigint NOT NULL COMMENT '关联bg_user_resume.id',
  `user_id` bigint NOT NULL COMMENT '用户ID',
  `grade` varchar(1) DEFAULT NULL COMMENT '评级 A/B/C/D',
  `summary` text COMMENT 'AI生成的整体评价',
  `urgent_total` int NOT NULL DEFAULT '0' COMMENT '紧急修复总数',
  `important_total` int NOT NULL DEFAULT '0' COMMENT '重点优化总数',
  `expression_total` int NOT NULL DEFAULT '0' COMMENT '表达提升总数',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  KEY `idx_resume_id` (`resume_id`),
  KEY `idx_user_id` (`user_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='简历诊断报告表';
 CREATE TABLE `bg_resume_diagnosis_issue` (
  `id` bigint NOT NULL COMMENT '主键，雪花ID',
  `report_id` bigint NOT NULL COMMENT '关联report.id',
  `resume_id` bigint NOT NULL COMMENT '关联bg_user_resume.id',
  `user_id` bigint NOT NULL COMMENT '用户ID',
  `module_type` varchar(32) NOT NULL COMMENT '模块类型: summary/education/work/internship/project/competition',
  `module_record_id` bigint NOT NULL COMMENT '模块记录ID，summary时为resume_id',
  `finding` text COMMENT '诊断发现',
  `importance` text COMMENT '为什么重要',
  `suggestion` text COMMENT '改进建议',
  `urgent_issues` json DEFAULT NULL COMMENT '紧急修复子类型计数 {"typo": 0}',
  `important_issues` json DEFAULT NULL COMMENT '重点优化子类型计数 {"no_result": 0, "no_quantify": 0, "weak_relevance": 0}',
  `expression_issues` json DEFAULT NULL COMMENT '表达提升子类型计数 {"not_concise": 0, "format_inconsistent": 0}',
  `optimized_content` json DEFAULT NULL COMMENT 'AI改写后的内容，子表模块与原description格式一致[{id,text}]保持原id只改写text，summary模块为纯文本字符串',
  `status` int NOT NULL DEFAULT '0' COMMENT '0=待处理 1=已处理',
  `user_feedback` int NOT NULL DEFAULT '0' COMMENT '0=未评价 1=符合 2=不符合',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  KEY `idx_report_id` (`report_id`),
  KEY `idx_resume_id` (`resume_id`),
  KEY `idx_user_id` (`user_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='简历诊断问题表';
 ```
 ---
 ## 五、API 设计
 ### 1. POST /resume/diagnose — 触发诊断
 **请求体：** `{"resume_id": 123}`
 **执行流程：**
 ```
 ┌──────────────────────────────────────────────────┐
 │ 1. RequestContext.user_id.get() 获取当前用户        │
 ├──────────────────────────────────────────────────┤
 │ 2. 短事务1（只读）                                  │
 │    async for session in get_db():                  │
 │      service = ResumeDiagnoseService(session)      │
 │      resume, tasks = await service                 │
 │          .load_resume_data(resume_id, user_id)     │
 │    → 加载主表 + 5 张子表数据                         │
 │    → 组装 AI 任务列表                                │
 │    → 事务结束，释放数据库连接                          │
 ├──────────────────────────────────────────────────┤
 │ 3. 校验 tasks 非空                                  │
 │    → 空则 raise ValueError("无可诊断内容")           │
 ├──────────────────────────────────────────────────┤
 │ 4. 并行 AI 诊断（不持有数据库连接）                   │
 │    ai_results = await diagnose_all(tasks)          │
 │    → asyncio.gather 并行调用 N 条诊断链              │
 │    → 每条链独立容错，失败返回空结果                    │
 ├──────────────────────────────────────────────────┤
 │ 5. 汇总统计 + 评级（纯计算，无 IO）                  │
 │    → 遍历 ai_results 统计问题数量                    │
 │    → 过滤无问题的记录（所有计数都为 0 则跳过）         │
 │    → 代码硬算评级（_calc_grade）                     │
 ├──────────────────────────────────────────────────┤
 │ 6. AI 生成整体评价（不持有数据库连接）                │
 │    summary = await generate_summary(...)           │
 │    → 传入评级、统计、all_findings                    │
 │    → 返回纯文本评价                                  │
 ├──────────────────────────────────────────────────┤
 │ 7. 短事务2（纯写入，无 AI 调用）                     │
 │    async for session in get_db():                  │
 │      service = ResumeDiagnoseService(session)      │
 │      report_id = await service.save_report(...)    │
 │    → 写入 report + issues 表                        │
 │    → 事务提交                                       │
 └──────────────────────────────────────────────────┘
 ```
 **响应：** `{"reportId": 456}`
 ### 2. GET /resume/diagnose/{resume_id} — 查询最近一次报告
 **执行流程：**
 1. 获取 user_id
 2. 查 report 表（ORDER BY create_time DESC LIMIT 1）
 3. 查该 report 下所有 issues
 4. 返回 `{"report": {...}, "issues": [...]}`
 **响应示例：**
 ```json
 {
  "report": {
    "id": "123",
    "resumeId": "456",
    "grade": "B",
    "summary": "您的简历整体质量良好...",
    "urgentTotal": 0,
    "importantTotal": 3,
    "expressionTotal": 1,
    "createTime": "2026-04-07 10:30:00"
  },
  "issues": [
    {
      "id": "789",
      "moduleType": "work",
      "moduleRecordId": "1001",
      "finding": "工作描述缺少量化数据...",
      "importance": "量化数据能让招聘者...",
      "suggestion": "建议在描述中添加...",
      "urgentIssues": {"typo": 0},
      "importantIssues": {"no_result": 0, "no_quantify": 1, "weak_relevance": 0},
      "expressionIssues": {"not_concise": 0, "format_inconsistent": 0},
      "optimizedContent": [{"id": "abc123", "text": "负责XX系统后端开发，日均处理100万+请求..."}],
      "status": 0,
      "userFeedback": 0
    }
  ]
 }
 ```
 ### 3. PUT /resume/diagnose/issue/{issue_id}/resolve — 标记已处理
 **请求体：** `{"user_feedback": 1}` （1=符合 2=不符合）
 **执行流程：**
 1. 获取 user_id
 2. 查 issue 记录（校验 user_id）
 3. 设置 status=1, user_feedback
 4. 返回 null
 ---
 ## 六、文件变更清单
 ### 新增 7 个文件
 ```
 app/models/resume_diagnosis_report.py      — 诊断报告 ORM
 app/models/resume_diagnosis_issue.py       — 诊断问题 ORM
 app/ai/resume_diagnoser/__init__.py        — 包初始化（空文件）
 app/ai/resume_diagnoser/prompts.py         — Prompt 模板
 app/ai/resume_diagnoser/diagnoser.py       — AI 诊断引擎
 app/services/resume_diagnose_service.py    — 业务逻辑
 app/api/resume_diagnose.py                 — API 路由
 ```
 ### 修改 1 个文件
 ```
 app/main.py                                — 注册新路由（加 2 行）
 ```
 ---
 ## 七、模块设计
 ### AI 诊断模块 (`app/ai/resume_diagnoser/`)
 **prompts.py** — 两个 Prompt 模板：
 - `DIAGNOSE_MODULE_PROMPT`：第一阶段，单条记录诊断，输入 module_type/target_position/context/description_text（子表传原始 JSON `[{id,text}]`，summary 传纯文本），输出 JSON
 - `SUMMARY_PROMPT`：第二阶段，汇总评价，输入统计数据 + 所有 findings，输出纯文本
 **diagnoser.py** — 参照 `app/ai/resume_extractor/extractor.py` 的模式：
 - 诊断链：`ChatPromptTemplate → DOUBAO_SEED_PRO(temperature=0) → JsonOutputParser`
 - 汇总链：`ChatPromptTemplate → DOUBAO_SEED_PRO(temperature=0.3) → StrOutputParser`
 - `diagnose_all(tasks)` — asyncio.gather 并行
 - `generate_summary(...)` — AI 生成整体评价
 - `_safe_invoke()` — 容错
 ### Service 层 (`app/services/resume_diagnose_service.py`)
 ```
 ResumeDiagnoseService(session: AsyncSession)
  ├─ load_resume_data(resume_id, user_id) → (resume, tasks)
  │    加载主表 + 5 张子表，组装 AI 任务列表
  │
  ├─ save_report(resume_id, user_id, grade, summary, stats, tasks, ai_results) → report_id
  │    纯写入：接收已算好的 grade、summary、统计数据，写入 report + issues
  │    跳过无问题的记录（所有计数为 0 则不创建 issue 行）
  │
  ├─ get_latest_report(resume_id, user_id) → dict | None
  └─ resolve_issue(issue_id, user_id, user_feedback) → None
 工具函数（无状态，路由层或 Service 外部可调用）：
  ├─ _build_description_text(description) — 子表传原始 JSON 字符串 [{id,text}]，summary 传纯文本
  ├─ _calc_grade(urgent, important, expression, has_weak_relevance) — 评级硬算
  ├─ _aggregate_results(tasks, ai_results) — 统计汇总 + 过滤无问题记录
  └─ _issue_to_dict(issue) — ORM → camelCase dict
 ```
 **description_text 传入格式：**
 - 子表模块：传原始 JSON 字符串 `[{"id": "abc123", "text": "负责XX系统..."}, ...]`，AI 能看到每个段落的 id
 - summary 模块：传纯文本（summary 字段本身就是 VARCHAR）
 **optimized_content 返回格式：**
 - 子表模块：与原始 description 格式一致 `[{id, text}]`，保持原 id 不变，只改写 text
 - summary 模块：纯文本字符串（与原始 summary 字段格式一致）
 ### API 路由 (`app/api/resume_diagnose.py`)
 ```
 router = APIRouter(prefix="/resume/diagnose", tags=["简历诊断"])
  ├─ POST ""              — 触发诊断
  ├─ GET "/{resume_id}"   — 查询报告
  └─ PUT "/issue/{issue_id}/resolve" — 标记已处理
 ```
 ---
 ## 八、实施顺序
 1. ORM 模型（`resume_diagnosis_report.py` + `resume_diagnosis_issue.py`）
 2. AI 模块（`__init__.py` + `prompts.py` + `diagnoser.py`）
 3. Service 层（`resume_diagnose_service.py`）
 4. API 路由（`resume_diagnose.py`）+ 修改 `main.py`
 5. 更新项目结构文档
@@ -32,13 +32,17 @@ offerpie_python_ai/
    │
    ├─ ai/                        # **AI 能力层**
    │   ├─ models.py              # LLM 模型枚举（LLM.DOUBAO_PRO_256K、DEEPSEEK_V3、GPT_4O 等），基于 LangChain ChatOpenAI
-    │   └─ resume_extractor/      # 简历 AI 提取模块
+    │   ├─ resume_extractor/      # 简历 AI 提取模块
-    │       ├─ prompts.py             # 5 个提取任务的 System Prompt（个人信息/教育/工作+实习/项目/竞赛）
+    │   │   ├─ prompts.py             # 5 个提取任务的 System Prompt（个人信息/教育/工作+实习/项目/竞赛）
-    │       └─ extractor.py           # AI 并行提取（extract_all 入口，asyncio.gather 5 路并行）
+    │   │   └─ extractor.py           # AI 并行提取（extract_all 入口，asyncio.gather 5 路并行）
    │   └─ resume_diagnoser/      # 简历 AI 诊断模块
    │       ├─ prompts.py             # 诊断 Prompt 模板（分模块诊断 + 汇总评价）
    │       └─ diagnoser.py           # AI 并行诊断（diagnose_all 入口 + generate_summary 汇总评价）
    │
    ├─ api/                       # **路由层**（REST API 接口）
    │   ├─ health.py              # 健康检查接口 GET /health/
-    │   └─ resume.py              # 简历接口 POST /resume/upload（上传文件AI解析）
+    │   ├─ resume.py              # 简历接口 POST /resume/upload（上传文件AI解析）
    │   └─ resume_diagnose.py     # 简历诊断接口（POST 触发诊断 / GET 查询报告 / PUT 标记处理）
    │
    ├─ models/                    # **ORM 模型层**（SQLAlchemy 声明式映射）
    │   ├─ func_permission.py             # 功能权限定义表（bg_func_permission）
@@ -49,7 +53,9 @@ offerpie_python_ai/
    │   ├─ user_resume_work.py            # 简历-工作经历表（bg_user_resume_work）
    │   ├─ user_resume_internship.py      # 简历-实习经历表（bg_user_resume_internship）
    │   ├─ user_resume_project.py         # 简历-项目经历表（bg_user_resume_project）
-    │   └─ user_resume_competition.py     # 简历-竞赛经历表（bg_user_resume_competition）
+    │   ├─ user_resume_competition.py     # 简历-竞赛经历表（bg_user_resume_competition）
    │   ├─ resume_diagnosis_report.py     # 简历诊断报告表（bg_resume_diagnosis_report）
    │   └─ resume_diagnosis_issue.py      # 简历诊断问题表（bg_resume_diagnosis_issue）
    │
    ├─ tool/                      # **工具层**（无状态、无业务依赖的通用工具）
    │   ├─ file_parser.py             # 文件解析工具（PDF/Word/TXT → 纯文本，parse_to_text 入口方法）
@@ -57,7 +63,8 @@ offerpie_python_ai/
    │
    └─ services/                  # **业务逻辑层**
        ├─ func_permission_service.py     # 功能权限服务（校验+扣减+回退，逻辑与Java端一致）
-        └─ resume_parse_service.py        # 简历解析服务（文件解析→AI结构化→写入主表+5张子表）
+        ├─ resume_parse_service.py        # 简历解析服务（文件解析→AI结构化→写入主表+5张子表）
        └─ resume_diagnose_service.py    # 简历诊断服务（加载简历→AI并行诊断→统计评级→写入报告）
 ```
 ## 2️⃣ 各层模块职责
@@ -65,11 +72,11 @@ offerpie_python_ai/
 |------|----------|-------------|
 | **config** | 统一配置管理，基于 Pydantic Settings，支持 .env 文件加载 | `Settings`（数据库、Redis、LLM供应商、JWT、CORS、日志等全部配置项） |
 | **core** | 核心基础设施：数据库连接、Redis连接、鉴权、日志、中间件、异常处理、统一响应 | `database.py`、`redis.py`、`auth.py`、`middleware.py`、`exceptions.py`、`logger.py`、`StandardResponse` |
-| **ai** | AI 模型管理 + 业务 AI 能力 | `LLM` 枚举、`resume_extractor/`（简历并行提取：5路 AI 同时提取个人信息/教育/工作+实习/项目/竞赛） |
+| **ai** | AI 模型管理 + 业务 AI 能力 | `LLM` 枚举、`resume_extractor/`（简历并行提取：5路 AI 同时提取个人信息/教育/工作+实习/项目/竞赛）、`resume_diagnoser/`（简历诊断：并行诊断各模块描述 + 汇总评价） |
-| **api** | REST API 路由定义 | `health.py`（健康检查）、`resume.py`（简历上传解析） |
+| **api** | REST API 路由定义 | `health.py`（健康检查）、`resume.py`（简历上传解析）、`resume_diagnose.py`（简历诊断） |
-| **models** | SQLAlchemy ORM 模型，与 Java 端共享同一数据库 | `FuncPermission`、`UserFuncPermissionStock`、`UserFuncUsageLog`、`UserResume`、`UserResumeEducation`/`Work`/`Internship`/`Project`/`Competition` |
+| **models** | SQLAlchemy ORM 模型，与 Java 端共享同一数据库 | `FuncPermission`、`UserFuncPermissionStock`、`UserFuncUsageLog`、`UserResume`、`UserResumeEducation`/`Work`/`Internship`/`Project`/`Competition`、`ResumeDiagnosisReport`、`ResumeDiagnosisIssue` |
 | **tool** | 无状态通用工具，不依赖数据库/Redis/用户上下文 | `file_parser.py`（PDF/Word/TXT 文件解析为纯文本）、`snowflake.py`（雪花ID生成） |
-| **services** | 业务逻辑实现 | `FuncPermissionService`（功能权限校验、扣减、回退）、`ResumeParseService`（简历文件解析→AI结构化→入库） |
+| **services** | 业务逻辑实现 | `FuncPermissionService`（功能权限校验、扣减、回退）、`ResumeParseService`（简历文件解析→AI结构化→入库）、`ResumeDiagnoseService`（简历诊断→AI并行分析→评级→入库） |
 ## 3️⃣ 技术栈
 | 类别 | 技术选型 | 说明 |
@@ -0,0 +1,89 @@
 """简历诊断 AI 引擎：并行诊断 + 汇总评价"""
 import asyncio
 import json
 import re
 from langchain_core.output_parsers import StrOutputParser
 from langchain_core.prompts import ChatPromptTemplate
 from app.ai.models import LLM
 from app.ai.resume_diagnoser.prompts import DIAGNOSE_MODULE_PROMPT, SUMMARY_PROMPT
 from app.core.logger import log
 def _parse_json(text: str) -> dict:
    """解析 AI 输出的 JSON，自动去除 markdown 代码块包裹，容错处理"""
    cleaned = re.sub(r"^```(?:json)?\s*\n?", "", text.strip())
    cleaned = re.sub(r"\n?```\s*$", "", cleaned)
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        # AI 可能在 JSON 字符串值中嵌入了未转义的引号，尝试提取最外层 { }
        match = re.search(r"\{[\s\S]*\}", cleaned)
        if match:
            return json.loads(match.group())
        raise
 # 诊断链（StrOutputParser 拿原始文本，再手动解析 JSON，避免 markdown 代码块导致解析失败）
 _diagnose_chain = (
    ChatPromptTemplate.from_messages([("system", DIAGNOSE_MODULE_PROMPT), ("human", "请开始诊断。")])
    | LLM.CLAUDE_SONNET_4.create(temperature=0)
    | StrOutputParser()
 )
 # 汇总评价链（纯文本输出）
 _summary_chain = (
    ChatPromptTemplate.from_messages([("system", SUMMARY_PROMPT), ("human", "请生成整体评价。")])
    | LLM.CLAUDE_SONNET_4.create(temperature=0.3)
    | StrOutputParser()
 )
 async def diagnose_all(tasks: list[dict]) -> list[dict]:
    """并行诊断所有模块记录
    tasks: [{"module_type": ..., "target_position": ..., "context": ..., "description_text": ...}, ...]
    返回: 与 tasks 一一对应的诊断结果列表
    """
    log.info(f"开始{len(tasks)}路并行AI诊断")
    results = await asyncio.gather(*[_safe_invoke(task) for task in tasks])
    log.info("并行AI诊断完成")
    return results
 async def generate_summary(grade: str, urgent_total: int, important_total: int,
                           expression_total: int, target_position: str, all_findings: str) -> str:
    """AI 生成整体评价文本"""
    inp = {
        "grade": grade, "urgent_total": str(urgent_total),
        "important_total": str(important_total), "expression_total": str(expression_total),
        "target_position": target_position or "未指定", "all_findings": all_findings,
    }
    try:
        return await _summary_chain.ainvoke(inp)
    except Exception as e:
        log.warning(f"AI生成整体评价失败: {e}")
        return "简历诊断已完成，请查看各模块的详细诊断结果。"
 async def _safe_invoke(task: dict) -> dict:
    """单条记录诊断，失败返回空结果"""
    raw = ""
    try:
        raw = await _diagnose_chain.ainvoke(task)
        return _parse_json(raw)
    except Exception as e:
        log.warning(f"AI诊断[{task.get('module_type', '')}]失败: {e}\n原始输出: {raw[:500]}")
        return _empty_result()
 def _empty_result() -> dict:
    return {
        "finding": "", "importance": "", "suggestion": "",
        "urgent_issues": {"typo": 0},
        "important_issues": {"no_result": 0, "no_quantify": 0, "weak_relevance": 0},
        "expression_issues": {"not_concise": 0, "format_inconsistent": 0},
        "optimized_content": None,
    }
@@ -0,0 +1,79 @@
 """简历诊断 Prompt 模板
 注意：prompt 中的 JSON 示例花括号必须用 {{ }} 转义，避免被 ChatPromptTemplate 当作变量。
 """
 DIAGNOSE_MODULE_PROMPT = """你是一位资深简历顾问和求职专家。请对以下简历模块的描述文本进行专业诊断。
 ## 模块信息
 - 模块类型：{module_type}
 - 目标岗位：{target_position}
 - 模块上下文：{context}
 ## 待诊断文本
 {description_text}
 ## 诊断维度
 ### 紧急修复
 - typo：错别字、语法错误、语病、标点符号使用错误、中英文标点混用、用词不当
 ### 重点优化
 - no_result：只描述了做了什么（任务/职责），但没有体现最终结果、产出或影响，像流水账
 - no_quantify：有成果描述但缺少具体数字支撑，使用了"大幅""显著""有效"等模糊表达，缺少人数、金额、百分比、时间等量化数据
 - weak_relevance：描述内容与目标岗位的核心职责关联度低，花大量篇幅描述与目标岗位无关的内容（注意：如果目标岗位为"未指定"，此项必须为0）
 ### 表达提升
 - not_concise：句子偏长信息密度低，存在赘词重复表达（如"进行了开发"应简化为"开发了"），使用空泛修饰词（"充分""积极""认真"）
 - format_inconsistent：时间格式、标点风格、数字写法、项目符号、人称使用不统一
 ## 输出要求
 严格输出以下JSON格式，每个问题类别的值为该类问题出现的次数（0表示无此问题）：
 ```json
 {{
  "finding": "用2-3句话概述发现的主要问题",
  "importance": "用1-2句话说明为什么这些问题对简历质量很重要",
  "suggestion": "给出具体可执行的改进建议",
  "urgent_issues": {{"typo": 0}},
  "important_issues": {{"no_result": 0, "no_quantify": 0, "weak_relevance": 0}},
  "expression_issues": {{"not_concise": 0, "format_inconsistent": 0}},
  "optimized_content": ["改写后的段落1", "改写后的段落2"]
 }}
 ```
 ## 关于 optimized_content 的格式要求
 - optimized_content 必须是一个纯文本字符串数组
 - 如果原文是 JSON 数组格式（如 [{{"id": "xxx", "text": "段落内容"}}]），则只提取每个元素的 text 内容进行改写，返回改写后的纯文本数组，段落数量必须与原文一一对应
 - 如果原文是纯文本（非JSON），则返回包含一个元素的数组：["改写后的完整文本"]
 - 如果原文没有明显问题，返回原文内容不做修改
 只输出JSON，不要输出其他内容。"""
 SUMMARY_PROMPT = """你是一位资深简历顾问。请根据以下简历诊断结果，生成一段整体评价。
 ## 诊断统计
 - 评级：{grade}
 - 紧急修复问题：{urgent_total} 个
 - 重点优化问题：{important_total} 个
 - 表达提升问题：{expression_total} 个
 ## 目标岗位
 {target_position}
 ## 各模块诊断发现
 {all_findings}
 ## 评级含义
 - A（优秀）：简历相当出彩，在求职市场中格外抢眼
 - B（良好）：简历已经很棒，但还有提升潜力
 - C（一般）：简历还有打磨空间，需要推敲细节
 - D（待提升）：简历有较大提升空间，需要尽快完善
 ## 输出要求
 请用3-5句话生成简历整体评价，包括：
 1. 背景概括（基于模块内容简要描述求职者背景）
 2. 优势总结（如果有值得肯定的地方）
 3. 主要问题（最需要改进的方面）
 4. 一句鼓励或行动建议
 直接输出评价文本，不要输出JSON或其他格式标记。控制在300字以内。"""
@@ -0,0 +1,76 @@
 """简历诊断接口"""
 from fastapi import APIRouter
 from pydantic import BaseModel, Field
 from app.ai.resume_diagnoser.diagnoser import diagnose_all, generate_summary
 from app.core.context import RequestContext
 from app.core.database import get_db
 from app.services.resume_diagnose_service import ResumeDiagnoseService, aggregate_results
 router = APIRouter(prefix="/resume/diagnose", tags=["简历诊断"])
 class DiagnoseParam(BaseModel):
    resume_id: int = Field(..., alias="resumeId")
 class ResolveParam(BaseModel):
    user_feedback: int = Field(..., alias="userFeedback")
@router.post("", summary="触发简历诊断")
 async def diagnose_resume(param: DiagnoseParam):
    """触发简历AI诊断，返回报告ID"""
    user_id = RequestContext.user_id.get()
    # 1. 短事务：加载简历数据
    async for session in get_db():
        service = ResumeDiagnoseService(session)
        resume, tasks = await service.load_resume_data(param.resume_id, user_id)
    if not tasks:
        raise ValueError("简历没有可诊断的描述内容")
    # 2. 并行 AI 诊断（不持有数据库连接）
    ai_tasks = [{k: v for k, v in t.items() if not k.startswith("_")} for t in tasks]
    ai_results = await diagnose_all(ai_tasks)
    # 3. 统计 + 评级（纯计算）
    stats = aggregate_results(tasks, ai_results)
    # 4. AI 生成整体评价（不持有数据库连接）
    summary = await generate_summary(
        grade=stats["grade"], urgent_total=stats["urgent_total"],
        important_total=stats["important_total"], expression_total=stats["expression_total"],
        target_position=resume.target_position or "", all_findings=stats["all_findings"],
    )
    # 5. 短事务：纯写入
    async for session in get_db():
        service = ResumeDiagnoseService(session)
        report_id = await service.save_report(
            param.resume_id, user_id, stats["grade"], summary,
            stats["urgent_total"], stats["important_total"], stats["expression_total"],
            tasks, ai_results,
        )
    return {"reportId": report_id}
@router.get("/{resume_id}", summary="查询最近一次诊断报告")
 async def get_diagnosis_report(resume_id: int):
    """查询指定简历的最近一次诊断报告 + 所有诊断问题"""
    user_id = RequestContext.user_id.get()
    async for session in get_db():
        service = ResumeDiagnoseService(session)
        return await service.get_latest_report(resume_id, user_id)
@router.put("/issue/{issue_id}/resolve", summary="标记问题已处理")
 async def resolve_issue(issue_id: int, param: ResolveParam):
    """标记诊断问题已处理 + 用户评价"""
    user_id = RequestContext.user_id.get()
    async for session in get_db():
        service = ResumeDiagnoseService(session)
        await service.resolve_issue(issue_id, user_id, param.user_feedback)
@@ -32,9 +32,11 @@ app.add_middleware(
 # ========== 路由注册 ==========
 from app.api.health import router as health_router
 from app.api.resume import router as resume_router
 from app.api.resume_diagnose import router as resume_diagnose_router
 app.include_router(health_router)
 app.include_router(resume_router)
 app.include_router(resume_diagnose_router)
 # ==============================
 if __name__ == "__main__":
@@ -0,0 +1,32 @@
 """简历诊断问题表（bg_resume_diagnosis_issue）"""
 from datetime import datetime
 from typing import Optional
 from sqlalchemy import BigInteger, Integer, String, Text, DateTime, JSON
 from sqlalchemy.orm import Mapped, mapped_column
 from app.core.database import Base
 class ResumeDiagnosisIssue(Base):
    """简历诊断问题表 bg_resume_diagnosis_issue"""
    __tablename__ = "bg_resume_diagnosis_issue"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    report_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="关联report.id")
    resume_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="关联bg_user_resume.id")
    user_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="用户ID")
    module_type: Mapped[str] = mapped_column(String(32), nullable=False, comment="模块类型: summary/education/work/internship/project/competition")
    module_record_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="模块记录ID，summary时为resume_id")
    finding: Mapped[Optional[str]] = mapped_column(Text, nullable=True, comment="诊断发现")
    importance: Mapped[Optional[str]] = mapped_column(Text, nullable=True, comment="为什么重要")
    suggestion: Mapped[Optional[str]] = mapped_column(Text, nullable=True, comment="改进建议")
    urgent_issues: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True, comment='紧急修复子类型计数 {"typo": 0}')
    important_issues: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True, comment='重点优化子类型计数 {"no_result": 0, "no_quantify": 0, "weak_relevance": 0}')
    expression_issues: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True, comment='表达提升子类型计数 {"not_concise": 0, "format_inconsistent": 0}')
    optimized_content: Mapped[Optional[list | str]] = mapped_column(JSON, nullable=True, comment="AI改写后的内容，子表模块与原description格式一致[{id,text}]保持原id只改写text，summary模块为纯文本字符串")
    status: Mapped[int] = mapped_column(Integer, default=0, comment="0=待处理 1=已处理")
    user_feedback: Mapped[int] = mapped_column(Integer, default=0, comment="0=未评价 1=符合 2=不符合")
    create_time: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")
@@ -0,0 +1,25 @@
 """简历诊断报告表（bg_resume_diagnosis_report）"""
 from datetime import datetime
 from typing import Optional
 from sqlalchemy import BigInteger, Integer, String, Text, DateTime
 from sqlalchemy.orm import Mapped, mapped_column
 from app.core.database import Base
 class ResumeDiagnosisReport(Base):
    """简历诊断报告表 bg_resume_diagnosis_report"""
    __tablename__ = "bg_resume_diagnosis_report"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    resume_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="关联bg_user_resume.id")
    user_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="用户ID")
    grade: Mapped[Optional[str]] = mapped_column(String(1), nullable=True, comment="评级 A/B/C/D")
    summary: Mapped[Optional[str]] = mapped_column(Text, nullable=True, comment="AI生成的整体评价")
    urgent_total: Mapped[int] = mapped_column(Integer, default=0, comment="紧急修复总数")
    important_total: Mapped[int] = mapped_column(Integer, default=0, comment="重点优化总数")
    expression_total: Mapped[int] = mapped_column(Integer, default=0, comment="表达提升总数")
    create_time: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")
@@ -0,0 +1,257 @@
 """简历诊断 Service
 加载简历描述数据 → 并行 AI 诊断 → 统计评级 → AI 汇总评价 → 写入数据库。
 依赖：resume_diagnoser（AI诊断引擎）
 使用表：bg_user_resume + 5张子表（读）、bg_resume_diagnosis_report + issue（写）
 """
 import json
 import shortuuid
 from sqlalchemy import select, desc
 from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.logger import log
 from app.models.resume_diagnosis_issue import ResumeDiagnosisIssue
 from app.models.resume_diagnosis_report import ResumeDiagnosisReport
 from app.models.user_resume import UserResume
 from app.models.user_resume_competition import UserResumeCompetition
 from app.models.user_resume_education import UserResumeEducation
 from app.models.user_resume_internship import UserResumeInternship
 from app.models.user_resume_project import UserResumeProject
 from app.models.user_resume_work import UserResumeWork
 from app.tool.snowflake import next_id
 # 模块中文名映射
 _MODULE_LABELS = {
    "summary": "个人概述", "education": "教育经历", "work": "工作经历",
    "internship": "实习经历", "project": "项目经历", "competition": "竞赛经历",
 }
 class ResumeDiagnoseService:
    def __init__(self, session: AsyncSession):
        self.session = session
    async def load_resume_data(self, resume_id: int, user_id: int) -> tuple[UserResume, list[dict]]:
        """加载简历主表 + 5 张子表数据，组装 AI 任务列表"""
        result = await self.session.execute(
            select(UserResume).where(UserResume.id == resume_id, UserResume.user_id == user_id))
        resume = result.scalar_one_or_none()
        if resume is None:
            raise ValueError("简历不存在")
        target_position = resume.target_position or ""
        tasks: list[dict] = []
        # summary
        if resume.summary and resume.summary.strip():
            tasks.append({
                "module_type": "个人概述", "target_position": target_position or "未指定",
                "context": f"姓名: {resume.name or '未填写'}",
                "description_text": resume.summary,
                "_module_type_key": "summary", "_module_record_id": resume_id,
            })
        # 子表
        await self._collect_tasks(tasks, target_position, "education", UserResumeEducation, resume_id,
                                  lambda r: f"学校: {r.school or ''}, 专业: {r.major or ''}, 学历: {r.degree or ''}")
        await self._collect_tasks(tasks, target_position, "work", UserResumeWork, resume_id,
                                  lambda r: f"公司: {r.company_name or ''}, 职位: {r.position or ''}")
        await self._collect_tasks(tasks, target_position, "internship", UserResumeInternship, resume_id,
                                  lambda r: f"公司: {r.company_name or ''}, 职位: {r.position or ''}")
        await self._collect_tasks(tasks, target_position, "project", UserResumeProject, resume_id,
                                  lambda r: f"公司: {r.company_name or ''}, 项目: {r.project_name or ''}, 角色: {r.role or ''}")
        await self._collect_tasks(tasks, target_position, "competition", UserResumeCompetition, resume_id,
                                  lambda r: f"竞赛: {r.competition_name or ''}, 获奖: {r.award or ''}")
        return resume, tasks
    async def _collect_tasks(self, tasks: list[dict], target_position: str,
                             module_type: str, model_cls, resume_id: int, context_fn):
        """查询子表记录，将有 description 的记录加入 tasks"""
        result = await self.session.execute(select(model_cls).where(model_cls.resume_id == resume_id))
        for record in result.scalars().all():
            desc_text = _build_description_text(record.description)
            if not desc_text:
                continue
            tasks.append({
                "module_type": _MODULE_LABELS[module_type],
                "target_position": target_position or "未指定",
                "context": context_fn(record),
                "description_text": desc_text,
                "_module_type_key": module_type, "_module_record_id": record.id,
                "_original_description": record.description,  # 原始 [{id,text}]，用于映射 optimized_content
            })
    async def save_report(self, resume_id: int, user_id: int, grade: str, summary: str,
                          urgent_total: int, important_total: int, expression_total: int,
                          tasks: list[dict], ai_results: list[dict]) -> int:
        """纯写入：接收已算好的 grade、summary、统计数据，写入 report + issues"""
        report_id = next_id()
        self.session.add(ResumeDiagnosisReport(
            id=report_id, resume_id=resume_id, user_id=user_id,
            grade=grade, summary=summary,
            urgent_total=urgent_total, important_total=important_total, expression_total=expression_total,
        ))
        for task, ai_result in zip(tasks, ai_results):
            if not _has_issues(ai_result):
                continue
            self.session.add(ResumeDiagnosisIssue(
                id=next_id(), report_id=report_id, resume_id=resume_id, user_id=user_id,
                module_type=task["_module_type_key"], module_record_id=task["_module_record_id"],
                finding=ai_result.get("finding", ""), importance=ai_result.get("importance", ""),
                suggestion=ai_result.get("suggestion", ""),
                urgent_issues=ai_result.get("urgent_issues"), important_issues=ai_result.get("important_issues"),
                expression_issues=ai_result.get("expression_issues"),
                optimized_content=_build_optimized_content(task, ai_result.get("optimized_content")),
                status=0, user_feedback=0,
            ))
        await self.session.flush()
        log.info(f"诊断报告保存完成 reportId:{report_id} grade:{grade}")
        return report_id
    async def get_latest_report(self, resume_id: int, user_id: int) -> dict | None:
        """查询最近一次诊断报告 + 所有 issues"""
        result = await self.session.execute(
            select(ResumeDiagnosisReport).where(
                ResumeDiagnosisReport.resume_id == resume_id, ResumeDiagnosisReport.user_id == user_id,
            ).order_by(desc(ResumeDiagnosisReport.create_time)).limit(1))
        report = result.scalar_one_or_none()
        if report is None:
            return None
        result = await self.session.execute(
            select(ResumeDiagnosisIssue).where(ResumeDiagnosisIssue.report_id == report.id))
        issues = result.scalars().all()
        return {
            "report": {
                "id": str(report.id), "resumeId": str(report.resume_id),
                "grade": report.grade, "summary": report.summary,
                "urgentTotal": report.urgent_total, "importantTotal": report.important_total,
                "expressionTotal": report.expression_total,
                "createTime": report.create_time.strftime("%Y-%m-%d %H:%M:%S") if report.create_time else None,
            },
            "issues": [_issue_to_dict(i) for i in issues],
        }
    async def resolve_issue(self, issue_id: int, user_id: int, user_feedback: int) -> None:
        """标记问题已处理 + 用户评价"""
        result = await self.session.execute(
            select(ResumeDiagnosisIssue).where(
                ResumeDiagnosisIssue.id == issue_id, ResumeDiagnosisIssue.user_id == user_id))
        issue = result.scalar_one_or_none()
        if issue is None:
            raise ValueError("诊断问题不存在")
        issue.status = 1
        issue.user_feedback = user_feedback
        await self.session.flush()
 # ===== 工具函数 =====
 def _build_optimized_content(task: dict, ai_texts: list[str] | None):
    """将 AI 返回的纯文本数组映射回存储格式
    - summary 模块：取第一个元素作为纯文本字符串
    - 子表模块：用原始 description 的 id + AI 改写的 text 组合成 [{id, text}]
    """
    if not ai_texts or not isinstance(ai_texts, list):
        return None
    original = task.get("_original_description")
    if original is None:
        # summary 模块，存纯文本
        return ai_texts[0] if ai_texts else None
    # 子表模块，映射回 [{id, text}]
    result = []
    for i, item in enumerate(original):
        if not isinstance(item, dict):
            continue
        text = ai_texts[i] if i < len(ai_texts) else item.get("text", "")
        result.append({"id": item.get("id"), "text": text})
    return result
 def _build_description_text(description: list[dict] | None) -> str:
    """子表 description [{id, text}] → JSON 字符串传给 AI（保留 id 以便 AI 返回同格式）"""
    if not description:
        return ""
    valid = [item for item in description if isinstance(item, dict) and item.get("text")]
    if not valid:
        return ""
    return json.dumps(valid, ensure_ascii=False)
 def aggregate_results(tasks: list[dict], ai_results: list[dict]) -> dict:
    """统计汇总 + 评级，返回 {grade, urgent_total, important_total, expression_total, has_weak_relevance, all_findings}"""
    urgent_total = 0
    important_total = 0
    expression_total = 0
    has_weak_relevance = False
    all_findings: list[str] = []
    for task, ai_result in zip(tasks, ai_results):
        urgent = ai_result.get("urgent_issues", {})
        important = ai_result.get("important_issues", {})
        expression = ai_result.get("expression_issues", {})
        urgent_total += sum(v for v in urgent.values() if isinstance(v, int))
        important_total += sum(v for v in important.values() if isinstance(v, int))
        expression_total += sum(v for v in expression.values() if isinstance(v, int))
        if important.get("weak_relevance", 0) > 0:
            has_weak_relevance = True
        finding = ai_result.get("finding", "")
        if finding and _has_issues(ai_result):
            label = _MODULE_LABELS.get(task["_module_type_key"], task["_module_type_key"])
            all_findings.append(f"【{label}】{finding}")
    grade = _calc_grade(urgent_total, important_total, expression_total, has_weak_relevance)
    return {
        "grade": grade, "urgent_total": urgent_total,
        "important_total": important_total, "expression_total": expression_total,
        "all_findings": "\n".join(all_findings),
    }
 def _calc_grade(urgent: int, important: int, expression: int, has_weak_relevance: bool) -> str:
    """评级硬算：D → C → B → A
    A：urgent=0, important<=1, expression<=1
    B：urgent=0, important<=3, expression<=2（且不满足A）
    C：urgent=1, 或 important 3-4
    D：urgent>=2, 或 (important>=4 且 has_weak_relevance)
    """
    if urgent >= 2 or (important >= 4 and has_weak_relevance):
        return "D"
    if urgent == 1 or 3 <= important <= 4:
        return "C"
    if urgent == 0 and important <= 1 and expression <= 1:
        return "A"
    if urgent == 0 and important <= 3 and expression <= 2:
        return "B"
    return "C"
 def _has_issues(ai_result: dict) -> bool:
    """判断诊断结果是否存在问题（所有计数都为 0 则无问题）"""
    for key in ("urgent_issues", "important_issues", "expression_issues"):
        counts = ai_result.get(key, {})
        if any(v > 0 for v in counts.values() if isinstance(v, int)):
            return True
    return False
 def _issue_to_dict(issue: ResumeDiagnosisIssue) -> dict:
    """ORM → API 响应字典"""
    return {
        "id": str(issue.id), "moduleType": issue.module_type,
        "moduleRecordId": str(issue.module_record_id),
        "finding": issue.finding, "importance": issue.importance, "suggestion": issue.suggestion,
        "urgentIssues": issue.urgent_issues, "importantIssues": issue.important_issues,
        "expressionIssues": issue.expression_issues, "optimizedContent": issue.optimized_content,
        "status": issue.status, "userFeedback": issue.user_feedback,
    }