feat: add cache prune-scores to drop outdated match results

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 
+### New Features
+
+- **`jobradar cache prune-scores`** (`cli.py` / `cache.py`)
+  Drops match results that no longer reflect the current scoring setup, so the next search or `assess` recomputes them. By default it removes rows whose `prompt_version` differs from the current one — useful right after bumping `matching.PROMPT_VERSION`. `--stale-cv` also drops rows belonging to any CV other than the latest, and `--orphans` drops rows whose job is gone from `job_cache`. The command previews the counts and asks for confirmation before deleting (`--yes` skips the prompt). Categories can overlap, so each is reported separately alongside a de-duplicated total.
+
 ### Changes
 
 - **Scores now come from `job_matches` alone** (`schemas.py` / `display.py` / `cli.py` / `search_assessment_stage.py` / `search_prefilter.py` / `cache.py`)

README.es.md

@@ -33,6 +33,7 @@ uv run jobradar find cv.docx  # Modo CLI
 | `uv run jobradar assess` | Reejecuta la evaluación LLM sobre JDs en caché |
 | `uv run jobradar model` | Selecciona interactivamente el proveedor y modelo LLM |
 | `uv run jobradar cache clear` | Limpia toda la caché |
+| `uv run jobradar cache prune-scores` | Elimina resultados de coincidencia de una versión de prompt o CV obsoletos |
 | `uv run jobradar --version` | Muestra la versión actual |
 
 ## Visión General del Pipeline

README.md

@@ -33,6 +33,7 @@ uv run jobradar find cv.docx  # CLI mode
 | `uv run jobradar assess` | Re-run LLM assessment on cached JDs |
 | `uv run jobradar model` | Interactively choose LLM provider and model |
 | `uv run jobradar cache clear` | Clear all caches |
+| `uv run jobradar cache prune-scores` | Drop match results from an outdated prompt version or CV |
 | `uv run jobradar --version` | Show current version |
 
 ## Pipeline Overview

README.zh.md

@@ -33,6 +33,7 @@ uv run jobradar find cv.docx  # CLI 模式
 | `uv run jobradar assess` | 对缓存 JD 单独补跑 LLM 评估 |
 | `uv run jobradar model` | 交互式选择 LLM provider 和模型 |
 | `uv run jobradar cache clear` | 清空所有缓存 |
+| `uv run jobradar cache prune-scores` | 删除过时 prompt 版本或旧 CV 的匹配结果 |
 | `uv run jobradar --version` | 显示当前版本号 |
 
 ## Pipeline 概览

jobradar/cache.py

@@ -1464,6 +1464,66 @@ def clear_search_stats() -> None:
         con.execute("DELETE FROM filter_events")
 
 
+def _stale_match_conditions(
+    prompt_version: str, keep_cv_hash: str, drop_orphans: bool
+) -> tuple[list[str], list[str]]:
+    """构造 job_matches 的过时行判定条件，返回 (SQL 片段, 参数)。"""
+    clauses: list[str] = []
+    params: list[str] = []
+    if prompt_version:
+        clauses.append("prompt_version != ?")
+        params.append(prompt_version)
+    if keep_cv_hash:
+        clauses.append("cv_hash != ?")
+        params.append(keep_cv_hash)
+    if drop_orphans:
+        clauses.append("job_id NOT IN (SELECT dedup_key FROM job_cache)")
+    return clauses, params
+
+
+def prune_job_matches(
+    *,
+    prompt_version: str = "",
+    keep_cv_hash: str = "",
+    drop_orphans: bool = False,
+    dry_run: bool = True,
+) -> dict[str, int]:
+    """删除过时的匹配结果，被删除的职位会在下次搜索或 assess 时按当前口径重算。
+
+    三类条件独立启用，彼此可重叠（同一行可能既是旧 prompt 版本又属于旧 CV），
+    因此各分类计数之和可能大于 ``total``；``total`` 才是唯一行数。
+    ``dry_run=True`` 时只统计不删除。
+    """
+    clauses, params = _stale_match_conditions(prompt_version, keep_cv_hash, drop_orphans)
+    result = {"stale_version": 0, "stale_cv": 0, "orphan": 0, "total": 0, "deleted": 0}
+    if not clauses:
+        return result
+
+    with _conn() as con:
+        if prompt_version:
+            result["stale_version"] = con.execute(
+                "SELECT count(*) FROM job_matches WHERE prompt_version != ?", (prompt_version,)
+            ).fetchone()[0]
+        if keep_cv_hash:
+            result["stale_cv"] = con.execute(
+                "SELECT count(*) FROM job_matches WHERE cv_hash != ?", (keep_cv_hash,)
+            ).fetchone()[0]
+        if drop_orphans:
+            result["orphan"] = con.execute(
+                "SELECT count(*) FROM job_matches WHERE job_id NOT IN (SELECT dedup_key FROM job_cache)"
+            ).fetchone()[0]
+
+        where = " OR ".join(clauses)
+        result["total"] = con.execute(
+            f"SELECT count(*) FROM job_matches WHERE {where}", params
+        ).fetchone()[0]
+        if not dry_run:
+            result["deleted"] = con.execute(
+                f"DELETE FROM job_matches WHERE {where}", params
+            ).rowcount
+    return result
+
+
 def clean_expired() -> int:
     """删除过期 JD 和 Session，返回删除条数。"""
     now = datetime.utcnow().isoformat()

