feat(finder): кэш директоров (не платить дважды) + дедуп фирм по карточке 2ГИС + «обновить контакты»

Работа другой (закрытой) сессии, уже выкаченной на боевой finder. Коммичу по
разрешению владельца («почисти за той сессией») после сверки: finder pytest 421
passed на текущем дереве.

- contactcache.py (новый): кэш собранного по личному ИНН директора — один человек
  бывает директором нескольких фирм, платим за него ОДИН раз (в т.ч. из прошлых
  списков). Изоляция каталога кэша на тест — tests/conftest.py.
- dedup.py (новый): fix_and_collapse — чинит ошибочный сайт фирмы по её карточке
  2ГИС (реклама соседей не путает) и схлопывает дубли.
- contactflow/web(app,bulk): подключён cache в сбор контактов (одиночный и массовый)
  + флаг refresh («обновить» — пересобрать заново даже уже собранных, платно) +
  счётчик «повторный директор N — не платили дважды».
- app.py _real_pipeline: dedupe-хук (_card_site/_reprocess — пересбор фирмы по
  исправленному домену).
- pipeline.py: точка вызова dedupe. requirements.txt: +pypdf (чтение выписки ЕГРЮЛ).
- Тесты: test_contactcache/test_dedup (новые), test_bulk/test_contactflow/
  test_pipeline/test_ui обновлены. Всего finder pytest 421 passed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

моя/sales-finder/requirements.txt

@@ -3,3 +3,5 @@ uvicorn==0.49.0
 requests==2.34.2
 openpyxl==3.1.5
 python-multipart==0.0.32
+pypdf==6.14.2
+

моя/sales-finder/salesfinder/contactcache.py

@@ -0,0 +1,46 @@
+"""Кэш собранных контактов по ЛИЧНОМУ ИНН директора.
+
+Зачем: сбор контактов — ПЛАТНЫЙ запрос (сервис inn_fl + проверка номеров ДаДата/HLR).
+Один человек бывает директором нескольких фирм — в одном списке, а то и в разных
+городах/отраслях. Платить за него каждый раз заново — деньги на ветер. Собрали раз —
+помним и переиспользуем ВЕЗДЕ, даже в других списках.
+
+Файл на ИНН (как cache/egrul). ПДн — внутренний инструмент, каталог cache/ в .gitignore.
+Пустой результат («человека нет») тоже кэшируем — чтобы не платить повторно за заведомо
+пустого. Ошибку сети НЕ кэшируем (это забота вызывающего — просто не звать put).
+"""
+import os
+import json
+
+
+class ContactCache:
+    def __init__(self, cache_dir=None):
+        # По умолчанию — cache/contacts рядом с прогонами (переживает перезапуск, в .gitignore).
+        # Тесты подменяют каталог через SF_CONTACT_CACHE, чтобы не писать в боевой кэш.
+        self.dir = cache_dir or os.environ.get("SF_CONTACT_CACHE", "cache/contacts")
+        self.hits = 0        # сколько раз отдали из кэша (= сэкономленных платных запросов)
+
+    def _path(self, inn):
+        return os.path.join(self.dir, f"{str(inn).strip()}.json")
+
+    def get(self, inn):
+        """Собранный контакт по ИНН или None. Хит увеличивает счётчик экономии."""
+        if not inn or not str(inn).strip():
+            return None
+        try:
+            with open(self._path(inn), encoding="utf-8") as fh:
+                data = json.load(fh)
+        except (OSError, ValueError):
+            return None
+        self.hits += 1
+        return data
+
+    def put(self, inn, bundle):
+        """Сохранить собранный контакт по ИНН (атомарно: пишем во временный, потом replace)."""
+        if not inn or not str(inn).strip():
+            return
+        os.makedirs(self.dir, exist_ok=True)
+        tmp = self._path(inn) + ".tmp"
+        with open(tmp, "w", encoding="utf-8") as fh:
+            json.dump(bundle, fh, ensure_ascii=False)
+        os.replace(tmp, self._path(inn))

моя/sales-finder/salesfinder/contactflow.py

@@ -7,13 +7,19 @@
 """
 from salesfinder import phonecheck as _phonecheck
 
+# Поля, из которых складываем «досье контакта» для кэша по личному ИНН директора.
+_SNAP = ("contact_phone", "contact_phones_checked", "contact_check_warning",
+         "contact_email", "contact_status")
+
 
 def collect_for_firm(firm, deps) -> bool:
     """Собрать контакты фирмы. True — что-то нашли. Фирму НИКОГДА не теряем:
     при любой ошибке пишем причину в contact_status и идём дальше.
 
     deps: lookup(inn, tpl), check(phones, city, dadata_keys, smsc) -> (размеченные, warning),
