parser-v1/audit_scores.py

"""Аудит скоринга — независимая проверка всей leads.db.

Запуск:  python audit_scores.py   (из папки parser_v1)

Что делает:
  1. Пересчитывает score каждого лида из сырых полей (scoring.calculate_score)
     и сверяет со значением в БД — ловит «устаревший» score / band / pain_products.
  2. Прогоняет САНИТАРНЫЕ правила: боль не должна противоречить данным
     (напр. причина «нет сайта», а сайт есть). Это потенциальные баги скоринга.
  3. Прогоняет правила ЦЕЛОСТНОСТИ ДАННЫХ (has_website=1 но website пуст и т.п.).
  4. Собирает очередь НА РУЧНОЙ ВЗГЛЯД (не баги — пограничные: hot при низкой
     полноте диагностики, «процветающие» в hot).

Только чтение БД. Ничего не пишет.
Выход: код 1 если найдены HARD-аномалии (для CI/хука), иначе 0.
"""
import json
import sqlite3
import sys

sys.path.insert(0, ".")
import config
import scoring

DB = "leads.db"
CONSTRUCTORS = scoring.CONSTRUCTOR_CMS


def _pp(s):
    try:
        return json.loads(s or "{}")
    except (ValueError, TypeError):
        return {}


def main():
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass

    con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    con.row_factory = sqlite3.Row
    rows = con.execute("SELECT * FROM leads").fetchall()
    con.close()

    hard = []   # (rule, lead_id, name, detail) — противоречия = баги
    soft = []   # (rule, lead_id, name, detail) — на ручной взгляд

    for r in rows:
        L = dict(r)
        lid, name = L["id"], (L.get("name") or "")[:28]
        cat = (L.get("category") or "").lower()
        avg = float(L.get("reviews_avg") or 0)
        rc = int(L.get("reviews_count") or 0)

        # ── 1. Пересчёт и сверка со стором ───────────────────────────
        rescore, breakdown = scoring.calculate_score(L)
        reasons = breakdown["reasons"]
        if (L.get("score") or 0) != rescore:
            hard.append(("STALE_SCORE", lid, name, f"в БД {L.get('score')} != пересчёт {rescore} (нужен --rescore)"))
        if L.get("band") and L["band"] != breakdown["band"]:
            hard.append(("BAND_MISMATCH", lid, name, f"в БД {L['band']} != {breakdown['band']}"))
        if _pp(L.get("pain_products")) != breakdown["pain_products"]:
            hard.append(("PAIN_STALE", lid, name, "pain_products в БД != пересчёт"))

        # ── 2. Санитарные правила: боль vs данные ────────────────────
        if "нет сайта" in reasons and (L.get("website") or L.get("has_website") == 1 or L.get("site_alive") == 1):
            hard.append(("FALSE_NO_SITE", lid, name, f"боль 'нет сайта', но website={L.get('website')} has_website={L.get('has_website')} alive={L.get('site_alive')}"))
        if "нет соцсетей (VK/Telegram)" in reasons and (L.get("vk_url") or L.get("telegram_url")):
            hard.append(("FALSE_NO_SOCIAL", lid, name, f"боль 'нет соцсетей', но vk={L.get('vk_url')} tg={L.get('telegram_url')}"))
        if "сайт на конструкторе" in reasons and (L.get("cms_type") or "").lower() not in CONSTRUCTORS:
            hard.append(("FALSE_CONSTRUCTOR", lid, name, f"боль 'конструктор', но cms={L.get('cms_type')}"))
        if any("рейтинг" in x for x in reasons) and avg >= 4.5:
            hard.append(("FALSE_RATING", lid, name, f"боль по рейтингу, но avg={avg}"))
        if "нет онлайн-записи" in reasons and not scoring._cat_in(cat, config.APPT_CATEGORIES):
            hard.append(("BOOKING_WRONG_CAT", lid, name, f"'нет онлайн-записи' у не-сервисной категории '{cat}'"))
        if "нет соцсетей (VK/Telegram)" in reasons and not scoring._cat_in(cat, config.SOCIAL_SALES_CATEGORIES):
            hard.append(("SOCIAL_WRONG_CAT", lid, name, f"'нет соцсетей' у не-B2C категории '{cat}'"))

        # ── 3. Целостность данных ────────────────────────────────────
        if L.get("has_website") == 1 and not L.get("website"):
            hard.append(("WEBSITE_LOST", lid, name, "has_website=1, но website пуст (потеря URL)"))
        if L.get("site_alive") == 1 and L.get("has_online_booking") is None:
            soft.append(("PARTIAL_ENRICH", lid, name, "site_alive=1, но has_online_booking не проверен"))
        if rc > 0 and avg == 0:
            soft.append(("REVIEWS_NO_AVG", lid, name, f"reviews_count={rc}, но avg=0"))
        if not name:
            hard.append(("NO_NAME", lid, name, "пустое имя"))

        # ── 4. На ручной взгляд (пограничное, не баг) ────────────────
        cov = L.get("diagnostic_coverage")
        band = breakdown["band"]
        if band == "hot" and cov is not None and cov < config.MIN_COVERAGE:
            soft.append(("HOT_LOW_COVERAGE", lid, name, f"hot при coverage={cov} (тонкие данные)"))
        if rescore > 0 and not breakdown["pain_products"]:
            hard.append(("SCORE_NO_PAIN", lid, name, f"score={rescore}, но pain_products пуст"))

    # ── Отчёт ────────────────────────────────────────────────────────
    print(f"АУДИТ СКОРИНГА — {len(rows)} лидов\n" + "=" * 60)

    def group(items, title):
        from collections import defaultdict
        by = defaultdict(list)
        for rule, lid, nm, det in items:
            by[rule].append((lid, nm, det))
        if not items:
            print(f"\n{title}: чисто ✅")
            return
        print(f"\n{title}: {len(items)} (правил: {len(by)})")
        for rule, lst in by.items():
            print(f"  ▸ {rule}: {len(lst)}")
            for lid, nm, det in lst[:5]:
                print(f"      #{lid} {nm} — {det}")
            if len(lst) > 5:
                print(f"      … ещё {len(lst) - 5}")

    group(hard, "🔴 HARD — противоречия (баги)")
    group(soft, "🟡 SOFT — на ручной взгляд")

    print("\n" + "=" * 60)
    if hard:
        print(f"ИТОГ: ⚠️ {len(hard)} HARD-аномалий — есть что чинить.")
        return 1
    print("ИТОГ: ✅ HARD-аномалий нет — скоринг консистентен с данными.")
    return 0


if __name__ == "__main__":
    sys.exit(main())