diff --git a/CHANGELOG.md b/CHANGELOG.md index 69c66f1..1a2112f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## [Unreleased] +### Bug Fixes + +- **`assess` skipped every job that carried a legacy assessment** (`cache.py` / `cli.py` / `search_assessment_stage.py`) + `get_unassessed_jobs` selected on `assessment IS NULL` before consulting `job_matches`. Since the legacy `assessment` column records no `cv_hash`, any job scored under an earlier CV was treated as done forever — on a local cache of 293 jobs the command reported "all assessed" while 205 had no match under the current CV, and 95 rejected by a previous CV could never be reconsidered. Selection is now driven solely by the presence of a match for the current `cv_hash`. + Its filter predicate also read `job.match_score` — a leftover loop variable — instead of `j.match_score`, so the whole batch was kept or dropped according to the last row alone. + `assess` now writes `job_matches` through the shared evaluation path (`evaluate_cached_jobs`) rather than only refreshing the legacy column, which keeps repeated runs idempotent: previously the new selection rule would have re-queued the same jobs on every invocation. + ### Documentation - **Release highlights in Chinese, English and Spanish** (`docs/release-highlights.*.md` / `README*.md`) diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 22f27fe..1718e0e 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -294,7 +294,7 @@ jobradar_email_sync_runs_total{status="failed",reason="auth"} | Indeed 抓取结果为 0 | JobSpy 被 Indeed 限流（常见，非故障） | 等几小时重试；查日志中 JobSpy WARNING | | Adzuna 返回 429 | 速率限制 | 调大 `_MIN_INTERVAL`（当前 1.2s）；核对 `.env` 中 `ADZUNA_APP_ID` / `ADZUNA_APP_KEY` | | LLM 评估全部拒绝 | `cv_summary` / `cv_skills` 提取失败 | 检查 CV 解析结果；`uv run jobradar assess` 补跑评估 | -| 职位无模型评分 | 来自旧缓存（assessment 为 NULL） | `uv run jobradar assess` | +| 职位无模型评分 | 当前 `cv_hash` 下尚无 `job_matches` 记录（换 CV 后常见） | `uv run jobradar assess` 为当前 CV 补算 | ### 安全注意 diff --git a/jobradar/cache.py b/jobradar/cache.py index 0afbd00..2098fbe 100644 --- a/jobradar/cache.py +++ b/jobradar/cache.py @@ -981,16 +981,27 @@ def update_job_assessment(dedup_key: str, assessment: JobAssessment) -> None: def get_unassessed_jobs(limit: int = 200) -> list[JobResult]: - """返回尚无现代匹配结果且缺少 legacy assessment 的未过期职位。""" + """返回当前 CV 下尚无匹配结果的未过期职位。 + + 只按现代 ``job_matches`` 判定：legacy ``assessment`` 列没有 cv_hash 归属， + 用它筛选会把换 CV 后本该重评的职位挡在门外。 + """ with _conn() as con: - rows = con.execute( - "SELECT * FROM job_cache WHERE assessment IS NULL ORDER BY fetched_at DESC LIMIT ?", - (limit,), - ).fetchall() - jobs = [_row_to_job(r) for r in rows] - for job in jobs: + rows = con.execute("SELECT * FROM job_cache ORDER BY fetched_at DESC").fetchall() + + unassessed: list[JobResult] = [] + for row in rows: + job = _row_to_job(row) + if job.is_expired: + continue + # 先过滤过期再挂载 match，避免为已过期职位做多余的查询。 _attach_latest_match(job) - return [j for j in jobs if not j.is_expired and job.match_score is None] + if job.match_score is not None: + continue + unassessed.append(job) + if len(unassessed) >= limit: + break + return unassessed # ─── 流式搜索候选缓存 ───────────────────────────────────────────────────────── diff --git a/jobradar/cli.py b/jobradar/cli.py index aac6b4d..367b43b 100644 --- a/jobradar/cli.py +++ b/jobradar/cli.py @@ -1,6 +1,7 @@ """CLI 入口：Typer 命令定义。""" from __future__ import annotations +import hashlib import json import os import webbrowser @@ -14,7 +15,7 @@ from jobradar import cache from jobradar.agent import run_search -from jobradar.assessment import batch_assess_jds +from jobradar.assessment import gate_worker_count from jobradar.cv_extractor import extract_cv_profile from jobradar.cv_reader import read_cv from jobradar.display import ( @@ -43,6 +44,7 @@ get_saved_defaults, save_env_key, ) +from jobradar.search_assessment_stage import evaluate_cached_jobs from jobradar.telemetry import telemetry from jobradar.tools import verify_job_active @@ -447,11 +449,13 @@ def assess( llm = LLMConfig(provider=effective_provider, model=effective_model) console.print(f"[dim]使用模型：{llm.provider} / {llm.model}[/dim]") - # Step 1：获取 CVProfile + # Step 1：获取 CVProfile 及其 cv_hash（匹配结果按 cv_hash 归属，必须一并确定） profile = None + cv_hash = "" if cv_path is not None: try: cv_text = read_cv(cv_path) + cv_hash = hashlib.sha256(cv_text.encode()).hexdigest() with Progress(SpinnerColumn(), TextColumn("{task.description}"), transient=True) as p: p.add_task("解析 CV 信息...", total=None) profile = extract_cv_profile(cv_text, llm=llm) @@ -463,26 +467,34 @@ def assess( if profile is None: console.print("[red]缓存中没有 CVProfile，请提供 CV 文件路径。[/red]") raise typer.Exit(1) + cv_hash = cache.get_latest_cv_hash() console.print(f"[dim]使用缓存的 CVProfile：{profile.summary[:40]}（{profile.seniority_display}）[/dim]") - # Step 2：加载未评估的 JD 并批量评估 + if not cv_hash: + console.print("[red]无法确定 CV 版本（cv_hash），请提供 CV 文件路径。[/red]") + raise typer.Exit(1) + + # Step 2：为当前 CV 下尚无匹配结果的 JD 补算评分 unassessed = cache.get_unassessed_jobs(limit=limit) if unassessed: - console.print(f"\n[bold]待评估 JD：{len(unassessed)} 条[/bold]") - - job_inputs = [(j.title, j.description_snippet or "") for j in unassessed] + console.print(f"\n[bold]待评估 JD：{len(unassessed)} 条[/bold]（CV 版本 {cv_hash[:8]}）") with Progress(SpinnerColumn(), TextColumn("{task.description}"), transient=True) as p: - p.add_task(f"批量评估（每批 8 条，共 {len(unassessed)} 条）...", total=None) - with telemetry.timer("JD 批量评估"): - assessments = batch_assess_jds(job_inputs, profile, llm) - - for job, jda in zip(unassessed, assessments): - cache.update_job_assessment(job.dedup_key, jda.to_job_assessment()) + p.add_task(f"评估中（共 {len(unassessed)} 条）...", total=None) + with telemetry.timer("JD 评估"): + succeeded, failed = evaluate_cached_jobs( + unassessed, + profile=profile, + llm=llm, + cv_hash=cv_hash, + workers=gate_worker_count(llm.provider), + ) - console.print(f"[green]已更新 {len(unassessed)} 条评估结果。[/green]") + console.print(f"[green]已更新 {succeeded} 条评估结果。[/green]") + if failed: + console.print(f"[yellow]{failed} 条评估失败，详见 logs/jobradar.log。[/yellow]") else: - console.print("[dim]所有 JD 已有评估，跳过评估步骤。[/dim]") + console.print("[dim]当前 CV 下所有 JD 均已评估，跳过评估步骤。[/dim]") # Step 3：加载全部已评估 JD 排序展示 all_jobs = cache.get_recent_jobs(limit) diff --git a/jobradar/search_assessment_stage.py b/jobradar/search_assessment_stage.py index 5688399..7067d00 100644 --- a/jobradar/search_assessment_stage.py +++ b/jobradar/search_assessment_stage.py @@ -187,6 +187,51 @@ def _evaluate_jobs( return completed +def evaluate_cached_jobs( + jobs: list[JobResult], + *, + profile: CVProfile, + llm, + cv_hash: str, + language: str = "zh", + workers: int = 1, +) -> tuple[int, int]: + """为已缓存职位补算 JD profile 与 CV 匹配，返回 (成功, 失败) 条数。 + + ``assess`` 命令用此入口补全 ``job_matches``。写入 legacy ``assessment`` 列是不够的： + 待评估判定看的是当前 cv_hash 下有无匹配结果，只写旧列会让同一批职位被反复捞出。 + """ + if not jobs: + return 0, 0 + tasks = [ + JobEvaluationTask( + key=job.dedup_key, + job=job, + full_jd=job.description_snippet, + kind="assess", + source=job.sources[0] if job.sources else "unknown", + ) + for job in jobs + ] + metrics = AssessmentConcurrencyMetrics(workers=workers) + results = _evaluate_jobs( + tasks, + profile=profile, + llm=llm, + cv_hash=cv_hash, + language=language, + executor=None, + workers=workers, + metrics=metrics, + ) + failed = 0 + for task, error in results: + if error is not None: + failed += 1 + logger.warning("Assessment failed for %s: %s", task.key, error) + return len(results) - failed, failed + + def flush_assessments( pf: PrefilterResult, job_all_sources: dict[str, list[dict]], diff --git a/tests/test_cache.py b/tests/test_cache.py index 1e3fa32..09a86c4 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -11,8 +11,10 @@ CoarseFilterResult, CoverLetter, CVOptimization, + CVProfile, InterviewPrep, JDProfile, + JobAssessment, JobResult, JobSummary, MatchScore, @@ -477,3 +479,169 @@ def test_delete_jobs_removes_cv_optimization(self, temp_db): temp_db.delete_jobs([job.dedup_key]) assert temp_db.get_cv_optimization(job.dedup_key, "cv123", "desc") is None + + +class TestUnassessedJobs: + """get_unassessed_jobs 只按现代 job_matches 判定，不看 legacy assessment 列。""" + + CV_HASH = "cv-current" + + def _seed_cv(self, temp_db): + temp_db.save_cv_profile( + self.CV_HASH, + CVProfile(summary="Backend engineer", skills=["Python"]), + ) + + def _save_match(self, temp_db, job, cv_hash=None): + from jobradar.matching import match_prompt_version + + match = MatchScore( + job_id=job.dedup_key, + cv_hash=cv_hash or self.CV_HASH, + overall_score=82, + title_score=80, + seniority_score=70, + must_have_score=90, + nice_to_have_score=60, + domain_score=75, + location_score=85, + language_score=100, + risk_penalty=3, + recommendation="apply", + ) + temp_db.save_job_match( + match, + job.description_snippet, + prompt_version=match_prompt_version("zh"), + ) + + def test_legacy_assessment_does_not_hide_unmatched_job(self, temp_db): + """核心修复：旧评分存在不代表当前 CV 评过，该职位仍需重评。""" + self._seed_cv(temp_db) + job = make_job( + description_snippet="Build Python APIs.", + assessment=JobAssessment(score=7, is_relevant=True), + ) + temp_db.save_job(job) + + result = temp_db.get_unassessed_jobs() + assert [j.dedup_key for j in result] == [job.dedup_key] + + def test_rejected_legacy_assessment_still_returned(self, temp_db): + """旧 CV 判定的 is_relevant=False 不应永久排除该职位。""" + self._seed_cv(temp_db) + job = make_job( + description_snippet="Build Python APIs.", + assessment=JobAssessment(score=2, is_relevant=False), + ) + temp_db.save_job(job) + + assert [j.dedup_key for j in temp_db.get_unassessed_jobs()] == [job.dedup_key] + + def test_job_matched_under_current_cv_is_excluded(self, temp_db): + self._seed_cv(temp_db) + job = make_job(description_snippet="Build Python APIs.") + temp_db.save_job(job) + self._save_match(temp_db, job) + + assert temp_db.get_unassessed_jobs() == [] + + def test_match_from_another_cv_does_not_count(self, temp_db): + """换 CV 后旧 cv_hash 的匹配结果不算已评估。""" + self._seed_cv(temp_db) + job = make_job(description_snippet="Build Python APIs.") + temp_db.save_job(job) + self._save_match(temp_db, job, cv_hash="cv-outdated") + + assert [j.dedup_key for j in temp_db.get_unassessed_jobs()] == [job.dedup_key] + + def test_each_job_filtered_independently(self, temp_db): + """回归：过滤条件曾误用循环残留变量，导致整批结果由最后一条决定。""" + self._seed_cv(temp_db) + unmatched = [ + make_job(company=f"Company{i}", url=f"http://example.com/{i}", description_snippet="desc") + for i in range(3) + ] + for job in unmatched: + temp_db.save_job(job) + + # 最后写入的职位已有当前 CV 的匹配结果，不应影响前面三条的判定。 + matched = make_job(company="Matched", url="http://example.com/matched", description_snippet="desc") + temp_db.save_job(matched) + self._save_match(temp_db, matched) + + result = {j.dedup_key for j in temp_db.get_unassessed_jobs()} + assert result == {j.dedup_key for j in unmatched} + + def test_expired_jobs_excluded(self, temp_db): + self._seed_cv(temp_db) + expired = make_job( + company="Expired", + url="http://example.com/expired", + description_snippet="desc", + expires_at=datetime.utcnow() - timedelta(days=1), + ) + live = make_job(company="Live", url="http://example.com/live", description_snippet="desc") + temp_db.save_job(expired) + temp_db.save_job(live) + + assert [j.dedup_key for j in temp_db.get_unassessed_jobs()] == [live.dedup_key] + + def test_limit_counts_returned_jobs(self, temp_db): + """limit 作用于实际返回条数，而非过滤前的扫描量。""" + self._seed_cv(temp_db) + for i in range(5): + temp_db.save_job( + make_job(company=f"Company{i}", url=f"http://example.com/{i}", description_snippet="desc") + ) + + assert len(temp_db.get_unassessed_jobs(limit=2)) == 2 + + def test_assess_cycle_is_idempotent(self, temp_db, monkeypatch): + """回归：补跑评估必须写入 job_matches，否则同一批职位会被反复捞出。""" + from jobradar import search_assessment_stage as stage + from jobradar.llm_backend import LLMConfig + + self._seed_cv(temp_db) + for i in range(3): + temp_db.save_job( + make_job( + company=f"Company{i}", + url=f"http://example.com/{i}", + description_snippet="Build Python APIs.", + ) + ) + + def fake_evaluate(job, profile, llm, cv_hash="", language="zh"): + jd_profile = JDProfile(job_id=job.dedup_key, title=job.title, company=job.company) + match = MatchScore( + job_id=job.dedup_key, + cv_hash=cv_hash, + overall_score=80, + title_score=80, + seniority_score=80, + must_have_score=80, + nice_to_have_score=80, + domain_score=80, + location_score=100, + language_score=100, + risk_penalty=0, + recommendation="apply", + ) + return jd_profile, match + + monkeypatch.setattr(stage, "evaluate_job_once", fake_evaluate) + + pending = temp_db.get_unassessed_jobs() + assert len(pending) == 3 + + succeeded, failed = stage.evaluate_cached_jobs( + pending, + profile=CVProfile(summary="Backend engineer", skills=["Python"]), + llm=LLMConfig(provider="gemini", model="test-model"), + cv_hash=self.CV_HASH, + ) + + assert (succeeded, failed) == (3, 0) + # 第二次调用必须为空，否则 assess 会陷入每次重评同一批职位的循环。 + assert temp_db.get_unassessed_jobs() == []