-          tpl, dadata_keys, smsc, enrich(firm) -> добирает личный ИНН из ФНС (по требованию).
+          tpl, dadata_keys, smsc, enrich(firm) -> добирает личный ИНН из ФНС (по требованию),
+          cache -> ContactCache: собранное по личному ИНН, чтобы не платить дважды за одного
+          человека (он бывает директором нескольких фирм — в т.ч. в разных списках/отраслях).
     """
     if not firm.director_inn and firm.inn and deps.get("enrich"):
         # Личный ИНН директора мог не добраться при сборе списка (ФНС отдаёт капчу/пустоту
@@ -25,11 +31,22 @@ def collect_for_firm(firm, deps) -> bool:
             pass
     if not firm.director_inn:          # ни личного ИНН, ни возможности его добрать — искать не по чему
         return False
+
+    # Этого директора уже собирали (в этом списке или в прошлом) — берём из кэша, НЕ платим.
+    # Режим «обновить» (deps['refresh']) кэш НЕ читает — собирает заново и перезапишет свежим.
+    cache = deps.get("cache")
+    if cache is not None and not deps.get("refresh"):
+        hit = cache.get(firm.director_inn)
+        if hit is not None:
+            for k in _SNAP:
+                setattr(firm, k, hit.get(k))
+            return bool(firm.contact_phone or firm.contact_email)
+
     try:
         res = deps["lookup"](firm.director_inn, deps["tpl"]) or {}
     except Exception:
         firm.contact_status = "error"
-        return False
+        return False                    # сбой сети НЕ кэшируем — при другой фирме попробуем снова
 
     raw = [p for p in (res.get("phone") or "").split(", ") if p]
     try:
@@ -46,4 +63,7 @@ def collect_for_firm(firm, deps) -> bool:
         firm.contact_phone = (res.get("phone") or "").strip() or None
     firm.contact_email = (res.get("email") or "").strip() or None
     firm.contact_status = res.get("status")
+    # Запомнили результат по личному ИНН (в т.ч. пустой «not_found») — больше за него не платим.
+    if cache is not None:
+        cache.put(firm.director_inn, {k: getattr(firm, k) for k in _SNAP})
     return bool(firm.contact_phone or firm.contact_email)

моя/sales-finder/salesfinder/dedup.py

@@ -0,0 +1,66 @@
+"""Починка ошибочно склеенного ИНН + схлопывание дублей одной компании.
+
+Зачем: список 2ГИС иногда приклеивает карточке чужой сайт (рядом крутится реклама
+крупного клуба — DDX/World Class), из-за чего РАЗНЫЕ фирмы получают ОДИН ИНН.
+А ещё у одной компании в 2ГИС бывает несколько карточек (направления, филиалы).
+В «списке на прозвон» звонят директору один раз — значит одна компания = одна строка.
+
+Два шага, строго по порядку:
+  1. ПОЧИНКА (вариант А). Среди фирм, у которых ИНН совпал с чьим-то ещё, проверяем
+     каждую по ЕЁ карточке 2ГИС (там сайт всегда верный). Если сайт карточки не совпал
+     с текущим — значит ИНН приклеился чужой: ставим верный сайт и заново выводим ИНН.
+     Одиночек (уникальный ИНН) не трогаем — карточку зря не дёргаем.
+  2. СХЛОПЫВАНИЕ. Группируем по (уже исправленному) ИНН, оставляем САМУЮ ГОРЯЧУЮ
+     карточку, на ней ставим merged_count = сколько ещё карточек той же компании скрыли.
+
+Внешние операции (карточка, пере-обогащение) — через deps, чтобы тестировать без сети.
+"""
+
+
+def fix_and_collapse(firms, deps):
+    """deps: card_site(firm)->домен|None (сайт из карточки 2ГИС),
+             reprocess(firm)->None (пере-вывести ИНН/данные по firm.domain)."""
+    _fix_misglued_inns(firms, deps)
+    return collapse_by_inn(firms)
+
+
+def _fix_misglued_inns(firms, deps):
+    by_inn = {}
+    for f in firms:
+        if f.inn:
+            by_inn.setdefault(f.inn, []).append(f)
+    for grp in by_inn.values():
+        if len(grp) < 2:               # уникальный ИНН — проверять нечего
+            continue
+        for f in grp:
+            try:                            # сбой сети на одной фирме не роняет весь сбор
+                card = deps["card_site"](f)
+                if card and card != f.domain:   # карточка знает другой сайт — ИНН был чужой
+                    f.domain = card
+                    deps["reprocess"](f)        # заново выведет верный ИНН по верному сайту
+            except Exception:
+                pass                        # не починили — оставим как есть, но не упадём
+
+
+def collapse_by_inn(firms):
+    """Одна компания (один ИНН) = одна строка: оставляем самую горячую, считаем скрытых.
+    Позиция группы — по первому появлению; фирмы без ИНН не схлопываются."""
+    out, pos, extra = [], {}, {}
+    for f in firms:
+        inn = f.inn
+        if not inn:
+            f.merged_count = 0
+            out.append(f)
+            continue
+        if inn not in pos:
+            pos[inn] = len(out)
+            extra[inn] = 0
+            out.append(f)
+        else:
+            extra[inn] += 1
+            kept = out[pos[inn]]
+            if (f.hotness or 0) > (kept.hotness or 0):
+                out[pos[inn]] = f       # горячее — на то же место
+    for inn, p in pos.items():
+        out[p].merged_count = extra[inn]
+    return out

моя/sales-finder/salesfinder/models.py

@@ -43,6 +43,9 @@ class Firm:
     # Название поля осталось прежним — оно уже лежит в сохранённых прогонах в БД.
     # Ставится только после успешной загрузки — чтобы групповая загрузка не дублировала.
     warmed_channels: list = field(default_factory=list)
+    # Сколько ещё карточек той же компании (тот же ИНН) схлопнуто в эту строку.
+    # 0 — единственная. Ставится dedup.collapse_by_inn после сбора. Для бейджа «+N».
+    merged_count: int = 0
 
     def key(self) -> str:
         if self.domain:

моя/sales-finder/salesfinder/pipeline.py

@@ -77,6 +77,12 @@ def run_pipeline(rubric_slug, city, queries, deps, workers: int = 6,
     # 5) дорогое обогащение — ТОЛЬКО по оставшимся (реквизиты, ДаДата, ЕГРЮЛ)
     pmap(lambda f: deps["enrich"](f), firms, workers=workers)
 
+    # 5.5) починка ошибочно склеенного ИНН (сайт из карточки) + схлопывание дублей
+    # одной компании (один ИНН = одна строка). До ранга — чтобы бейдж «+N» был на
+    # правильной, самой горячей карточке.
+    if deps.get("dedupe"):
+        firms = deps["dedupe"](firms)
+
     # Лог НИКОГДА не должен ронять прогон (на Windows print с не-cp1251 символом
     # кидал UnicodeEncodeError и убивал весь сбор — реальный инцидент 12.07.2026).
     if deps.get("log"):

моя/sales-finder/tests/conftest.py

@@ -0,0 +1,8 @@
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _isolate_contact_cache(tmp_path, monkeypatch):
+    """Каждый тест — свой каталог кэша контактов, чтобы ContactCache() в обработчиках сбора
+    не писал в боевой cache/contacts и не «протекал» собранным между тестами."""
+    monkeypatch.setenv("SF_CONTACT_CACHE", str(tmp_path / "contacts"))

моя/sales-finder/tests/test_bulk.py

@@ -213,6 +213,52 @@ def test_bulk_contacts_reports_no_inn_separately_from_not_found(tmp_path):
     assert "не нашлось" not in r.text.lower(), "никого реально не искали — «не нашлось» врёт"
 
 
+def test_bulk_contacts_pays_once_per_director_and_reports_reuse(tmp_path):
+    """Один директор (личный ИНН) у двух горячих фирм — платный запрос ОДИН, в отчёте
+    честно «повторный директор 1 — не платили дважды»."""
+    c, appmod = make_client(tmp_path)
+    rid = _seed(appmod, [Firm(name="Ф1", domain="f1.ru", hotness=90, director_inn="777"),
+                         Firm(name="Ф2", domain="f2.ru", hotness=90, director_inn="777")])
+    called = []
+    appmod.CONTACT_LOOKUP = lambda inn, tpl: called.append(inn) or {
+        "phone": "79991112233", "email": "", "status": "ok"}
+    appmod.PHONE_CHECK = lambda phones, city, dk, sm: (
+        [{"phone": p, "kind": "mobile", "alive": "alive"} for p in phones], None)
+    appmod._contact_template = lambda: "https://x/{inn}"
+    r = c.post("/bulk/contacts", data={"ids": [rid]})
+    assert called == ["777"], "заплатили за одного директора дважды"
+    assert "повторный директор 1" in r.text.lower()
+    # обе фирмы получили телефон
+    saved = appmod.STORE.get(rid)["firms"]
+    assert all(f["contact_phone"] for f in saved)
+
+
+def test_bulk_contacts_refresh_recollects_already_collected(tmp_path):
+    """«Обновить»: фирма с уже собранным телефоном пересобирается заново (платим) и
+    получает свежий номер — на случай, когда у директора сменился телефон."""
+    c, appmod = make_client(tmp_path)
+    rid = _seed(appmod, [Firm(name="Есть", domain="e.ru", hotness=90, director_inn="55",
+                              contact_phone="70000000000")])     # уже собран, старый
+    called = []
+    appmod.CONTACT_LOOKUP = lambda inn, tpl: called.append(inn) or {
+        "phone": "79995556677", "email": "", "status": "ok"}
+    appmod.PHONE_CHECK = lambda phones, city, dk, sm: (
+        [{"phone": p, "kind": "mobile", "alive": "alive"} for p in phones], None)
+    appmod._contact_template = lambda: "https://x/{inn}"
+    r = c.post("/bulk/contacts", data={"ids": [rid], "refresh": "1"})
+    assert called == ["55"], "в обычном режиме собранного бы пропустили — а тут должны пересобрать"
+    assert appmod.STORE.get(rid)["firms"][0]["contact_phone"] == "79995556677"   # обновлён
+
+
+def test_bulk_confirm_contacts_refresh_says_recollect_and_carries_flag(tmp_path):
+    c, appmod = make_client(tmp_path)
+    rid = _seed(appmod, [Firm(name="A", domain="a.ru", hotness=90, director_inn="1",
+                              contact_phone="79990000000")])
+    r = c.post("/bulk/confirm", data={"ids": [rid], "action": "contacts", "refresh": "1"})
+    assert "ересобрать" in r.text                       # «Пересобрать телефоны …»
+    assert 'name=refresh value="1"' in r.text or 'name="refresh" value="1"' in r.text
+
+
 def test_bulk_delete_unknown_id_does_not_crash_or_delete_others(tmp_path):
     """Id несуществующего списка не должен ронять обработчик и не должен
     зацепить чужие/существующие списки."""

моя/sales-finder/tests/test_contactcache.py

@@ -0,0 +1,34 @@
+from salesfinder.contactcache import ContactCache
+
+
+def test_put_then_get_returns_bundle(tmp_path):
+    c = ContactCache(str(tmp_path / "contacts"))
+    c.put("500100732259", {"contact_phone": "79990001122", "contact_status": "ok"})
+    got = c.get("500100732259")
+    assert got == {"contact_phone": "79990001122", "contact_status": "ok"}
+
+
+def test_miss_returns_none_and_is_not_a_hit(tmp_path):
+    c = ContactCache(str(tmp_path / "contacts"))
+    assert c.get("000") is None
+    assert c.hits == 0
+
+
+def test_hit_increments_counter(tmp_path):
+    c = ContactCache(str(tmp_path / "contacts"))
+    c.put("1", {"contact_phone": "x"})
+    c.get("1"); c.get("1")
+    assert c.hits == 2
+
+
+def test_empty_inn_is_ignored(tmp_path):
+    c = ContactCache(str(tmp_path / "contacts"))
+    c.put("", {"contact_phone": "x"})      # не пишем по пустому ИНН
+    assert c.get("") is None
+
+
+def test_not_found_bundle_is_cacheable(tmp_path):
+    """«человека нет» тоже кэшируем — чтобы не платить повторно за заведомо пустого."""
+    c = ContactCache(str(tmp_path / "contacts"))
+    c.put("2", {"contact_phone": None, "contact_status": "not_found"})
+    assert c.get("2") == {"contact_phone": None, "contact_status": "not_found"}

моя/sales-finder/tests/test_contactflow.py

@@ -79,6 +79,91 @@ def test_enrich_attempted_but_empty_skips_paid_lookup():
     assert f.director_inn is None
 
 
+class _Cache:
+    """Кэш в памяти для тестов (как ContactCache, но без файлов)."""
+    def __init__(self):
+        self.store, self.hits = {}, 0
+
+    def get(self, inn):
+        if inn in self.store:
+            self.hits += 1
+            return self.store[inn]
+        return None
+
+    def put(self, inn, bundle):
+        self.store[inn] = bundle
+
+
+def test_same_director_inn_paid_once_and_reused():
+    """Один директор (личный ИНН) у двух фирм — платный запрос ОДИН, второй из кэша."""
+    calls, checks = [], []
+    deps = _deps()
+    base_lookup, base_check = deps["lookup"], deps["check"]
+    deps["lookup"] = lambda inn, tpl: (calls.append(inn), base_lookup(inn, tpl))[1]
+    deps["check"] = lambda *a, **k: (checks.append(1), base_check(*a, **k))[1]
+    deps["cache"] = _Cache()
+    f1 = Firm(name="A", director_inn="500")
+    f2 = Firm(name="B", director_inn="500")     # тот же человек, другая фирма
+    assert contactflow.collect_for_firm(f1, deps) is True
+    assert contactflow.collect_for_firm(f2, deps) is True
+    assert calls == ["500"], "заплатили за одного директора дважды"
+    assert len(checks) == 1, "проверку номеров (тоже платную) сделали дважды"
+    assert f2.contact_phone == f1.contact_phone
+    assert f2.contact_phones_checked == f1.contact_phones_checked
+    assert deps["cache"].hits == 1
+
+
+def test_refresh_bypasses_cache_and_overwrites_it():
+    """Режим «обновить»: даже если директор в кэше — собираем ЗАНОВО (платим) и
+    перезаписываем кэш свежим. Для случая, когда у директора сменился телефон."""
+    calls = []
+    deps = _deps()
+    base_lookup = deps["lookup"]
+    deps["lookup"] = lambda inn, tpl: (calls.append(inn), base_lookup(inn, tpl))[1]
+    cache = _Cache()
+    cache.store["500"] = {"contact_phone": "70000000000", "contact_status": "ok",
+                          "contact_phones_checked": [], "contact_check_warning": None,
+                          "contact_email": None}       # старое значение в кэше
+    deps["cache"] = cache
+    deps["refresh"] = True
+    f = Firm(name="A", director_inn="500")
+    assert contactflow.collect_for_firm(f, deps) is True
+    assert calls == ["500"], "в режиме обновления взяли из кэша, а не пересобрали"
+    assert f.contact_phone == "79991112233"                       # свежий, не старый
+    assert cache.store["500"]["contact_phone"] == "79991112233"   # кэш перезаписан
+    assert cache.hits == 0                                          # get даже не звали
+
+
+def test_not_found_is_cached_no_second_paid_call():
+    """«Человека нет» тоже помним — не платим повторно за заведомо пустого."""
+    calls = []
+    deps = _deps()
+    deps["lookup"] = lambda inn, tpl: (calls.append(inn),
+                                       {"phone": "", "email": "", "status": "not_found"})[1]
+    deps["cache"] = _Cache()
+    f1 = Firm(name="A", director_inn="9")
+    f2 = Firm(name="B", director_inn="9")
+    assert contactflow.collect_for_firm(f1, deps) is False
+    assert contactflow.collect_for_firm(f2, deps) is False
+    assert calls == ["9"]
+
+
+def test_lookup_error_is_not_cached_and_retried():
+    """Сбой сервиса не кэшируем — при следующей фирме пробуем снова (вдруг оживёт)."""
+    calls = []
+    def boom(inn, tpl):
+        calls.append(inn)
+        raise RuntimeError("сервис лёг")
+    deps = _deps()
+    deps["lookup"] = boom
+    deps["cache"] = _Cache()
+    f1 = Firm(name="A", director_inn="7")
+    f2 = Firm(name="B", director_inn="7")
+    assert contactflow.collect_for_firm(f1, deps) is False
+    assert contactflow.collect_for_firm(f2, deps) is False
+    assert calls == ["7", "7"]
+
+
 def test_no_company_inn_does_not_even_try_to_enrich():
     """Нет ни личного, ни ИНН фирмы — добирать нечем, в ФНС не ходим."""
     f = Firm(name="A")

моя/sales-finder/tests/test_dedup.py

@@ -0,0 +1,114 @@
+from salesfinder.models import Firm
+from salesfinder import dedup
+
+
+# ── Схлопывание по ИНН (чистая логика, без сети) ──────────────────────────
+
+def test_same_inn_collapses_to_one_keeping_hottest():
+    a = Firm(name="A", inn="111", hotness=60)
+    b = Firm(name="B", inn="111", hotness=90)     # горячее — её и оставляем
+    out = dedup.collapse_by_inn([a, b])
+    assert len(out) == 1
+    assert out[0].name == "B"
+    assert out[0].merged_count == 1
+
+
+def test_firms_without_inn_are_never_merged():
+    a = Firm(name="A", hotness=50)
+    b = Firm(name="B", hotness=50)
+    out = dedup.collapse_by_inn([a, b])
+    assert [f.name for f in out] == ["A", "B"]
+    assert all(f.merged_count == 0 for f in out)
+
+
+def test_group_position_is_first_appearance():
+    x = Firm(name="X", inn="1", hotness=10)
+    y = Firm(name="Y", inn="2", hotness=99)
+    x2 = Firm(name="X2", inn="1", hotness=5)      # та же компания, что X
+    out = dedup.collapse_by_inn([x, y, x2])
+    assert [f.name for f in out] == ["X", "Y"]    # группа «1» держит первую позицию
+    assert out[0].merged_count == 1 and out[1].merged_count == 0
+
+
+def test_three_of_one_company_count_two_extra_and_keep_hottest():
+    fs = [Firm(name="a", inn="7", hotness=30),
+          Firm(name="b", inn="7", hotness=95),
+          Firm(name="c", inn="7", hotness=40)]
+    out = dedup.collapse_by_inn(fs)
+    assert len(out) == 1 and out[0].name == "b" and out[0].merged_count == 2
+
+
+# ── Починка ошибочно склеенного ИНН (вариант А) ───────────────────────────
+
+def _deps(card_sites):
+    """card_sites: домен-карточки по фирме (по имени). reprocess пере-выводит ИНН из домена."""
+    calls = {"card": [], "reprocess": []}
+
+    def card_site(f):
+        calls["card"].append(f.name)
+        return card_sites.get(f.name)
+
+    def reprocess(f):
+        calls["reprocess"].append(f.name)
+        # «пере-обогащение»: ИНН выводим из (правильного) домена
+        f.inn = {"worldclass.ru": "WC", "ddxfitness.ru": "DDX"}.get(f.domain, f.inn)
+
+    return {"card_site": card_site, "reprocess": reprocess}, calls
+
+
+def test_wrong_glued_firm_is_reresolved_from_its_card():
+    # обе с ИНН DDX, но карточка второй говорит worldclass — её и чиним
+    ddx = Firm(name="DDX", inn="DDX", domain="ddxfitness.ru", hotness=80)
+    wc = Firm(name="WC-прога", inn="DDX", domain="ddxfitness.ru", hotness=90)
+    deps, calls = _deps({"DDX": "ddxfitness.ru", "WC-прога": "worldclass.ru"})
+    out = dedup.fix_and_collapse([ddx, wc], deps)
+    # у WC-проги домен исправлен и ИНН пере-выведен, значит это РАЗНЫЕ компании
+    assert wc.domain == "worldclass.ru" and wc.inn == "WC"
+    assert calls["reprocess"] == ["WC-прога"]      # чинили только ошибочную
+    assert {f.name for f in out} == {"DDX", "WC-прога"} and len(out) == 2
+
+
+def test_genuine_same_company_is_confirmed_not_reprocessed_then_collapsed():
+    # два бренда одного юрлица: карточки совпадают с доменами → не чиним, но схлопываем
+    a = Firm(name="Crocus", inn="SAME", domain="crocus.ru", hotness=70)
+    b = Firm(name="DreamFit", inn="SAME", domain="dreamfit.ru", hotness=85)
+    deps, calls = _deps({"Crocus": "crocus.ru", "DreamFit": "dreamfit.ru"})
+    out = dedup.fix_and_collapse([a, b], deps)
+    assert calls["reprocess"] == []                # ничего не пере-обогащали
+    assert len(out) == 1 and out[0].name == "DreamFit" and out[0].merged_count == 1
+
+
+def test_card_or_reprocess_failure_does_not_abort_the_run():
+    """Сбой сети на одной фирме не должен ронять весь сбор — её оставляем как есть."""
+    a = Firm(name="A", inn="DUP", domain="a.ru", hotness=70)
+    b = Firm(name="B", inn="DUP", domain="b.ru", hotness=80)
+
+    def card_site(f):
+        raise RuntimeError("рендер лёг")
+
+    out = dedup.fix_and_collapse([a, b], {"card_site": card_site, "reprocess": lambda f: None})
+    # ИНН не починили (сеть упала), значит схлопнулись как одна компания — но не упали
+    assert len(out) == 1 and out[0].merged_count == 1
+
+
+def test_unique_inn_firm_is_not_even_card_checked():
+    solo = Firm(name="Solo", inn="Z", domain="solo.ru", hotness=50)
+    deps, calls = _deps({"Solo": "solo.ru"})
+    out = dedup.fix_and_collapse([solo], deps)
+    assert calls["card"] == []                     # одиночку по карточке не дёргаем
+    assert len(out) == 1 and out[0].merged_count == 0
+
+
+def test_ddx_plus_four_worldclass_ends_as_two_rows():
+    firms = [Firm(name="DDX", inn="DDX", domain="ddxfitness.ru", hotness=80)]
+    cards = {"DDX": "ddxfitness.ru"}
+    for i, h in enumerate([70, 90, 60, 75]):
+        firms.append(Firm(name=f"WC{i}", inn="DDX", domain="ddxfitness.ru", hotness=h))
+        cards[f"WC{i}"] = "worldclass.ru"
+    deps, _ = _deps(cards)
+    out = dedup.fix_and_collapse(firms, deps)
+    names = {f.name for f in out}
+    assert len(out) == 2                            # DDX + один World Class
+    assert "DDX" in names
+    wc = next(f for f in out if f.name != "DDX")
+    assert wc.inn == "WC" and wc.hotness == 90 and wc.merged_count == 3

моя/sales-finder/tests/test_pipeline.py

@@ -18,3 +18,20 @@ def test_pipeline_wires_stages():
     assert f.advertises == "да"                    # сигнал проставлен
     assert f.director == "Иванов И.И."              # обогащено
     assert f.hotness > 0                            # ранжировано
+
+
+def test_pipeline_applies_dedupe_when_provided():
+    """Шаг 5.5: если задан deps['dedupe'] — он схлопывает список после обогащения."""
+    seen = {}
+    deps = dict(
+        collect_2gis=lambda city, qs: [
+            Firm(name="A", domain="a.ru", two_gis_promoted=True, sources=["2gis"]),
+            Firm(name="B", domain="b.ru", two_gis_promoted=True, sources=["2gis"])],
+        collect_keyso=lambda kw: [],
+        scan_ad=lambda firm: {"advertises": "да", "confidence": "высокая", "evidence": []},
+        enrich=lambda firm: setattr(firm, "inn", "SAME"),   # обе стали одной компанией
+        dedupe=lambda firms: (seen.update(got=firms), firms[:1])[1],
+    )
+    firms = run_pipeline("s", "spb", ["q"], deps)
+    assert len(seen["got"]) == 2       # dedupe получил обе фирмы
+    assert len(firms) == 1             # и его результат ушёл дальше (в ранг)

моя/sales-finder/tests/test_ui.py

@@ -369,6 +369,29 @@ def test_results_checkbox_enabled_with_personal_or_company_inn():
     assert 'name=sel value="1"' in html          # есть ИНН фирмы — тоже можно
 
 
+def test_results_collect_form_has_refresh_checkbox():
+    """У «Собрать контакты» есть галочка «обновить» — пересобрать уже собранных."""
+    run = {"id": 40, "rubric": "s", "city": "moscow", "status": "done",
+           "firms": [{"name": "X", "inn": "1", "director_inn": "2", "hotness": 90, "evidence": []}]}
+    html = ui.results_html(run)
+    assert 'name=refresh value="1"' in html
+    assert "обновить" in html
+
+
+def test_results_shows_merged_badge_for_collapsed_company():
+    """Схлопнутые карточки одной компании — бейдж «+N» у названия."""
+    run = {"id": 30, "rubric": "s", "city": "moscow", "status": "done",
+           "firms": [
+               {"name": "World Class", "inn": "1", "hotness": 90, "merged_count": 3, "evidence": []},
+               {"name": "DDX", "inn": "2", "hotness": 80, "merged_count": 0, "evidence": []},
+           ]}
+    html = ui.results_html(run)
+    assert "+3" in html
+    assert "той же компании" in html
+    # у одиночной (merged_count=0) бейджа быть не должно
+    assert html.count("той же компании") == 1
+
+
 def test_results_checkbox_disabled_without_any_inn():
     """Ни личного ИНН, ни ИНН фирмы — добирать нечем, галочка серая."""
     run = {"id": 9, "rubric": "s", "city": "moscow", "status": "done",

моя/sales-finder/web/app.py

@@ -110,6 +110,7 @@ def _real_pipeline(rubric, city, queries):
                                      check_ip as dd_check_ip)
     from salesfinder.requisites import find_inn_candidates
     from salesfinder.egrul import enrich_director as egrul_enrich_director
+    from salesfinder import dedup as _dedup
     from salesfinder.cities import city_ru, keyso_base
     city_name = city_ru(city) or city   # русское имя для ДаДаты (слаг латиницей — для 2ГИС)
     base = keyso_base(city)             # региональная база Keys.so; None = регион не покрыт
@@ -178,6 +179,26 @@ def _real_pipeline(rubric, city, queries):
         неотличим от «фирм в городе нет» (инцидент 19.07.2026: 12 городов молча по нулям)."""
         print(f"[sales-finder] СТРАНИЦА НЕ СОБРАНА (рендерщик не ответил): {url}", flush=True)
 
