🎯 같은 사진 재사용을 내용 해시로 막습니다

blog-automation/fix_unlicensed_images.py

@@ -125,28 +125,53 @@ def _iter_posts(blog_id: str, token: str):
 def _find_safe_replacement(title: str, keyword: str = "") -> dict | None:
     """제목 기반으로 저작권 안전한 대체 이미지를 확보합니다.
 
-    _fetch_best_image는 이미 화이트리스트가 적용된 소스만 검색하지만,
-    반환값을 한 번 더 검증합니다 — 여기서 안전하지 않은 것이 통과하면
-    무단 이미지를 다른 무단 이미지로 바꾸는 최악의 결과가 됩니다.
+    같은 사진이 여러 글에 반복되지 않도록, 후보를 관련성 순으로 훑으면서
+    "이미 많이 쓴 사진"은 건너뜁니다.
+
+    URL이나 제목으로는 이 판정을 할 수 없습니다 — Pixabay는 같은 사진에
+    매번 다른 서명 URL을 주고, 태그 문자열도 조금씩 달라집니다. 내려받아
+    내용 해시를 계산한 뒤에야 같은 사진인 줄 압니다. 그래서 복제기가
+    돌려주는 uses를 보고 판단합니다 (2026-08-25 실측: 사진 한 장이 61편).
+
+    후보가 전부 과다 사용이면 그중 가장 덜 쓴 것을 씁니다 — 이미지 없는
+    글보다는 반복이라도 낫고, 여기서 None을 돌려주면 교체 대상 이미지가
+    삭제됩니다.
     """
+    import image_mirror
     from image_fetcher import (
-        _fetch_best_image, _simplify_query, adopt_image, generate_fallback_image,
-        is_license_safe,
+        _score_title_relevance, _search_all_sources, _simplify_query, adopt_image,
+        filter_license_safe, generate_fallback_image, is_license_safe,
     )
 
     query = _simplify_query(title, 3) or title
-    img = _fetch_best_image(
-        query, n_candidates=6, pixabay_api_key=os.getenv("PIXABAY_API_KEY", ""),
-    )
-    if img and is_license_safe(img.get("url", "")):
-        img["alt_text"] = query[:50]
-        return adopt_image(img)
+    candidates = filter_license_safe(_search_all_sources(
+        query, "", "", 8, pixabay_api_key=os.getenv("PIXABAY_API_KEY", ""),
+    ))
+    candidates.sort(key=lambda c: -_score_title_relevance(c, query))
+
+    least_used = None
+    for cand in candidates[:5]:
+        if not is_license_safe(cand.get("url", "")):
+            continue
+        cand["alt_text"] = query[:50]
+        got = adopt_image(cand)
+        if not got:
+            continue
+        uses = image_mirror.times_used(got.get("url", ""))
+        if uses <= image_mirror.MAX_REUSE:
+            return got
+        logger.info(f"    이미 {uses}편에 쓴 사진 — 다른 후보 확인: {query[:20]}")
+        if least_used is None or uses < least_used[0]:
+            least_used = (uses, got)
 
     thumb = generate_fallback_image(title, keyword)
     if thumb and is_license_safe(thumb.get("url", "")):
         kind = "AI 생성 이미지" if thumb.get("source") == "ai_generated" else "생성 썸네일"
-        logger.info(f"    검색 결과 없음 → {kind} 사용")
+        logger.info(f"    쓸 만한 새 사진 없음 → {kind} 사용")
         return thumb
+    if least_used:
+        logger.warning(f"    후보가 전부 과다 사용 — {least_used[0]}편짜리 사진을 재사용")
+        return least_used[1]
     return None
 
 

blog-automation/image_mirror.py

@@ -233,8 +233,11 @@ def mirror_url(url: str) -> dict | None:
         rec = manifest["items"][known]
         path = os.path.join(MIRROR_DIR, rec["file"])
         if os.path.exists(path):
+            rec["uses"] = times_used_by_record(rec) + 1
+            save_manifest(manifest)
             return {"url": f"{public_base()}/{rec['file']}", "file": rec["file"],
-                    "sha": known, "bytes": rec.get("bytes", 0), "reused": True}
+                    "sha": known, "bytes": rec.get("bytes", 0), "reused": True,
+                    "uses": rec["uses"]}
         # 매니페스트에는 있는데 파일이 없으면 다시 내려받습니다
 
     got = _download(url)
@@ -258,12 +261,46 @@ def mirror_url(url: str) -> dict | None:
         digest, {"file": name, "bytes": len(raw), "mime": mime,
                  "firstSeen": datetime.now(timezone.utc).isoformat(), "origins": []})
     rec["file"] = name
+    # 같은 사진을 몇 편에 넣었는지 셉니다. Pixabay는 같은 사진에 매번 다른
+    # 서명 URL을 주므로 origins 개수로는 정확히 셀 수 없습니다 — 내려받아
+    # 해시를 계산한 지금이 유일하게 확실한 시점입니다.
+    #
+    # 이번 원본을 목록에 넣기 **전에** 세야 합니다. 넣고 나서 세면 방금 것을
+    # 이미 쓴 것으로 잘못 포함해 한 번씩 부풀려집니다.
+    prior = times_used_by_record(rec)
     if url not in rec["origins"]:
         rec["origins"].append(url)