jobradar/cli.py

@@ -558,6 +558,54 @@ def cache_clean() -> None:
     console.print(f"[green]已清理 {count} 条过期记录。[/green]")
 
 
+@cache_app.command("prune-scores")
+def cache_prune_scores(
+    stale_cv: Annotated[bool, typer.Option("--stale-cv", help="同时删除非当前 CV 的匹配结果")] = False,
+    orphans: Annotated[bool, typer.Option("--orphans", help="同时删除职位已不存在的匹配结果")] = False,
+    language: Annotated[str, typer.Option("--language", help="匹配 prompt 使用的语言")] = "zh",
+    yes: Annotated[bool, typer.Option("--yes", "-y", help="跳过确认")] = False,
+) -> None:
+    """删除过时的匹配结果。被删除的职位会在下次搜索或 assess 时按当前口径重算。"""
+    from jobradar.matching import match_prompt_version
+
+    current_version = match_prompt_version(language)
+    keep_cv_hash = cache.get_latest_cv_hash() if stale_cv else ""
+    if stale_cv and not keep_cv_hash:
+        console.print("[red]缓存中没有 CVProfile，无法确定当前 CV，--stale-cv 不可用。[/red]")
+        raise typer.Exit(1)
+
+    preview = cache.prune_job_matches(
+        prompt_version=current_version,
+        keep_cv_hash=keep_cv_hash,
+        drop_orphans=orphans,
+        dry_run=True,
+    )
+    if preview["total"] == 0:
+        console.print("[dim]没有需要清理的匹配结果。[/dim]")
+        return
+
+    console.print(f"\n[bold]将删除 {preview['total']} 条匹配结果：[/bold]")
+    console.print(f"  prompt 版本不是 {current_version}：{preview['stale_version']} 条")
+    if stale_cv:
+        console.print(f"  不属于当前 CV（{keep_cv_hash[:8]}）：{preview['stale_cv']} 条")
+    if orphans:
+        console.print(f"  职位已不存在：{preview['orphan']} 条")
+    console.print("[dim]（分类可能重叠，总数为去重后的行数）[/dim]")
+
+    if not yes and not prompt_confirm("确认删除？被删除的职位需要重新评估"):
+        console.print("[yellow]已取消。[/yellow]")
+        return
+
+    result = cache.prune_job_matches(
+        prompt_version=current_version,
+        keep_cv_hash=keep_cv_hash,
+        drop_orphans=orphans,
+        dry_run=False,
+    )
+    console.print(f"[green]已删除 {result['deleted']} 条匹配结果。[/green]")
+    console.print("[dim]运行 `jobradar assess` 可按当前口径补算。[/dim]")
+
+
 # ─── serve 命令 ──────────────────────────────────────────────────────────────
 
 

tests/test_cache.py

@@ -645,3 +645,102 @@ def fake_evaluate(job, profile, llm, cv_hash="", language="zh"):
         assert (succeeded, failed) == (3, 0)
         # 第二次调用必须为空，否则 assess 会陷入每次重评同一批职位的循环。
         assert temp_db.get_unassessed_jobs() == []