+    def _card_site(firm):
+        """Верный сайт фирмы — из ЕЁ карточки 2ГИС (там реклама соседей не путает)."""
+        if not firm.card_url or not xfetch_key:
+            return None
+        from salesfinder.xfetch import render as xf_render
+        return site_from_card(xf_render(firm.card_url, xfetch_key))
+
+    def _reprocess(firm):
+        """Домен только что исправлен по карточке — сбрасываем всё, что было выведено по
+        ошибочному сайту (ИНН, телефон, директор, бюджет, каналы), и выводим заново."""
+        firm.inn = firm.ogrn = firm.address = firm.legal_name = firm.legal_status = None
+        firm.phone = firm.director = firm.director_inn = firm.director_post = None
+        firm.director_is_ip = False
+        firm.channels = []
+        firm.direct_keys = firm.direct_ads = firm.direct_budget = firm.direct_budget_min = 0
+        firm.evidence = []
+        scan_ad(firm)          # новая главная: телефон, ИНН-с-главной, каналы, home_html
+        fetch_budget(firm)     # бюджет Директа по новому домену
+        enrich(firm)           # ИНН из реквизитов + ДаДата + ЕГРЮЛ по новому домену
+
     deps = dict(
         collect_2gis=lambda c, qs: collect_two_gis_parallel(
             c, qs, fetch=xf_fetcher(xfetch_key, on_fail=_page_lost), workers=WORKERS),
@@ -186,6 +207,8 @@ def _real_pipeline(rubric, city, queries):
         fetch_budget=fetch_budget,
         scan_ad=scan_ad,
         enrich=enrich,
+        dedupe=lambda firms: _dedup.fix_and_collapse(
+            firms, {"card_site": _card_site, "reprocess": _reprocess}),
         log=lambda m: print(f"[sales-finder] {m}", flush=True),
     )
     return run_pipeline(rubric, city, queries, deps, workers=WORKERS, city_label=city_name)