+    rec["uses"] = prior + 1
     save_manifest(manifest)
 
     return {"url": f"{public_base()}/{name}", "file": name, "sha": digest,
-            "bytes": len(raw), "reused": False}
+            "bytes": len(raw), "reused": prior > 0, "uses": rec["uses"]}
+
+
+#: 한 사진을 이 편수까지만 씁니다. 넘으면 다른 후보를 찾습니다.
+#: 2026-08-25 실측에서 사진 한 장이 61편에 들어가 있었습니다.
+MAX_REUSE = 3
+
+
+def times_used_by_record(rec: dict) -> int:
+    """이 사진이 몇 편에 쓰였는지.
+
+    uses가 없는 옛 기록은 origins 개수로 셉니다 — 같은 사진에 서명 URL이
+    여러 개 붙은 만큼은 실제로 쓰인 것이므로 최소값으로는 맞습니다.
+    """
+    return int(rec.get("uses") or len(rec.get("origins", [])) or 0)
+
+
+def times_used(mirrored_url: str) -> int:
+    """복제본 URL이 몇 편에 쓰였는지 (모르면 0)."""
+    m = re.search(r"/([0-9a-f]{16})\.[a-z]+$", mirrored_url or "")
+    if not m:
+        return 0
+    return times_used_by_record(load_manifest().get("items", {}).get(m.group(1), {}))
+
+
+def is_overused(mirrored_url: str, limit: int = MAX_REUSE) -> bool:
+    return times_used(mirrored_url) > limit
 
 
 def mirror_images(images: list[dict]) -> int:

blog-automation/tests/test_fix_unlicensed_images.py

@@ -121,8 +121,9 @@ def test_rejects_unsafe_search_result(self, monkeypatch):
         import fix_unlicensed_images as fx
         import image_fetcher
 
-        monkeypatch.setattr(image_fetcher, "_fetch_best_image",
-                            lambda *a, **k: {"url": NAVER, "title": "x"})
+        monkeypatch.setattr(image_fetcher, "_search_all_sources",
+                            lambda *a, **k: [{"url": NAVER, "title": "x"}])
+        monkeypatch.setattr(image_fetcher, "adopt_image", lambda img: img)
         monkeypatch.setattr(image_fetcher, "generate_fallback_image",
                             lambda *a, **k: {"url": PIXA, "source": "generated_thumbnail"})
         out = fx._find_safe_replacement("제목")
@@ -132,8 +133,9 @@ def test_returns_none_when_everything_unsafe(self, monkeypatch):
         import fix_unlicensed_images as fx
         import image_fetcher
 
-        monkeypatch.setattr(image_fetcher, "_fetch_best_image",
-                            lambda *a, **k: {"url": NAVER})
+        monkeypatch.setattr(image_fetcher, "_search_all_sources",
+                            lambda *a, **k: [{"url": NAVER}])
+        monkeypatch.setattr(image_fetcher, "adopt_image", lambda img: img)
         monkeypatch.setattr(image_fetcher, "generate_fallback_image",
                             lambda *a, **k: {"url": NAVER})
         assert fx._find_safe_replacement("제목") is None
@@ -142,8 +144,9 @@ def test_prefers_search_result_when_safe(self, monkeypatch):
         import fix_unlicensed_images as fx
         import image_fetcher
 
-        monkeypatch.setattr(image_fetcher, "_fetch_best_image",
-                            lambda *a, **k: {"url": WIKI, "title": "x"})
+        monkeypatch.setattr(image_fetcher, "_search_all_sources",
+                            lambda *a, **k: [{"url": WIKI, "title": "x"}])
+        monkeypatch.setattr(image_fetcher, "adopt_image", lambda img: img)
         monkeypatch.setattr(image_fetcher, "generate_fallback_image",
                             lambda *a, **k: {"url": PIXA})
         assert fx._find_safe_replacement("제목")["url"] == WIKI

blog-automation/tests/test_image_reuse.py

