fix: устранены все найденные аудитом баги и тихие падения

- SQL injection паттерн → параметризованные запросы во всех местах
- except: pass/continue → logger.warning() везде, ничего не тонет молча
- WAL mode + индекс domain_dedup_key в database.py
- try/finally для conn в main.py, утечка соединения устранена
- backoff 30с при 403/429 от Rusprofile/ЕГРЮЛ
- ликвидированные компании → egrul_status="liquidated"
- max_candidates в contacts_finder считает только реальных кандидатов
- DB_PATH абсолютный (Path(__file__).parent), HH_PAUSE_BETWEEN_QUERIES в config
- HH_SIGNAL_QUERIES дубль убран из launcher.py → импорт из config
- path traversal защита в egrul_enricher debug_dump_html

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Aks
2026-06-09 13:19:52 +03:00
parent f78f35fb3f
commit 98309dcc96
9 changed files with 208 additions and 186 deletions
+10
View File
@@ -8,6 +8,7 @@ Streamlit перезапускает скрипт на каждое действ
держать не имеет смысла. держать не имеет смысла.
""" """
import json import json
import logging
import sqlite3 import sqlite3
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
@@ -15,6 +16,8 @@ from typing import Any
import pandas as pd import pandas as pd
logger = logging.getLogger(__name__)
def _conn(db_path: Path | str) -> sqlite3.Connection: def _conn(db_path: Path | str) -> sqlite3.Connection:
conn = sqlite3.connect(str(db_path)) conn = sqlite3.connect(str(db_path))
@@ -311,10 +314,17 @@ def add_lead_manual(db_path, data: dict) -> int:
conn.rollback() conn.rollback()
msg = str(e).lower() msg = str(e).lower()
if "inn" in msg: if "inn" in msg:
logger.warning(f"[db_layer] add_lead_manual: дубль по ИНН ({data.get('inn')}): {e}")
raise ValueError(f"Компания с таким ИНН уже есть в базе ({data.get('inn')}).") from e raise ValueError(f"Компания с таким ИНН уже есть в базе ({data.get('inn')}).") from e
if "phone" in msg: if "phone" in msg:
logger.warning(f"[db_layer] add_lead_manual: дубль по телефону: {e}")
raise ValueError(f"Компания с таким телефоном уже есть в базе.") from e raise ValueError(f"Компания с таким телефоном уже есть в базе.") from e
logger.warning(f"[db_layer] add_lead_manual: IntegrityError: {e}")
raise ValueError(f"Не удалось добавить (дубль): {e}") from e raise ValueError(f"Не удалось добавить (дубль): {e}") from e
except Exception as e:
conn.rollback()
logger.warning(f"[db_layer] add_lead_manual: неожиданная ошибка: {e}")
raise
finally: finally:
conn.close() conn.close()
+5 -1
View File
@@ -2,6 +2,7 @@
Все настройки в одном месте. Меняем здесь — отражается во всех парсерах. Все настройки в одном месте. Меняем здесь — отражается во всех парсерах.
""" """
from pathlib import Path
# ─────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────
# Города и их Yandex.Maps ID # Города и их Yandex.Maps ID
@@ -113,9 +114,12 @@ MAX_CARDS_PER_CATEGORY = 100 # сколько карточек открыват
# ─────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────
# Пути # Пути
# ─────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────
DB_PATH = "leads.db" DB_PATH = Path(__file__).parent / "leads.db"
EXPORT_DIR = "exports" EXPORT_DIR = "exports"
# Пауза между HH-запросами (секунды) — задаётся здесь и используется в run_hh
HH_PAUSE_BETWEEN_QUERIES = 3
# ─────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────
# Скоринг лидов v5 — «решаемая нами боль» (шкала 0-10) # Скоринг лидов v5 — «решаемая нами боль» (шкала 0-10)
# ─────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────
+9 -3
View File
@@ -4,6 +4,7 @@
Сама занимается дедупликацией: ИНН > телефон > домен. Сама занимается дедупликацией: ИНН > телефон > домен.
""" """
import json import json
import logging
import sqlite3 import sqlite3
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
@@ -12,6 +13,8 @@ from typing import Optional
import config import config
from normalization import phone_dedup_key, normalize_domain from normalization import phone_dedup_key, normalize_domain
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────
# Схема таблиц # Схема таблиц
# ─────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────────
@@ -119,6 +122,7 @@ CREATE INDEX IF NOT EXISTS idx_leads_score ON leads(score DESC);
CREATE INDEX IF NOT EXISTS idx_leads_source ON leads(source); CREATE INDEX IF NOT EXISTS idx_leads_source ON leads(source);
CREATE INDEX IF NOT EXISTS idx_leads_outreach ON leads(outreach_status); CREATE INDEX IF NOT EXISTS idx_leads_outreach ON leads(outreach_status);
CREATE INDEX IF NOT EXISTS idx_leads_has_website ON leads(has_website); CREATE INDEX IF NOT EXISTS idx_leads_has_website ON leads(has_website);
CREATE INDEX IF NOT EXISTS idx_leads_domain ON leads(domain_dedup_key);
CREATE TABLE IF NOT EXISTS sources_log ( CREATE TABLE IF NOT EXISTS sources_log (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -208,6 +212,7 @@ def get_connection(db_path: str = "leads.db") -> sqlite3.Connection:
"""Открыть соединение с включённым row_factory (sqlite3.Row для dict-доступа).""" """Открыть соединение с включённым row_factory (sqlite3.Row для dict-доступа)."""
conn = sqlite3.connect(db_path) conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn return conn
@@ -748,8 +753,9 @@ def cleanup_bad_director_names(conn: sqlite3.Connection) -> int:
"Производство", "Услуги", "Работ", "Производство", "Услуги", "Работ",
] ]
# Совпадение если первое слово в director_name — должность # Совпадение если первое слово в director_name — должность
where_parts = [f"director_name LIKE '{m}%'" for m in bad_markers] where_parts = ["director_name LIKE ?"] * len(bad_markers)
where_clause = " OR ".join(where_parts) where_clause = " OR ".join(where_parts)
params = [f"{m}%" for m in bad_markers]
sql = f""" sql = f"""
UPDATE leads UPDATE leads
SET director_name = NULL, SET director_name = NULL,
@@ -757,7 +763,7 @@ def cleanup_bad_director_names(conn: sqlite3.Connection) -> int:
egrul_status = NULL egrul_status = NULL
WHERE {where_clause} WHERE {where_clause}
""" """
cursor = conn.execute(sql) cursor = conn.execute(sql, params)
conn.commit() conn.commit()
return cursor.rowcount return cursor.rowcount
@@ -878,7 +884,7 @@ def get_stats(conn: sqlite3.Connection) -> dict:
).fetchall()) ).fetchall())
threshold = config.HOT_LEAD_THRESHOLD threshold = config.HOT_LEAD_THRESHOLD
hot = conn.execute( hot = conn.execute(
f"SELECT COUNT(*) FROM leads WHERE score >= {int(threshold)}" "SELECT COUNT(*) FROM leads WHERE score >= ?", (int(threshold),)
).fetchone()[0] ).fetchone()[0]
with_phone = conn.execute( with_phone = conn.execute(
"SELECT COUNT(*) FROM leads WHERE phone_primary IS NOT NULL" "SELECT COUNT(*) FROM leads WHERE phone_primary IS NOT NULL"
+4 -2
View File
@@ -303,7 +303,8 @@ def find_company_website(
} }
seen: set[str] = set() seen: set[str] = set()
max_candidates = 6 # проверим не больше 6 разных сайтов max_candidates = 6 # проверим не больше 6 реальных (не агрегаторных) кандидатов
real_candidates = 0
for query in queries: for query in queries:
try: try:
@@ -324,7 +325,8 @@ def find_company_website(
seen.add(host) seen.add(host)
if not _is_company_site(found_url): if not _is_company_site(found_url):
continue continue
if len(seen) > max_candidates: real_candidates += 1
if real_candidates > max_candidates:
break break
root = f"{urlparse(found_url).scheme}://{host}" root = f"{urlparse(found_url).scheme}://{host}"
+6 -2
View File
@@ -52,7 +52,7 @@ def _api_key() -> Optional[str]:
load_dotenv() load_dotenv()
key = os.environ.get("DADATA_API_KEY") key = os.environ.get("DADATA_API_KEY")
except ImportError: except ImportError:
pass logger.warning("[dadata] python-dotenv не установлен, .env не загружен")
return key return key
@@ -282,8 +282,12 @@ def enrich_via_dadata(
best = d best = d
break break
if not best: if not best:
# Все ликвидированы — берём первого, не записываем в БД # Все ликвидированы — берём первого, логируем, выставляем статус
logger.warning(f"[dadata] Все совпадения ликвидированы для '{name}'")
best = suggestions[0].get("data") or {} best = suggestions[0].get("data") or {}
parsed = _parse_party(best)
parsed["egrul_status"] = "liquidated"
return parsed
parsed = _parse_party(best) parsed = _parse_party(best)
return parsed return parsed
+20 -9
View File
@@ -7,7 +7,7 @@
Стратегия запросов: Стратегия запросов:
- Случайный User-Agent - Случайный User-Agent
- timeout 10 сек - timeout 10 сек
- При 403/429 — пропускаем (логируем как 'error') - При 403/429 — пауза 30 сек + пропускаем (логируем как 'error')
- Пауза между лидами в run_egrul_enrichment, не здесь - Пауза между лидами в run_egrul_enrichment, не здесь
Ограничения: Ограничения:
@@ -18,6 +18,7 @@
""" """
import logging import logging
import re import re
import time
from datetime import datetime from datetime import datetime
from typing import Optional from typing import Optional
from urllib.parse import quote from urllib.parse import quote
@@ -379,7 +380,8 @@ def enrich_egrul_by_inn(
allow_redirects=True, allow_redirects=True,
) )
if resp.status_code in (403, 429): if resp.status_code in (403, 429):
logger.warning(f" Rusprofile blocked us ({resp.status_code}) для ИНН {inn}") logger.warning(f"[egrul] Rusprofile вернул {resp.status_code}, пауза 30 сек (ИНН {inn})")
time.sleep(30)
return result return result
if resp.status_code != 200: if resp.status_code != 200:
return result return result
@@ -610,10 +612,11 @@ def enrich_egrul(
) )
if resp.status_code in (403, 429): if resp.status_code in (403, 429):
logger.warning(f" Rusprofile blocked us ({resp.status_code}) для '{name}'") logger.warning(f"[egrul] Rusprofile вернул {resp.status_code}, пауза 30 сек ('{name}')")
time.sleep(30)
return result return result
if resp.status_code != 200: if resp.status_code != 200:
logger.debug(f" Rusprofile вернул {resp.status_code} для '{name}'") logger.warning(f"[egrul] Rusprofile вернул {resp.status_code} для '{name}'")
return result return result
# Если редирект на /id/N или /ip/N — Rusprofile уверен в матче. # Если редирект на /id/N или /ip/N — Rusprofile уверен в матче.
@@ -657,12 +660,20 @@ def enrich_egrul(
# Debug: сохранить HTML для отладки regex'ов # Debug: сохранить HTML для отладки regex'ов
if debug_dump_html: if debug_dump_html:
from pathlib import Path
dump_path = Path(debug_dump_html).resolve()
try: try:
with open(debug_dump_html, "w", encoding="utf-8") as f: dump_path.relative_to(Path.cwd())
f.write(html) except ValueError:
logger.info(f" [debug] HTML сохранён в {debug_dump_html}") logger.warning(f"[egrul] debug_dump_html путь вне CWD, пропускаю: {debug_dump_html}")
except Exception as e: debug_dump_html = None
logger.warning(f" [debug] Не удалось сохранить HTML: {e}") if debug_dump_html:
try:
with open(dump_path, "w", encoding="utf-8") as f:
f.write(html)
logger.info(f" [debug] HTML сохранён в {dump_path}")
except Exception as e:
logger.warning(f" [debug] Не удалось сохранить HTML: {e}")
# Извлекаем все поля (один общий парсер для by_name и by_inn). # Извлекаем все поля (один общий парсер для by_name и by_inn).
# Передаём company_name=name_for_search (без HR-суффиксов) чтобы # Передаём company_name=name_for_search (без HR-суффиксов) чтобы
+6 -1
View File
@@ -13,6 +13,7 @@ False positives возможны (например, упоминание "yclien
""" """
import logging import logging
import re import re
import time
from datetime import datetime from datetime import datetime
from typing import Optional from typing import Optional
@@ -273,6 +274,7 @@ def analyze_website(url: str, timeout: float = 6.0) -> dict:
timeout=(timeout, timeout), verify=False, stream=False, timeout=(timeout, timeout), verify=False, stream=False,
) )
if sub_resp.status_code != 200: if sub_resp.status_code != 200:
time.sleep(0.5)
continue continue
sub_html = sub_resp.text[:MAX_HTML_BYTES] sub_html = sub_resp.text[:MAX_HTML_BYTES]
sub_inn, sub_ogrn, sub_kpp = _extract_inn_ogrn_kpp(sub_html) sub_inn, sub_ogrn, sub_kpp = _extract_inn_ogrn_kpp(sub_html)
@@ -281,8 +283,11 @@ def analyze_website(url: str, timeout: float = 6.0) -> dict:
ogrn_val = ogrn_val or sub_ogrn ogrn_val = ogrn_val or sub_ogrn
kpp_val = kpp_val or sub_kpp kpp_val = kpp_val or sub_kpp
logger.debug(f" ИНН найден на {path}: {inn_val}") logger.debug(f" ИНН найден на {path}: {inn_val}")
time.sleep(0.5)
break break
except Exception: time.sleep(0.5)
except Exception as e:
logger.warning(f"[website_analyzer] подстраница {path} сайта {url}: {e}")
continue continue
if inn_val: if inn_val:
result["inn"] = inn_val result["inn"] = inn_val
+1 -18
View File
@@ -7,6 +7,7 @@ import subprocess
import sys import sys
import config # для синхронизации дефолтных категорий (фокус-ЦА) с CLI import config # для синхронизации дефолтных категорий (фокус-ЦА) с CLI
from config import HH_SIGNAL_QUERIES
try: try:
import questionary import questionary
@@ -224,24 +225,6 @@ MOSCOW_DISTRICTS = [
] ]
# ── HH signal-запросы (из config) ────────────────────────────────────
HH_SIGNAL_QUERIES = [
"оператор ПК",
"оператор колл-центра",
"оператор технической поддержки",
"менеджер чата",
"менеджер по продажам без CRM",
"помощник руководителя",
"ассистент руководителя",
"офис-менеджер",
"администратор записи",
"администратор салона красоты",
"администратор клиники",
"ресепшн",
"бухгалтер 1С",
"помощник бухгалтера",
]
def ask(fn, *args, **kwargs): def ask(fn, *args, **kwargs):
result = fn(*args, **kwargs, style=STYLE).ask() result = fn(*args, **kwargs, style=STYLE).ask()
+147 -150
View File
@@ -205,8 +205,7 @@ def run_hh(conn, queries: list[str], city: str, max_pages: int = None) -> list[i
# Пауза между запросами (anti-rate-limit) # Пауза между запросами (anti-rate-limit)
if idx < len(queries): if idx < len(queries):
import time time.sleep(config.HH_PAUSE_BETWEEN_QUERIES)
time.sleep(3)
return run_ids return run_ids
@@ -272,7 +271,7 @@ def run_enrichment(conn, limit: int | None = None) -> int:
conn.execute(f"UPDATE leads SET {sets} WHERE id = ?", vals) conn.execute(f"UPDATE leads SET {sets} WHERE id = ?", vals)
conn.commit() conn.commit()
except Exception as e: except Exception as e:
logger.debug(f" ⚠ update inn/ogrn: {e}") logger.warning(f" ⚠ update inn/ogrn: {e}")
# Если нашли ИНН на сайте, сразу идём в Rusprofile by-inn за директором # Если нашли ИНН на сайте, сразу идём в Rusprofile by-inn за директором
rusprofile_note = "" rusprofile_note = ""
@@ -302,7 +301,7 @@ def run_enrichment(conn, limit: int | None = None) -> int:
director = egrul_info.get("director_name") or "" director = egrul_info.get("director_name") or ""
rusprofile_note = f" +ЕГРЮЛ:директор={director[:30]}" rusprofile_note = f" +ЕГРЮЛ:директор={director[:30]}"
except Exception as e: except Exception as e:
logger.debug(f" ⚠ Rusprofile by-inn: {e}") logger.warning(f" ⚠ Rusprofile by-inn: {e}")
# Пересчитать score (теперь с Tier 2 данными + ИНН с сайта) # Пересчитать score (теперь с Tier 2 данными + ИНН с сайта)
full_lead = dict(conn.execute("SELECT * FROM leads WHERE id = ?", (lead_id,)).fetchone()) full_lead = dict(conn.execute("SELECT * FROM leads WHERE id = ?", (lead_id,)).fetchone())
@@ -888,7 +887,7 @@ def main():
# Раньше флаг ставился всегда → `--full --source yandex` гонял HH-обогащение по # Раньше флаг ставился всегда → `--full --source yandex` гонял HH-обогащение по
# ВСЕЙ базе HH-лидов без сайта (~час впустую). Теперь — только при source hh/all # ВСЕЙ базе HH-лидов без сайта (~час впустую). Теперь — только при source hh/all
# (как и обещает README: «+ --hh-enrich-websites если source=hh»). # (как и обещает README: «+ --hh-enrich-websites если source=hh»).
args.hh_enrich_websites = args.source in ("hh", "all") args.hh_enrich_websites = args.hh_enrich_websites or args.source in ("hh", "all")
args.find_sites = True # ⬅ ищем сайты для лидов где website пуст (DDG fallback) args.find_sites = True # ⬅ ищем сайты для лидов где website пуст (DDG fallback)
args.enrich = True # → website-analyzer достаёт email/phone со страниц args.enrich = True # → website-analyzer достаёт email/phone со страниц
args.enrich_egrul = True args.enrich_egrul = True
@@ -914,156 +913,154 @@ def main():
# Инициализация БД # Инициализация БД
init_db(config.DB_PATH) init_db(config.DB_PATH)
conn = get_connection(config.DB_PATH) conn = get_connection(config.DB_PATH)
try:
# Только статистика — выходим сразу
if args.stats:
stats = get_stats(conn)
print("\n📊 СТАТИСТИКА:")
for k, v in stats.items():
print(f" {k}: {v}")
return
# Только статистика — выходим сразу # Cleanup кривых ФИО директоров (одноразовая операция)
if args.stats: if args.cleanup_directors:
cleared = cleanup_bad_director_names(conn)
logger.info(f"🧹 Очищено директоров с должностями вместо ФИО: {cleared}")
# Не выходим — даём дальше pipeline отработать (например --enrich-egrul)
# Одноразовая миграция категорий: lead.category = sources_log.query
if args.fix_categories:
fixed = fix_categories_from_runs(conn)
logger.info(f"🏷️ Категории переписаны на поисковый запрос у {fixed} лидов.")
# Сброс site_checked_at для перепрогона enricher на уже обогащённых лидах
if args.rescan_sites:
cursor = conn.execute(
"UPDATE leads SET site_checked_at = NULL WHERE site_checked_at IS NOT NULL"
)
conn.commit()
logger.info(f"🔁 Сброшено site_checked_at у {cursor.rowcount} лидов — следующий --enrich пройдёт по ним заново.")
# Если не указан ни один из шагов — показать help
if not (
args.source
or args.enrich
or args.enrich_egrul
or args.enrich_finance
or args.rescore
or args.export
or args.cleanup_directors
or args.fix_categories
or args.rescan_sites
or args.hh_enrich_websites
or args.find_sites
or args.export_master
or args.export_run is not None
):
parser.print_help()
sys.exit(1)
# Собираем run_id всех прогонов этой сессии — для auto-export CSV по прогонам
collected_run_ids: list[int] = []
# ── Pipeline — шаги выполняются последовательно ─────────────────
# Шаг 1: Парсинг (если указан --source)
if args.source:
if args.category:
categories = [c.strip() for c in args.category.split(",") if c.strip()]
else:
categories = config.CATEGORIES
if args.source in ("yandex", "all"):
collected_run_ids.extend(
run_yandex(conn, categories, args.city, args.limit, district=args.district)
)
if args.source in ("hh", "all"):
# Для HH "категории" — это signal-запросы (оператор / администратор / etc.)
# Если пользователь явно указал --category — используем его список.
# Иначе — стандартный config.HH_SIGNAL_QUERIES.
hh_queries = categories if args.category else config.HH_SIGNAL_QUERIES
collected_run_ids.extend(
run_hh(conn, hh_queries, args.city, max_pages=args.limit if args.category else None)
)
if args.source in ("2gis", "all"):
logger.warning("⚠️ Парсер 2GIS пока не реализован (Phase 1.4)")
if args.source in ("vk", "all"):
logger.warning("⚠️ Парсер VK пока не реализован (Phase 1.5)")
# --limit относится только к парсингу (карточек на категорию).
# Enrichment всегда обрабатывает ВСЕХ непроверенных, чтобы не оставлять хвосты.
# Шаг 1.5: HH employer-page enrichment (дозаполняет website у HH-лидов).
# Должен идти ДО Tier 2 (--enrich), иначе у HH-лидов нет сайта и Tier 2
# их пропустит → email не соберутся.
if args.hh_enrich_websites:
run_hh_websites(conn, limit=args.hh_limit)
# Шаг 1.6: Поиск сайтов через DDG (для лидов без website).
# Должен идти ДО --enrich по той же причине что и hh_enrich_websites:
# сначала найти сайт → потом website-analyzer вытащит email/phone.
#
# ВАЖНО: ограничиваем поиск только лидами ТЕКУЩЕЙ сессии (collected_run_ids).
# Иначе пойдём по всей БД (2500+ лидов = час+ DDG-запросов).
# Если --find-sites вызван БЕЗ --source (например, для добивки старых лидов),
# тогда run_ids=None и идём по всем (медленно, но осознанно).
if args.find_sites:
run_find_sites(
conn,
limit=args.find_sites_limit,
run_ids=collected_run_ids if collected_run_ids else None,
source_filter=args.find_sites_source,
)
# Шаг 2: Enrichment сайтов (если указан --enrich)
if args.enrich:
run_enrichment(conn, limit=None)
# Шаг 3: Enrichment ЕГРЮЛ (если указан --enrich-egrul)
if args.enrich_egrul:
run_egrul_enrichment(conn, limit=None)
# Шаг 3.5: Финансы — сотрудники + оборот по ООО (если указан --enrich-finance)
if args.enrich_finance:
run_financials(conn, limit=None)
# Шаг 4: Rescore (если указан --rescore)
if args.rescore:
run_rescore(conn)
# Финальная статистика
stats = get_stats(conn) stats = get_stats(conn)
print("\n📊 СТАТИСТИКА:") print("\n" + "" * 60)
print("📊 РЕЗУЛЬТАТ:")
for k, v in stats.items(): for k, v in stats.items():
print(f" {k}: {v}") print(f" {k}: {v}")
print("" * 60)
# Шаг 5: Экспорт (если указан --export)
# Новое поведение: если в сессии были прогоны парсера → отдельный CSV
# каждого прогона в exports/YYYY-MM/ с шапкой-метаданными.
# Старое поведение (backward compat): --export без парсинга → плоский CSV
# по min_score в exports/YYYY-MM/leads_<timestamp>.csv.
if args.export:
if collected_run_ids:
print(f"\n📤 Экспорт {len(collected_run_ids)} прогонов сессии:")
for run_id in collected_run_ids:
export_run(config.DB_PATH, run_id)
# Сводный файл «всё, что собрано за день» в папку дня
export_day(config.DB_PATH)
else:
export_to_csv(config.DB_PATH, min_score=args.min_score)
# Шаг 6: master-snapshot всей БД (--export-master)
if args.export_master:
export_master(config.DB_PATH)
# Шаг 7: пересоздать CSV конкретного прогона (--export-run N)
if args.export_run is not None:
export_run(config.DB_PATH, args.export_run)
finally:
conn.close() conn.close()
return
# Cleanup кривых ФИО директоров (одноразовая операция)
if args.cleanup_directors:
cleared = cleanup_bad_director_names(conn)
logger.info(f"🧹 Очищено директоров с должностями вместо ФИО: {cleared}")
# Не выходим — даём дальше pipeline отработать (например --enrich-egrul)
# Одноразовая миграция категорий: lead.category = sources_log.query
if args.fix_categories:
fixed = fix_categories_from_runs(conn)
logger.info(f"🏷️ Категории переписаны на поисковый запрос у {fixed} лидов.")
# Сброс site_checked_at для перепрогона enricher на уже обогащённых лидах
if args.rescan_sites:
cursor = conn.execute(
"UPDATE leads SET site_checked_at = NULL WHERE site_checked_at IS NOT NULL"
)
conn.commit()
logger.info(f"🔁 Сброшено site_checked_at у {cursor.rowcount} лидов — следующий --enrich пройдёт по ним заново.")
# Если не указан ни один из шагов — показать help
if not (
args.source
or args.enrich
or args.enrich_egrul
or args.enrich_finance
or args.rescore
or args.export
or args.cleanup_directors
or args.fix_categories
or args.rescan_sites
or args.hh_enrich_websites
or args.find_sites
or args.export_master
or args.export_run is not None
):
parser.print_help()
conn.close()
sys.exit(1)
# Собираем run_id всех прогонов этой сессии — для auto-export CSV по прогонам
collected_run_ids: list[int] = []
# ── Pipeline — шаги выполняются последовательно ─────────────────
# Шаг 1: Парсинг (если указан --source)
if args.source:
if args.category:
categories = [c.strip() for c in args.category.split(",") if c.strip()]
else:
categories = config.CATEGORIES
if args.source in ("yandex", "all"):
collected_run_ids.extend(
run_yandex(conn, categories, args.city, args.limit, district=args.district)
)
if args.source in ("hh", "all"):
# Для HH "категории" — это signal-запросы (оператор / администратор / etc.)
# Если пользователь явно указал --category — используем его список.
# Иначе — стандартный config.HH_SIGNAL_QUERIES.
hh_queries = categories if args.category else config.HH_SIGNAL_QUERIES
collected_run_ids.extend(
run_hh(conn, hh_queries, args.city, max_pages=args.limit if args.category else None)
)
if args.source in ("2gis", "all"):
logger.warning("⚠️ Парсер 2GIS пока не реализован (Phase 1.4)")
if args.source in ("vk", "all"):
logger.warning("⚠️ Парсер VK пока не реализован (Phase 1.5)")
# --limit относится только к парсингу (карточек на категорию).
# Enrichment всегда обрабатывает ВСЕХ непроверенных, чтобы не оставлять хвосты.
# Шаг 1.5: HH employer-page enrichment (дозаполняет website у HH-лидов).
# Должен идти ДО Tier 2 (--enrich), иначе у HH-лидов нет сайта и Tier 2
# их пропустит → email не соберутся.
if args.hh_enrich_websites:
run_hh_websites(conn, limit=args.hh_limit)
# Шаг 1.6: Поиск сайтов через DDG (для лидов без website).
# Должен идти ДО --enrich по той же причине что и hh_enrich_websites:
# сначала найти сайт → потом website-analyzer вытащит email/phone.
#
# ВАЖНО: ограничиваем поиск только лидами ТЕКУЩЕЙ сессии (collected_run_ids).
# Иначе пойдём по всей БД (2500+ лидов = час+ DDG-запросов).
# Если --find-sites вызван БЕЗ --source (например, для добивки старых лидов),
# тогда run_ids=None и идём по всем (медленно, но осознанно).
if args.find_sites:
run_find_sites(
conn,
limit=args.find_sites_limit,
run_ids=collected_run_ids if collected_run_ids else None,
source_filter=args.find_sites_source,
)
# Шаг 2: Enrichment сайтов (если указан --enrich)
if args.enrich:
run_enrichment(conn, limit=None)
# Шаг 3: Enrichment ЕГРЮЛ (если указан --enrich-egrul)
if args.enrich_egrul:
run_egrul_enrichment(conn, limit=None)
# Шаг 3.5: Финансы — сотрудники + оборот по ООО (если указан --enrich-finance)
if args.enrich_finance:
run_financials(conn, limit=None)
# Шаг 4: Rescore (если указан --rescore)
if args.rescore:
run_rescore(conn)
# Финальная статистика
stats = get_stats(conn)
print("\n" + "" * 60)
print("📊 РЕЗУЛЬТАТ:")
for k, v in stats.items():
print(f" {k}: {v}")
print("" * 60)
# Шаг 5: Экспорт (если указан --export)
# Новое поведение: если в сессии были прогоны парсера → отдельный CSV
# каждого прогона в exports/YYYY-MM/ с шапкой-метаданными.
# Старое поведение (backward compat): --export без парсинга → плоский CSV
# по min_score в exports/YYYY-MM/leads_<timestamp>.csv.
if args.export:
if collected_run_ids:
print(f"\n📤 Экспорт {len(collected_run_ids)} прогонов сессии:")
for run_id in collected_run_ids:
export_run(config.DB_PATH, run_id)
# Сводный файл «всё, что собрано за день» в папку дня
export_day(config.DB_PATH)
else:
export_to_csv(config.DB_PATH, min_score=args.min_score)
# Шаг 6: master-snapshot всей БД (--export-master)
if args.export_master:
export_master(config.DB_PATH)
# Шаг 7: пересоздать CSV конкретного прогона (--export-run N)
if args.export_run is not None:
export_run(config.DB_PATH, args.export_run)
conn.close()
if __name__ == "__main__": if __name__ == "__main__":