@@ -327,9 +350,10 @@ def run_view(rid: int):
     return HTMLResponse(ui.results_html(r, managers=_portal_managers_safe()))
 
 @app.post("/view/{rid}/contacts")
-def collect_contacts(rid: int, sel: list = Form(default=[])):
+def collect_contacts(rid: int, sel: list = Form(default=[]), refresh: str = Form(default="")):
     """Сбор контактов по отмеченным галочками фирмам: по личному ИНН директора → телефон/почта.
-    Персональные данные — только по явной отметке. Домен/ключ API — в secrets/inn_fl_url.txt."""
+    Персональные данные — только по явной отметке. Домен/ключ API — в secrets/inn_fl_url.txt.
+    refresh — «обновить»: пересобрать заново даже тех, кого уже собирали (кэш не читаем)."""
     from salesfinder.models import Firm
     from salesfinder import contactflow
     r = STORE.get(rid)
@@ -347,12 +371,14 @@ def collect_contacts(rid: int, sel: list = Form(default=[])):
     firms = [Firm(**row) for row in r["firms"]]
     picked = sorted({int(s) for s in sel if str(s).isdigit()})
     dadata_keys, smsc_creds = _dadata_clean_keys(), _smsc_creds()   # секреты — раз на весь пакет
+    from salesfinder.contactcache import ContactCache
+    cache = ContactCache()   # не платим дважды за одного директора (в т.ч. из прошлых сборов)
     for i in picked:                                  # ПОСЛЕДОВАТЕЛЬНО: чужой API, лимиты неизвестны
         if not (0 <= i < len(firms)):
             continue
         f = firms[i]
         deps = {"lookup": CONTACT_LOOKUP, "check": PHONE_CHECK, "tpl": tpl,
-                "enrich": DIRECTOR_ENRICH,
+                "enrich": DIRECTOR_ENRICH, "cache": cache, "refresh": bool(refresh),
                 "dadata_keys": dadata_keys, "smsc": smsc_creds}
         contactflow.collect_for_firm(f, deps)
     STORE.save_result(rid, firms)

моя/sales-finder/web/bulk.py

@@ -52,8 +52,9 @@ def _runs(ids):
 
 @router.post("/bulk/confirm", response_class=HTMLResponse)
 def confirm(ids: list = Form(default=[]), action: str = Form(default=""),
-            manager_id: str = Form(default="")):
-    """Экран «вы уверены» с настоящими числами. Ничего не делает."""
+            manager_id: str = Form(default=""), refresh: str = Form(default="")):
+    """Экран «вы уверены» с настоящими числами. Ничего не делает.
+    refresh — «обновить»: пересобрать телефоны заново, даже уже собранных."""
     if action not in ACTIONS or not ids:
         return RedirectResponse(url=_u("/"), status_code=303)
     runs = _runs(ids)
@@ -68,10 +69,15 @@ def confirm(ids: list = Form(default=[]), action: str = Form(default=""),
         q = (f"Удалить {plural(len(runs), 'список', 'списка', 'списков')} и "
              f"{plural(total, 'фирму', 'фирмы', 'фирм')} в них? Это насовсем.")
     elif action == "contacts":
-        todo, skip = bulkops.split_for_contacts(hot)
-        q = (f"Собрать телефоны по {plural(len(todo), 'горячей фирме', 'горячим фирмам', 'горячим фирмам')} "
-             f"из {plural(len(runs), 'списка', 'списков', 'списков')}?"
-             + (f" Ещё {len(skip)} пропустим — телефон уже собран." if skip else ""))
+        if refresh:
+            q = (f"Пересобрать телефоны по {plural(len(hot), 'горячей фирме', 'горячим фирмам', 'горячим фирмам')} "
+                 f"из {plural(len(runs), 'списка', 'списков', 'списков')}? "
+                 f"Обновление — соберём заново даже уже собранных, это платный запрос.")
+        else:
+            todo, skip = bulkops.split_for_contacts(hot)
+            q = (f"Собрать телефоны по {plural(len(todo), 'горячей фирме', 'горячим фирмам', 'горячим фирмам')} "
+                 f"из {plural(len(runs), 'списка', 'списков', 'списков')}?"
+                 + (f" Ещё {len(skip)} пропустим — телефон уже собран." if skip else ""))
     elif action == "warmup":
         todo, skip = bulkops.split_for_warmup(hot)
         ready = [f for f in todo if (f.contact_phone or "").strip()]
@@ -88,6 +94,8 @@ def confirm(ids: list = Form(default=[]), action: str = Form(default=""),
     hidden = "".join(f'<input type=hidden name=ids value="{int(i)}">' for i in ids)
     if manager_id:
         hidden += f'<input type=hidden name=manager_id value="{ui._esc(manager_id)}">'
+    if refresh:
+        hidden += '<input type=hidden name=refresh value="1">'
     body = (f'<div class=card><h1>{ui._esc(ACTIONS[action])}</h1>'
             f'<p class=lead>{ui._esc(q)}</p>'
             f'<form method=post action="{ui._esc(_u("/bulk/" + action))}">{hidden}'
@@ -105,18 +113,23 @@ def bulk_delete(ids: list = Form(default=[])):
 
 
 @router.post("/bulk/contacts", response_class=HTMLResponse)
-def bulk_contacts(ids: list = Form(default=[])):
+def bulk_contacts(ids: list = Form(default=[]), refresh: str = Form(default="")):
     """Собрать личные телефоны по горячим фирмам пачки списков. Платный запрос на
     каждую фирму — уже собранных не трогаем. Номера размечает та же проверка,
