fix: deduplicate job source aliases

BUGFIX_LOG.md

@@ -7,6 +7,22 @@
 - **结果**：修复后验证情况
 
 ---
+## 2026-08-11
+
+### BUG · 切换 CV 后职位卡片出现重复 Indeed 标签
+
+**错误**
+缓存职位在不同 CV 下重新评估后，同一 Indeed 链接会同时保存为 `indeed.ie` 和 `ie.indeed.com`，前端因此渲染两个相同的 Indeed 标签。
+
+**原因**
+缓存重评路径调用 `write_cache` 时没有传递已有的 `sources`、`raw_sources` 和发布时间，写入层便从 URL 域名推导出新的来源名称。缓存合并又只按来源字符串去重，没有识别两个域名属于同一职位平台。
+
+**解决方案**
+重评时保留完整来源元数据；缓存插入、合并和读取统一按平台身份折叠来源别名。历史缓存无需重新抓取或重新评分，读取时即可把重复 Indeed 来源合并，同时保留独立的 LinkedIn 来源。
+
+**结果**
+新增四个回归测试，覆盖来源别名合并、流式来源补全、历史脏数据读取以及换 CV 重评。Ruff、Python 编译、前端静态检查、diff 检查和完整 pytest（245 tests）均通过；正式缓存中的 Foxit 记录只读验证为一个 Indeed 和一个 LinkedIn 来源。
+
 ## 2026-08-04
 
 ### BUG · 缓存职位缺少现代评分拆解

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 
+### Bug Fixes
+
+- **Duplicate Indeed source badges after CV changes** (`search_assessment_stage.py` / `cache.py`)
+  Re-assessing a cached job for a different CV now preserves its original `sources`, `raw_sources`, and posting date instead of deriving a second source name from the URL. Cache reads and merges also collapse hostname aliases from the same provider, so existing rows containing both `indeed.ie` and `ie.indeed.com` render as one Indeed source while retaining distinct LinkedIn sources.
+
 ## [0.5.0] — 2026-08-07
 
 Scoring became CV-aware. A job's match score is now owned by the CV that produced

jobradar/cache.py

@@ -10,6 +10,7 @@
 from contextlib import contextmanager
 from datetime import datetime
 from pathlib import Path
+from urllib.parse import urlparse
 
 from jobradar import artifact_store
 from jobradar.paths import DATA_DIR, ensure_parent
@@ -276,6 +277,72 @@ def _conn():
 # ─── JobResult ────────────────────────────────────────────────────────────────
 
 