@@ -0,0 +1,123 @@
+"""같은 사진이 여러 글에 반복되지 않게 — 내용 해시로 판정합니다.
+
+URL이나 제목으로는 이 판정을 할 수 없습니다. Pixabay는 같은 사진에 매번
+다른 서명 URL을 주고 태그 문자열도 조금씩 다릅니다. 2026-08-25 파일럿에서
+새로 채택한 13장 중 10장이 이미 저장소에 있는 사진과 바이트 단위로
+동일했고, 한 장은 61편에 들어가 있었습니다.
+
+내려받아 해시를 계산한 뒤에야 같은 사진인 줄 알 수 있으므로, 복제기가
+세어 주는 사용 횟수로 판정합니다.
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+import image_mirror  # noqa: E402
+
+PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 40
+
+
+@pytest.fixture
+def store(tmp_path, monkeypatch):
+    monkeypatch.setattr(image_mirror, "MIRROR_DIR", str(tmp_path / "images"))
+    monkeypatch.setattr(image_mirror, "MANIFEST_FILE", str(tmp_path / "m.json"))
+    monkeypatch.setenv("IMAGE_HOST_BASE", "https://example.test/images")
+    return tmp_path
+
+
+class _Resp:
+    status_code = 200
+
+    def __init__(self, body=PNG):
+        self.body = body
+        self.headers = {"Content-Type": "image/png"}
+
+    def raise_for_status(self):
+        pass
+
+    def iter_content(self, n):
+        yield self.body
+
+
+def test_same_photo_from_different_urls_counts_up(store, monkeypatch):
+    """서명 URL이 달라도 내용이 같으면 같은 사진입니다."""
+    monkeypatch.setattr(image_mirror.requests, "get", lambda *a, **k: _Resp())
+    a = image_mirror.mirror_url("https://pixabay.com/get/sig1.jpg")
+    b = image_mirror.mirror_url("https://pixabay.com/get/sig2.jpg")
+    assert a["file"] == b["file"], "같은 내용인데 다른 파일로 저장됐습니다"
+    assert image_mirror.times_used(b["url"]) == 2
+
+
+def test_reusing_the_same_url_also_counts(store, monkeypatch):
+    monkeypatch.setattr(image_mirror.requests, "get", lambda *a, **k: _Resp())
+    url = "https://pixabay.com/get/sig1.jpg"
+    image_mirror.mirror_url(url)
+    again = image_mirror.mirror_url(url)
+    assert again["reused"] is True
+    assert image_mirror.times_used(again["url"]) == 2
+
+
+def test_overuse_threshold(store, monkeypatch):
+    monkeypatch.setattr(image_mirror.requests, "get", lambda *a, **k: _Resp())
+    got = None
+    for i in range(image_mirror.MAX_REUSE):
+        got = image_mirror.mirror_url(f"https://pixabay.com/get/s{i}.jpg")
+    assert image_mirror.is_overused(got["url"]) is False
+    got = image_mirror.mirror_url("https://pixabay.com/get/extra.jpg")
+    assert image_mirror.is_overused(got["url"]) is True
+
+
+def test_old_records_without_uses_fall_back_to_origins(store):
+    """오늘 이전 기록에는 uses가 없습니다 — origins 개수가 최소값입니다."""
+    image_mirror.save_manifest({"items": {"a" * 16: {
+        "file": "a" * 16 + ".jpg", "origins": ["u1", "u2", "u3", "u4"]}}})
+    assert image_mirror.times_used(f"https://example.test/images/{'a' * 16}.jpg") == 4
+    assert image_mirror.is_overused(f"https://example.test/images/{'a' * 16}.jpg") is True
+
+
+def test_unknown_url_is_not_overused(store):
+    assert image_mirror.times_used("https://example.test/images/deadbeefdeadbeef.jpg") == 0
+    assert image_mirror.is_overused("https://elsewhere/x.jpg") is False
+
+
+# ── 후보 재선택 ──────────────────────────────────────────────────────────────
+
+def _wire(monkeypatch, candidates, uses_by_url):
+    """검색·채택·사용횟수를 갈아 끼웁니다."""
+    import fix_unlicensed_images as fx
+    monkeypatch.setattr("image_fetcher._search_all_sources",
+                        lambda *a, **k: [dict(c) for c in candidates])
+    monkeypatch.setattr("image_fetcher.filter_license_safe", lambda imgs: imgs)
+    monkeypatch.setattr("image_fetcher.is_license_safe", lambda u: True)
+    monkeypatch.setattr("image_fetcher._score_title_relevance",
+                        lambda img, q: img.get("score", 0))
+    monkeypatch.setattr("image_fetcher.adopt_image", lambda img: img)
+    monkeypatch.setattr("image_fetcher.generate_fallback_image", lambda t, k: None)
+    monkeypatch.setattr(image_mirror, "times_used",
+                        lambda url: uses_by_url.get(url, 0))
+    return fx
+
+
+def test_overused_candidate_is_skipped(monkeypatch):
+    """가장 관련성 높은 후보라도 이미 많이 썼으면 다음으로 넘어갑니다."""
+    fx = _wire(monkeypatch,
+               [{"url": "a", "score": 1.0}, {"url": "b", "score": 0.9}],
+               {"a": 99, "b": 0})
+    assert fx._find_safe_replacement("어떤 글 제목")["url"] == "b"
+
+
+def test_all_overused_reuses_the_least_used(monkeypatch):
+    """전부 과다 사용이면 가장 덜 쓴 것 — 여기서 None을 주면 이미지가 삭제됩니다."""
+    fx = _wire(monkeypatch,
+               [{"url": "a", "score": 1.0}, {"url": "b", "score": 0.9}],
+               {"a": 50, "b": 10})
+    assert fx._find_safe_replacement("어떤 글 제목")["url"] == "b"
+
+
+def test_fresh_candidate_wins_immediately(monkeypatch):
+    fx = _wire(monkeypatch, [{"url": "a", "score": 1.0}], {"a": 0})
+    assert fx._find_safe_replacement("어떤 글 제목")["url"] == "a"