-    что и в одиночном сборе (общая contactflow.collect_for_firm)."""
+    что и в одиночном сборе (общая contactflow.collect_for_firm).
+    refresh — «обновить»: пересобрать заново даже уже собранных (кэш не читаем)."""
     from salesfinder import contactflow
+    from salesfinder.contactcache import ContactCache
     from web.app import (STORE, CONTACT_LOOKUP, PHONE_CHECK, DIRECTOR_ENRICH,
                          _contact_template, _dadata_clean_keys, _smsc_creds)
     tpl = _contact_template()
     if not tpl:
         return HTMLResponse(ui.page("Нет доступа к сервису контактов", _no_service_body()))
+    # Кэш по личному ИНН директора: один человек бывает директором нескольких фирм —
+    # платим за него ОДИН раз, дальше берём готовое (в т.ч. из прошлых списков).
+    cache = ContactCache()
     deps_base = {"lookup": CONTACT_LOOKUP, "check": PHONE_CHECK, "tpl": tpl,
-                 "enrich": DIRECTOR_ENRICH,
+                 "enrich": DIRECTOR_ENRICH, "cache": cache, "refresh": bool(refresh),
                  "dadata_keys": _dadata_clean_keys(), "smsc": _smsc_creds()}
     runs = STORE.get_many(ids)
     done = skipped = failed = no_inn = 0