+def _source_provider_key(source: str, url: str = "") -> str:
+    source_name = str(source or "").strip()
+    try:
+        host = urlparse(url).netloc.removeprefix("www.").lower()
+    except ValueError:
+        host = ""
+    candidates = (source_name.lower(), host)
+    if any("linkedin" in candidate for candidate in candidates):
+        return "linkedin"
+    if any("indeed" in candidate for candidate in candidates):
+        return "indeed"
+    return source_name.casefold() or host or "unknown"
+
+
+def _source_display_name(source: str, url: str = "") -> str:
+    source_name = str(source or "").strip()
+    if source_name and source_name.lower() != "unknown":
+        return source_name
+    provider = _source_provider_key(source_name, url)
+    if provider == "linkedin":
+        return "linkedin.com"
+    if provider == "indeed":
+        return "indeed.ie"
+    return source_name or "unknown"
+
+
+def _deduplicate_job_sources(
+    sources: list[str] | list[dict],
+    raw_sources: list[dict],
+) -> tuple[list[str], list[dict]]:
+    """Collapse hostname aliases that represent the same job platform."""
+    names_by_provider: dict[str, str] = {}
+    for source in sources:
+        if isinstance(source, dict):
+            source_name = str(source.get("source") or "")
+            source_url = str(source.get("url") or "")
+        else:
+            source_name = str(source or "")
+            source_url = ""
+        provider = _source_provider_key(source_name, source_url)
+        names_by_provider.setdefault(provider, _source_display_name(source_name, source_url))
+
+    entries_by_provider: dict[str, dict] = {}
+    for raw_source in raw_sources:
+        if not isinstance(raw_source, dict):
+            continue
+        source_name = str(raw_source.get("source") or "")
+        source_url = str(raw_source.get("url") or "")
+        provider = _source_provider_key(source_name, source_url)
+        display_name = names_by_provider.setdefault(
+            provider,
+            _source_display_name(source_name, source_url),
+        )
+        entry = dict(raw_source)
+        entry["source"] = display_name
+        existing = entries_by_provider.get(provider)
+        if existing is None:
+            entries_by_provider[provider] = entry
+            continue
+        for field in ("url", "date_posted"):
+            if not existing.get(field) and entry.get(field):
+                existing[field] = entry[field]
+
+    return list(names_by_provider.values()), list(entries_by_provider.values())
+
+
 def get_job(dedup_key: str, language: str = "zh") -> JobResult | None:
     with _conn() as con:
         row = con.execute(
@@ -304,6 +371,7 @@ def save_job(job: JobResult) -> None:
 
 
 def _insert_job(job: JobResult) -> None:
+    sources, raw_sources = _deduplicate_job_sources(job.sources, job.raw_sources)
     with _conn() as con:
         con.execute(
             """
@@ -319,8 +387,8 @@ def _insert_job(job: JobResult) -> None:
                 job.location,
                 job.description_snippet,
                 job.url,
-                json.dumps(job.sources),
-                json.dumps(job.raw_sources),
+                json.dumps(sources),
+                json.dumps(raw_sources),
                 job.date_posted,
                 job.fetched_at.isoformat(),
                 job.expires_at.isoformat() if job.expires_at else None,
@@ -333,10 +401,10 @@ def _insert_job(job: JobResult) -> None:
 
 def _merge_job(existing: JobResult, new: JobResult) -> None:
     """追加新来源；若新记录有 expires_at / assessment，则更新。"""
-    merged_sources = list(dict.fromkeys(existing.sources + new.sources))
-    # raw_sources 按 source 去重合并
-    existing_src_names = {r["source"] for r in existing.raw_sources}
-    merged_raw = list(existing.raw_sources) + [r for r in new.raw_sources if r["source"] not in existing_src_names]
+    merged_sources, merged_raw = _deduplicate_job_sources(
+        existing.sources + new.sources,
+        existing.raw_sources + new.raw_sources,
+    )
     new_expires = new.expires_at or existing.expires_at
     new_coarse_filter = new.coarse_filter or existing.coarse_filter
     new_assessment = new.assessment or existing.assessment
@@ -367,9 +435,11 @@ def merge_job_source(dedup_key: str, source: str) -> None:
         ).fetchone()
         if row is None:
             return
-        existing = json.loads(row["sources"] or "[]")
-        if source not in existing:
-            existing.append(source)
+        existing, _ = _deduplicate_job_sources(
+            json.loads(row["sources"] or "[]") + [source],
+            [],
+        )
+        if existing != json.loads(row["sources"] or "[]"):
             con.execute(
                 "UPDATE job_cache SET sources = ? WHERE dedup_key = ?",
                 (json.dumps(existing), dedup_key),
@@ -385,12 +455,10 @@ def merge_job_raw_source(dedup_key: str, source_entry: dict) -> None:
         ).fetchone()
         if row is None:
             return
-        sources = json.loads(row["sources"] or "[]")
-        raw_sources = json.loads(row["raw_sources"] or "[]")
-        if source_name not in sources:
-            sources.append(source_name)
-        if not any(item.get("source") == source_name for item in raw_sources):
-            raw_sources.append(source_entry)
+        sources, raw_sources = _deduplicate_job_sources(
+            json.loads(row["sources"] or "[]") + [source_name],
+            json.loads(row["raw_sources"] or "[]") + [source_entry],
+        )
         con.execute(
             "UPDATE job_cache SET sources = ?, raw_sources = ? WHERE dedup_key = ?",
             (json.dumps(sources), json.dumps(raw_sources), dedup_key),
@@ -460,14 +528,18 @@ def _row_to_job(row: sqlite3.Row) -> JobResult:
     raw_assessment = row["assessment"] if "assessment" in keys else None
     coarse_filter = CoarseFilterResult.model_validate_json(raw_coarse_filter) if raw_coarse_filter else None
     assessment = JobAssessment.model_validate_json(raw_assessment) if raw_assessment else None
+    sources, raw_sources = _deduplicate_job_sources(
+        json.loads(row["sources"] or "[]"),
+        json.loads(row["raw_sources"] if "raw_sources" in row.keys() and row["raw_sources"] else "[]"),
+    )
     return JobResult(
         title=row["title"],
         company=row["company"],
         location=row["location"] or "",
         url=row["url"],
         description_snippet=row["description_snippet"] or "",
-        sources=[s if isinstance(s, str) else s.get("source", "") for s in json.loads(row["sources"] or "[]")],
-        raw_sources=json.loads(row["raw_sources"] if "raw_sources" in row.keys() and row["raw_sources"] else "[]"),
+        sources=sources,
+        raw_sources=raw_sources,
         date_posted=row["date_posted"] if "date_posted" in row.keys() and row["date_posted"] else "",
         fetched_at=datetime.fromisoformat(row["fetched_at"]),
         expires_at=datetime.fromisoformat(row["expires_at"]) if row["expires_at"] else None,

jobradar/search_assessment_stage.py

@@ -329,6 +329,9 @@ def _is_visible_job(job_obj) -> bool:
                     "location": cached_job.location,
                     "url": cached_job.url,
                     "description_snippet": cached_job.description_snippet,
+                    "sources": cached_job.sources,
+                    "raw_sources": cached_job.raw_sources,
+                    "date_posted": cached_job.date_posted,
                     "expires_at": cached_job.expires_at,
                     "is_complete": cached_job.is_complete,
                     "coarse_filter": cached_job.coarse_filter,

tests/test_cache.py

@@ -65,6 +65,91 @@ def test_dedup_only_appends_source(self, temp_db):
         assert "indeed.com" in result.sources
         assert result.location == ""  # 以第一次为准
 
+    def test_provider_aliases_do_not_create_duplicate_sources(self, temp_db):
+        url = "https://ie.indeed.com/viewjob?jk=123"
+        original = make_job(
+            url=url,
+            sources=["indeed.ie", "linkedin.com"],
+            raw_sources=[
+                {"source": "indeed.ie", "url": url, "date_posted": "2026-08-07"},
+                {
+                    "source": "linkedin.com",
+                    "url": "https://www.linkedin.com/jobs/view/123",
+                    "date_posted": "2026-08-07",
+                },
+            ],
+        )
+        alias = make_job(
+            url=url,
+            sources=["ie.indeed.com"],
+            raw_sources=[{"source": "ie.indeed.com", "url": url, "date_posted": ""}],
+        )
+
+        temp_db.save_job(original)
+        temp_db.save_job(alias)
+
+        result = temp_db.get_job(original.dedup_key)
+        assert result is not None
+        assert result.sources == ["indeed.ie", "linkedin.com"]
+        assert result.raw_sources == original.raw_sources
+
+    def test_read_deduplicates_existing_provider_aliases(self, temp_db):
+        url = "https://ie.indeed.com/viewjob?jk=123"
+        job = make_job(
+            url=url,
+            sources=["indeed.ie"],
+            raw_sources=[{"source": "indeed.ie", "url": url, "date_posted": "2026-08-07"}],
+        )
+        temp_db.save_job(job)
+
+        with sqlite3.connect(os.environ["CACHE_DB_PATH"]) as con:
+            con.execute(
+                "UPDATE job_cache SET sources = ?, raw_sources = ? WHERE dedup_key = ?",
+                (
+                    '["indeed.ie", "ie.indeed.com", "linkedin.com"]',
+                    (
+                        '[{"source":"indeed.ie","url":"https://ie.indeed.com/viewjob?jk=123",'
+                        '"date_posted":"2026-08-07"},'
+                        '{"source":"ie.indeed.com","url":"https://ie.indeed.com/viewjob?jk=123",'
+                        '"date_posted":""},'
+                        '{"source":"linkedin.com","url":"https://www.linkedin.com/jobs/view/123",'
+                        '"date_posted":"2026-08-07"}]'
+                    ),
+                    job.dedup_key,
+                ),
+            )
+
+        result = temp_db.get_job(job.dedup_key)
+        assert result is not None
+        assert result.sources == ["indeed.ie", "linkedin.com"]
+        assert result.raw_sources == [
+            {"source": "indeed.ie", "url": url, "date_posted": "2026-08-07"},
+            {
+                "source": "linkedin.com",
+                "url": "https://www.linkedin.com/jobs/view/123",
+                "date_posted": "2026-08-07",
+            },
+        ]
+
+    def test_merge_raw_source_treats_provider_alias_as_existing(self, temp_db):
+        url = "https://ie.indeed.com/viewjob?jk=123"
+        job = make_job(
+            url=url,
+            sources=["indeed.ie"],
+            raw_sources=[{"source": "indeed.ie", "url": url, "date_posted": "2026-08-07"}],
+        )
+        temp_db.save_job(job)
+
+        temp_db.merge_job_raw_source(
+            job.dedup_key,
+            {"source": "ie.indeed.com", "url": url, "date_posted": ""},
+        )
+
+        result = temp_db.get_job(job.dedup_key)
+        assert result is not None
+        assert result.sources == ["indeed.ie"]
+        assert result.raw_sources == job.raw_sources
+
     def test_expires_at_updated_on_merge(self, temp_db):
         job1 = make_job(sources=["linkedin.com"])
         future = datetime.utcnow() + timedelta(days=5)

tests/test_new_features.py

@@ -803,6 +803,64 @@ def fake_match_job_to_cv(
         assert rejected == 0
         assert saved == 0
 
+    def test_cv_reassessment_preserves_cached_sources(self, db, monkeypatch: pytest.MonkeyPatch):
+        import jobradar.search_assessment_stage as stage
+        from jobradar.search_prefilter import PrefilterResult
+
+        indeed_url = "https://ie.indeed.com/viewjob?jk=123"
+        linkedin_url = "https://www.linkedin.com/jobs/view/123"
+        cached_job = JobResult(
+            title="Software Development Engineer",
+            company="Foxit",
+            location="Dublin",
+            url=indeed_url,
+            description_snippet="Build Python services.",
+            sources=["indeed.ie", "linkedin.com"],
+            raw_sources=[
+                {"source": "indeed.ie", "url": indeed_url, "date_posted": "2026-08-07"},
+                {"source": "linkedin.com", "url": linkedin_url, "date_posted": "2026-08-07"},
+            ],
+            date_posted="2026-08-07",
+        )
+        db.save_job(cached_job)
+
+        monkeypatch.setattr(
+            stage,
+            "batch_assess_jds",
+            lambda jobs, profile, llm, language="zh": [
+                JDAssessment(
+                    relevant=True,
+                    reason="match",
+                    score=9,
+                    strengths=["Python"],
+                    weaknesses=[],
+                    matched_keywords=["Python"],
+                )
+            ],
+        )
+        monkeypatch.setattr(
+            stage,
+            "_evaluate_jobs",
+            lambda tasks, **kwargs: [(task, None) for task in tasks],
+        )
+
+        stage.flush_assessments(
+            PrefilterResult(patch_pending=[(cached_job, cached_job.description_snippet)]),
+            job_all_sources={},
+            profile=_make_profile(),
+            llm=_make_llm(),
+            cv_hash="new-cv",
+            cb=lambda msg: None,
+            on_job=None,
+            language="zh",
+        )
+
+        refreshed = db.get_job(cached_job.dedup_key)
+        assert refreshed is not None
+        assert refreshed.sources == cached_job.sources
+        assert refreshed.raw_sources == cached_job.raw_sources
+        assert refreshed.date_posted == cached_job.date_posted
+
     def test_pipeline_no_longer_writes_the_legacy_assessment_column(self, db, monkeypatch: pytest.MonkeyPatch):
         """评分只经 job_matches 落库；legacy 列不再由管道写入。"""
         import sqlite3