+
+
+class TestPruneJobMatches:
+    def _save(self, temp_db, job_id: str, cv_hash: str, prompt_version: str) -> None:
+        match = MatchScore(
+            job_id=job_id,
+            cv_hash=cv_hash,
+            overall_score=70,
+            title_score=70,
+            seniority_score=70,
+            must_have_score=70,
+            nice_to_have_score=70,
+            domain_score=70,
+            location_score=100,
+            language_score=100,
+            risk_penalty=0,
+            recommendation="apply",
+        )
+        temp_db.save_job_match(match, "desc", prompt_version=prompt_version)
+
+    def _remaining(self, temp_db) -> set[tuple[str, str]]:
+        with temp_db._conn() as con:
+            return {(r["job_id"], r["prompt_version"]) for r in con.execute("SELECT * FROM job_matches")}
+
+    def test_dry_run_reports_without_deleting(self, temp_db):
+        job = make_job(description_snippet="desc")
+        temp_db.save_job(job)
+        self._save(temp_db, job.dedup_key, "cv1", "match_v10:zh")
+
+        result = temp_db.prune_job_matches(prompt_version="match_v11:zh", dry_run=True)
+
+        assert result["stale_version"] == 1
+        assert result["total"] == 1
+        assert result["deleted"] == 0
+        assert len(self._remaining(temp_db)) == 1
+
+    def test_deletes_only_stale_prompt_versions(self, temp_db):
+        job = make_job(description_snippet="desc")
+        other = make_job(company="Other", url="http://example.com/2", description_snippet="desc")
+        temp_db.save_job(job)
+        temp_db.save_job(other)
+        self._save(temp_db, job.dedup_key, "cv1", "match_v10:zh")
+        self._save(temp_db, other.dedup_key, "cv1", "match_v11:zh")
+
+        result = temp_db.prune_job_matches(prompt_version="match_v11:zh", dry_run=False)
+
+        assert result["deleted"] == 1
+        assert self._remaining(temp_db) == {(other.dedup_key, "match_v11:zh")}
+
+    def test_stale_cv_hash_is_optional(self, temp_db):
+        job = make_job(description_snippet="desc")
+        temp_db.save_job(job)
+        self._save(temp_db, job.dedup_key, "cv-old", "match_v11:zh")
+
+        untouched = temp_db.prune_job_matches(prompt_version="match_v11:zh", dry_run=True)
+        assert untouched["total"] == 0
+
+        targeted = temp_db.prune_job_matches(
+            prompt_version="match_v11:zh", keep_cv_hash="cv-current", dry_run=True
+        )
+        assert targeted["stale_cv"] == 1
+        assert targeted["total"] == 1
+
+    def test_orphan_rows_removed_only_when_requested(self, temp_db):
+        self._save(temp_db, "ghost|role", "cv1", "match_v11:zh")
+
+        assert temp_db.prune_job_matches(prompt_version="match_v11:zh", dry_run=True)["total"] == 0
+
+        result = temp_db.prune_job_matches(
+            prompt_version="match_v11:zh", drop_orphans=True, dry_run=False
+        )
+        assert result["orphan"] == 1
+        assert result["deleted"] == 1
+        assert self._remaining(temp_db) == set()
+
+    def test_overlapping_rows_counted_once_in_total(self, temp_db):
+        """同一行同时命中多个条件时，total 不应重复计数。"""
+        job = make_job(description_snippet="desc")
+        temp_db.save_job(job)
+        self._save(temp_db, job.dedup_key, "cv-old", "match_v10:zh")
+
+        result = temp_db.prune_job_matches(
+            prompt_version="match_v11:zh", keep_cv_hash="cv-current", dry_run=False
+        )
+
+        assert result["stale_version"] == 1
+        assert result["stale_cv"] == 1
+        assert result["total"] == 1
+        assert result["deleted"] == 1
+
+    def test_no_conditions_is_a_noop(self, temp_db):
+        job = make_job(description_snippet="desc")
+        temp_db.save_job(job)
+        self._save(temp_db, job.dedup_key, "cv1", "match_v11:zh")
+
+        result = temp_db.prune_job_matches(dry_run=False)
+
+        assert result["total"] == 0 and result["deleted"] == 0
+        assert len(self._remaining(temp_db)) == 1