@@ -125,7 +138,10 @@ def bulk_contacts(ids: list = Form(default=[])):
             continue
         firms = bulkops.as_firms(r.get("firms") or [])
         hot = [f for f in firms if (f.hotness or 0) >= bulkops.HOT]
-        todo, skip = bulkops.split_for_contacts(hot)
+        if refresh:
+            todo, skip = hot, []          # обновляем ВСЕХ горячих, даже уже собранных
+        else:
+            todo, skip = bulkops.split_for_contacts(hot)
         skipped += len(skip)
         for f in todo:
             if contactflow.collect_for_firm(f, deps_base):
@@ -139,6 +155,7 @@ def bulk_contacts(ids: list = Form(default=[])):
         f"Собрано {done}, пропущено {skipped} — телефон уже был"
         + (f", не нашлось {failed}" if failed else "")
         + (f", без ИНН директора {no_inn} — искать не по чему" if no_inn else "")
+        + (f", повторный директор {cache.hits} — не платили дважды" if cache.hits else "")
         + "." + _lost_note(ids, runs))))
 
 

моя/sales-finder/web/ui.py

@@ -213,6 +213,9 @@ def landing_html(catalog, regions, recent, managers=None):
         '<div class="mt" id=bulkpanel>'
         f'<select name=manager_id id=bulkmgr>{manager_opts}</select> '
         '<button class="btn" name=action value="contacts" disabled>Собрать телефоны по горячим</button> '
+        '<label class=dim style="margin:0 8px" title="пересобрать телефоны заново, даже уже '
+        'собранных — на случай смены телефона (платный запрос)">'
+        '<input type=checkbox name=refresh value="1"> обновить</label> '
         '<button class="btn" name=action value="warmup" disabled>Загрузить в прогрев</button> '
         '<button class="btn ghost" name=action value="assign" disabled>Отдать менеджеру горячих</button> '
         '<button class="btn ghost" name=action value="delete" disabled>Удалить списки</button>'
@@ -394,6 +397,9 @@ def results_html(run, managers=None):
         rows.append(
             f'<tr><td>{chk}</td><td class=dim>{i}</td>'
             f'<td class=firm>{_esc(f.get("name") or "—")}'
+            + (f' <span class="badge b-grey" title="ещё карточки той же компании '
+               f'(тот же ИНН) схлопнуты в эту строку">+{int(f.get("merged_count") or 0)}</span>'
+               if (f.get("merged_count") or 0) else '')
             + (f'<small>{_esc(legal)}</small>' if legal else '') + '</td>'
             f'<td>{_esc(f.get("city") or "—")}</td>'
             f'<td class=mono>{_esc(f.get("phone") or "—")}</td>'
@@ -460,6 +466,9 @@ def results_html(run, managers=None):
     # клиентов» был мёртв 3 дня).
     collect = (f'<form method=post action="{_u(f"/view/{_esc(rid)}/contacts")}">'
                f'<div style="margin:0 0 12px"><button class=btn>Собрать контакты по отмеченным</button>'
+               f'<label class=dim style="margin-left:10px" title="пересобрать заново, даже если '
+               f'этот директор уже собран — на случай смены телефона (платный запрос)">'
+               f'<input type=checkbox name=refresh value="1"> обновить</label>'
                f'{assign}{warmup}'
                f'<span class=dim style="margin-left:12px">по ИНН → все телефоны и e-mail '
                f'директора; 2 платных запроса на фирму</span></div>'