"""Backfill objective_complex_mapping (#203).

Auto-match DOM.РФ комплексов к Objective dataset через fuzzy name matching.
Goal: coverage 3% → 40%+ для mv_layout_velocity.

Schema facts (confirmed via pg MCP):
- domrf_kn_objects: фильтр is_ekb = true (~1285 ЕКБ объектов).
  district_name ILIKE '%екатеринбург%' возвращает 0 строк — не использовать.
- objective_corpus_room_month: group_name = 'Екатеринбург' (единственное значение),
  263 distinct project_name.
- objective_complex_mapping: UNIQUE (objective_complex_name, objective_group),
  колонки match_method + match_score + is_reviewed поддерживают audit trail.

match_method history:
  'fuzzy'          — legacy Anton SQLite import (avg score 0.98, 127 rows)
  'fuzzy_trgm'     — pg_trgm backfill, auto-accept threshold=0.85 (первый запуск)
  'fuzzy_v2'       — pg_trgm backfill, pruned threshold=0.80 (второй запуск, #44)
  'auto_core_dev_v5' — core-name normalize (#2198 normalize_complex_name) обеих сторон +
                       dev-confirm. trgm по СЫРЫМ именам давал ~300 маппингов, 1253 obj_id
                       остались (низкая similarity: «Жилой комплекс "Малевич" (ЖК "Малевич")»
                       vs «Малевич»). Ядро-нормализация схлопывает обёртки → exact-core match;
                       dev-подтверждение (пересечение токенов / подстрока) отсеивает
                       одноимённые чужие проекты (Объектив = только ЕКБ). #2177 шаг 2.
  'manual'         — ручная корректура
"""

from __future__ import annotations

import argparse
import logging
import re
import sys
from dataclasses import dataclass, field

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.services.site_finder.best_layouts import normalize_complex_name

logger = logging.getLogger(__name__)

# Порог для auto-insert (высокая уверенность)
AUTO_ACCEPT_THRESHOLD = 0.85

# Pruned threshold для v2 run (ниже уверенность, is_reviewed=false → review queue)
AUTO_ACCEPT_THRESHOLD_V2 = 0.80

# Порог для review queue (средняя уверенность — Phase 2)
REVIEW_THRESHOLD = 0.6


@dataclass
class MatchCandidate:
    """Один candidate match DOM.РФ ↔ Objective."""

    domrf_obj_id: int
    domrf_comm_name: str
    domrf_dev_name: str | None
    objective_project_name: str
    similarity_score: float  # 0.0..1.0


def find_match_candidates(
    db: Session,
    *,
    only_unmapped: bool = True,
    min_threshold: float | None = None,
    limit: int | None = None,
) -> list[MatchCandidate]:
    """Поиск candidates через pg_trgm similarity.

    Использует CROSS JOIN LATERAL + similarity() для fuzzy match
    comm_name (DOM.РФ) ↔ project_name (Objective).

    Args:
        db: SQLAlchemy sync Session.
        only_unmapped: Если True — пропускает уже-mapped obj_id.
        min_threshold: Нижняя граница similarity для фильтрации кандидатов.
            По умолчанию REVIEW_THRESHOLD (0.6).
        limit: Максимальное число строк результата (для тестирования).

    Returns:
        Список MatchCandidate, отсортированных по убыванию similarity_score.
    """
    effective_min = min_threshold if min_threshold is not None else REVIEW_THRESHOLD

    # Формируем SQL. LIMIT добавляем через int() — SQL injection safe (только число).
    limit_clause = f"LIMIT {int(limit)}" if limit is not None else ""

    sql = text(
        f"""
        WITH domrf_unmapped AS (
            SELECT o.obj_id, o.comm_name, o.dev_name
            FROM domrf_kn_objects o
            WHERE o.is_ekb = true
              AND o.comm_name IS NOT NULL
              AND (
                  CAST(:only_unmapped AS boolean) = FALSE
                  OR NOT EXISTS (
                      SELECT 1 FROM objective_complex_mapping cm
                      WHERE cm.domrf_obj_id = o.obj_id
                  )
              )
        ),
        objective_distinct AS (
            SELECT DISTINCT project_name
            FROM objective_corpus_room_month
            WHERE group_name = 'Екатеринбург'
        )
        SELECT
            d.obj_id,
            d.comm_name,
            d.dev_name,
            obj.project_name,
            similarity(d.comm_name, obj.project_name) AS sim_score
        FROM domrf_unmapped d
        CROSS JOIN LATERAL (
            SELECT project_name
            FROM objective_distinct
            WHERE similarity(d.comm_name, project_name) > 0.4
            ORDER BY similarity(d.comm_name, project_name) DESC
            LIMIT 1
        ) obj
        WHERE similarity(d.comm_name, obj.project_name) >= CAST(:min_threshold AS float)
        ORDER BY sim_score DESC
        {limit_clause}
        """
    )

    rows = db.execute(
        sql,
        {"only_unmapped": only_unmapped, "min_threshold": effective_min},
    ).all()

    return [
        MatchCandidate(
            domrf_obj_id=int(r[0]),
            domrf_comm_name=str(r[1]),
            domrf_dev_name=str(r[2]) if r[2] is not None else None,
            objective_project_name=str(r[3]),
            similarity_score=float(r[4]),
        )
        for r in rows
    ]


def auto_apply_matches(
    db: Session,
    candidates: list[MatchCandidate],
    *,
    threshold: float = AUTO_ACCEPT_THRESHOLD,
    match_method: str = "fuzzy_trgm",
    dry_run: bool = False,
) -> dict[str, int]:
    """Apply candidates с score >= threshold в objective_complex_mapping.

    Кандидаты ниже threshold, но >= REVIEW_THRESHOLD попадают в review_queue
    (Phase 2 — UI для ручного review, в этом PR только считаются).

    ON CONFLICT DO NOTHING — если пара (objective_complex_name, objective_group)
    уже существует, строка пропускается без ошибки.

    Args:
        db: SQLAlchemy sync Session.
        candidates: Список из find_match_candidates().
        threshold: Минимальный score для auto-insert (default 0.85).
        match_method: Значение колонки match_method в БД. По умолчанию 'fuzzy_trgm'.
            Для re-run с пониженным порогом передавать 'fuzzy_v2'.
        dry_run: Если True — только логирует, не пишет в БД.

    Returns:
        dict с ключами auto_accepted, review_queue, skipped.
    """
    auto = [c for c in candidates if c.similarity_score >= threshold]
    review = [c for c in candidates if REVIEW_THRESHOLD <= c.similarity_score < threshold]

    if dry_run:
        logger.info(
            "DRY RUN: would auto-accept %d, review queue %d",
            len(auto),
            len(review),
        )
        return {"auto_accepted": len(auto), "review_queue": len(review), "skipped": 0}

    inserted = 0
    skipped = 0
    for c in auto:
        try:
            with db.begin_nested():
                result = db.execute(
                    text(
                        """
                        INSERT INTO objective_complex_mapping
                            (objective_complex_name, domrf_obj_id, objective_group,
                             match_method, match_score, is_reviewed)
                        VALUES (
                            CAST(:name AS text),
                            CAST(:obj_id AS bigint),
                            CAST(:group AS text),
                            CAST(:method AS text),
                            CAST(:score AS numeric),
                            CAST(:reviewed AS boolean)
                        )
                        ON CONFLICT (objective_complex_name, objective_group) DO NOTHING
                        """
                    ),
                    {
                        "name": c.objective_project_name,
                        "obj_id": c.domrf_obj_id,
                        "group": "Екатеринбург",
                        "method": match_method,
                        "score": c.similarity_score,
                        "reviewed": False,
                    },
                )
            if result.rowcount > 0:
                inserted += 1
            else:
                skipped += 1
        except Exception as e:
            logger.warning(
                "Insert failed для %s ↔ %s: %s",
                c.domrf_comm_name,
                c.objective_project_name,
                e,
            )
            skipped += 1

    db.commit()
    logger.info(
        "Backfill complete: auto_accepted=%d skipped=%d review_queue=%d",
        inserted,
        skipped,
        len(review),
    )
    return {"auto_accepted": inserted, "review_queue": len(review), "skipped": skipped}


def trigger_mv_refresh(db: Session) -> int:
    """REFRESH mv_layout_velocity после backfill.

    Вызывает существующий helper из layout_velocity_refresh.
    Передаём concurrently=True (MV уже заполнен).
    """
    from app.services.site_finder.layout_velocity_refresh import refresh_layout_velocity

    return refresh_layout_velocity(db, concurrently=True)


# ════════════════════════════════════════════════════════════════════════════
# Core-name pass (#2177 шаг 2) — auto_core_dev_v5
# ════════════════════════════════════════════════════════════════════════════

# objective_group у всех существующих строк mapping = 'Екатеринбург'
# (Объектив покрывает только ЕКБ). Новые строки пишем с тем же значением.
OBJECTIVE_GROUP = "Екатеринбург"

CORE_MATCH_METHOD = "auto_core_dev_v5"
CORE_MATCH_SCORE = 0.90
CORE_MATCH_NOTE = "#2177 step2: exact-core + dev-confirm"

# Префиксы/обёртки орг-формы застройщика, срезаемые перед сравнением.
# Порядок применения не важен (все режутся из целого норм-имени).
_DEV_JUNK_RE = re.compile(
    r"""
    \b(?:
        ооо | оао | пао | зао | ао | ип
      | сз                              # СЗ = специализированный застройщик (аббрев.)
      | специализированн\w*             # «специализированный застройщик»
      | застройщик\w*
      | девелопмент\w*
      | групп\w*                        # групп / группа
      | компани\w*
      | инвест\w*                       # инвест / инвестиции (частый шум)
      | строительн\w*
    )\b
    """,
    re.VERBOSE,
)


def _norm_dev(raw: str | None) -> str:
    """Нормализовать имя застройщика до сравниваемого ядра.

    lower → срез кавычек/скобок → срез орг-формы (ООО/АО/ПАО/ЗАО/«СЗ»/
    «специализированный застройщик»/«девелопмент»/«групп(а)») → схлопывание
    пробелов. Пустой / None → ''.

    Namespaced на _NAME_-уровне: возвращаем очищенную строку, дальнейшая
    классификация (пересечение токенов / подстрока) — в find_core_matches.
    """
    if not raw:
        return ""
    name = raw.strip().lower()
    # убрать кавычки-ёлочки/лапки/скобки — оставить только содержательные символы
    name = re.sub(r"[«»\"'`()\[\]]", " ", name)
    name = _DEV_JUNK_RE.sub(" ", name)
    # мусорные разделители → пробел (точки в «Астон.» и т.п. не трогаем на dev-стороне:
    # dev-имена обычно без них, но на всякий нормализуем распространённые)
    name = re.sub(r"[.,;/\\]", " ", name)
    name = re.sub(r"\s+", " ", name).strip()
    return name


def _dev_confirms(domrf_dev: str, objective_devs: set[str]) -> bool:
    """True если domrf-застройщик подтверждает связь с objective-проектом.

    У одного objective project_name бывает НЕСКОЛЬКО developers (SPV / очереди) —
    подтверждение по ЛЮБОМУ из них. Критерий (любой):
      • пересечение токенов норм-имён (общий содержательный токен), ИЛИ
      • одно норм-имя — подстрока другого (в любую сторону), НО только когда
        короткая сторона ≥ 4 символов: «дом»/«юг»/«ак» как подстрока чужого
        длинного имени («Домострой», «Югстройинвест») — ложное подтверждение,
        а именно dev-confirm — последняя защита от generic-core коллизий.

    Пустые имена не подтверждают (нет сигнала).
    """
    d = _norm_dev(domrf_dev)
    if not d:
        return False
    d_tokens = {t for t in d.split() if len(t) >= 3}
    for raw_odev in objective_devs:
        o = _norm_dev(raw_odev)
        if not o:
            continue
        if min(len(d), len(o)) >= 4 and (d in o or o in d):
            return True
        o_tokens = {t for t in o.split() if len(t) >= 3}
        if d_tokens & o_tokens:
            return True
    return False


@dataclass
class CoreMatch:
    """Один core-match domrf-объекта к objective-проекту."""

    domrf_obj_id: int
    domrf_comm_name: str
    domrf_dev_name: str | None
    core: str
    objective_project_name: str
    objective_project_id: int | None
    objective_developers: list[str]
    dev_confirmed: bool


@dataclass
class CoreMatchReport:
    """Итог классификации core-pass.

    tier_a  — ровно 1 objective-кандидат И dev-подтверждён → пишем в БД.
    tier_b  — ровно 1 кандидат, dev НЕ подтверждён → кандидат на DaData-гео
              проход, в БД НЕ пишем (только отчёт).
    ambiguous — >1 objective-кандидатов по core → в отчёт, разрешение вручную/гео.
    skipped_taken — objective_complex_name уже занят в mapping (UNIQUE-констрейнт;
                    его domrf-группа уже покрыта — дубли не нужны).
    """

    tier_a: list[CoreMatch] = field(default_factory=list)
    tier_b: list[CoreMatch] = field(default_factory=list)
    ambiguous: list[CoreMatch] = field(default_factory=list)
    skipped_taken: list[CoreMatch] = field(default_factory=list)

    def counts(self) -> dict[str, int]:
        return {
            "tier_a": len(self.tier_a),
            "tier_b": len(self.tier_b),
            "ambiguous": len(self.ambiguous),
            "skipped_taken": len(self.skipped_taken),
        }


# ── SQL: несопоставленные domrf ЕКБ-объекты, latest snapshot per obj_id ───────
# ЛОВУШКА is_ekb (прод-проверка 2026-07-03): флаг проставляется per-snapshot и на
# свежих снапшотах ПУСТ (все latest-строки is_ekb=false — см. также комментарий
# в 155_objective_mapping_trgm_backfill.sql). `WHERE is_ekb = true` внутри
# DISTINCT ON отдал бы устаревшие имена со старых снапшотов (или 0 строк после
# DISTINCT). Поэтому: ЕКБ-принадлежность = «obj_id имел is_ekb=true хоть на одном
# снапшоте» (1285 живых объектов), а строку берём с LATEST снапшота без флага.
_DOMRF_UNMAPPED_SQL = text("""
    SELECT DISTINCT ON (o.obj_id)
        o.obj_id, o.comm_name, o.dev_name
    FROM domrf_kn_objects o
    WHERE o.comm_name IS NOT NULL
      AND o.obj_id IN (
          SELECT DISTINCT e.obj_id FROM domrf_kn_objects e WHERE e.is_ekb = true
      )
      AND NOT EXISTS (
          SELECT 1 FROM objective_complex_mapping cm
          WHERE cm.domrf_obj_id = o.obj_id
      )
    ORDER BY o.obj_id, o.snapshot_date DESC NULLS LAST
""")

# ── SQL: objective-проекты — distinct project_name + агрегаты developer/id ────
# objective_project_id: в objective_lots есть колонка objective_project_id
# (68_schema_objective.sql). Берём MAX (стабилен per project_name; NULL если нет).
_OBJECTIVE_PROJECTS_SQL = text("""
    SELECT
        ol.project_name,
        ARRAY_REMOVE(ARRAY_AGG(DISTINCT ol.developer), NULL) AS developers,
        MAX(ol.objective_project_id) AS objective_project_id
    FROM objective_lots ol
    WHERE ol.project_name IS NOT NULL
      AND ol.project_name <> ''
    GROUP BY ol.project_name
""")

# ── SQL: уже занятые objective_complex_name в текущей группе ──────────────────
_TAKEN_NAMES_SQL = text("""
    SELECT objective_complex_name
    FROM objective_complex_mapping
    WHERE objective_group = CAST(:group AS text)
""")


def find_core_matches(db: Session) -> CoreMatchReport:
    """Сопоставить несопоставленные domrf ЕКБ-объекты с objective-проектами по ядру.

    Обе стороны нормализуются `normalize_complex_name` (#2198): обёртки вроде
    «Жилой комплекс "Малевич" (ЖК "Малевич")» схлопываются до ядра «малевич»,
    что даёт exact-core match там, где trgm по сырым именам не дотягивал.

    Классификация (см. CoreMatchReport):
      tier_a    — 1 objective-кандидат + dev-подтверждён (пишем);
      tier_b    — 1 кандидат, dev НЕ подтверждён (DaData-гео проход, не пишем);
      ambiguous — >1 кандидата (ручное/гео разрешение);
      skipped   — objective_complex_name уже в mapping (UNIQUE-констрейнт).

    Реального обращения к БД в тестах нет — db.execute мокается по паттерну файла.
    """
    report = CoreMatchReport()

    # objective-сторона: core → [(project_name, {devs}, project_id), ...]
    objective_by_core: dict[str, list[tuple[str, set[str], int | None]]] = {}
    for row in db.execute(_OBJECTIVE_PROJECTS_SQL).all():
        project_name = str(row[0])
        developers = {str(d) for d in (row[1] or []) if d}
        project_id = int(row[2]) if row[2] is not None else None
        core = normalize_complex_name(project_name)
        if not core:
            continue
        objective_by_core.setdefault(core, []).append((project_name, developers, project_id))

    # уже занятые objective_complex_name (skip — их группа уже покрыта)
    taken_names: set[str] = {
        str(r[0]) for r in db.execute(_TAKEN_NAMES_SQL, {"group": OBJECTIVE_GROUP}).all()
    }

    # domrf-сторона: несопоставленные ЕКБ, latest snapshot per obj_id
    for row in db.execute(_DOMRF_UNMAPPED_SQL).all():
        obj_id = int(row[0])
        comm_name = str(row[1])
        dev_name = str(row[2]) if row[2] is not None else None
        core = normalize_complex_name(comm_name)
        if not core:
            continue

        candidates = objective_by_core.get(core)
        if not candidates:
            continue  # нет objective-проекта с таким ядром — не наш случай

        if len(candidates) > 1:
            # >1 objective project_name делят одно ядро → ambiguous (не выбираем вслепую)
            proj_name, devs, proj_id = candidates[0]
            report.ambiguous.append(
                _build_core_match(obj_id, comm_name, dev_name, core, proj_name, proj_id, devs)
            )
            continue

        proj_name, devs, proj_id = candidates[0]

        if proj_name in taken_names:
            report.skipped_taken.append(
                _build_core_match(obj_id, comm_name, dev_name, core, proj_name, proj_id, devs)
            )
            continue

        match = _build_core_match(obj_id, comm_name, dev_name, core, proj_name, proj_id, devs)
        if match.dev_confirmed:
            report.tier_a.append(match)
        else:
            report.tier_b.append(match)

    logger.info("find_core_matches: %s", report.counts())
    return report


def _build_core_match(
    obj_id: int,
    comm_name: str,
    dev_name: str | None,
    core: str,
    proj_name: str,
    proj_id: int | None,
    devs: set[str],
) -> CoreMatch:
    """Сконструировать CoreMatch + посчитать dev_confirmed."""
    return CoreMatch(
        domrf_obj_id=obj_id,
        domrf_comm_name=comm_name,
        domrf_dev_name=dev_name,
        core=core,
        objective_project_name=proj_name,
        objective_project_id=proj_id,
        objective_developers=sorted(devs),
        dev_confirmed=_dev_confirms(dev_name or "", devs),
    )


def apply_core_matches(
    db: Session,
    report: CoreMatchReport,
    *,
    dry_run: bool = True,
) -> dict[str, int]:
    """Вставить ТОЛЬКО tier_a в objective_complex_mapping (auto_core_dev_v5).

    tier_b / ambiguous / skipped НЕ пишутся (см. CoreMatchReport). Каждая вставка
    в per-row SAVEPOINT (backend.md) с ON CONFLICT (objective_complex_name,
    objective_group) DO NOTHING — гонка/дубль не роняет весь проход.

    Args:
        db: SQLAlchemy sync Session.
        report: результат find_core_matches().
        dry_run: True (default) — только логирует counts + первые 30 пар, БЕЗ insert.

    Returns:
        dict: inserted, conflict_skipped, error_skipped, tier_a_total.
    """
    counts = report.counts()
    logger.info(
        "apply_core_matches (dry_run=%s): tier_a=%d tier_b=%d ambiguous=%d skipped_taken=%d",
        dry_run,
        counts["tier_a"],
        counts["tier_b"],
        counts["ambiguous"],
        counts["skipped_taken"],
    )

    for m in report.tier_a[:30]:
        logger.info(
            "  tier_a: obj_id=%s core=%r domrf=%r → objective=%r (dev %r ~ %r)",
            m.domrf_obj_id,
            m.core,
            m.domrf_comm_name,
            m.objective_project_name,
            m.domrf_dev_name,
            m.objective_developers,
        )

    if dry_run:
        return {
            "inserted": 0,
            "conflict_skipped": 0,
            "error_skipped": 0,
            "tier_a_total": counts["tier_a"],
        }

    inserted = 0
    conflict_skipped = 0
    error_skipped = 0
    for m in report.tier_a:
        try:
            with db.begin_nested():
                result = db.execute(
                    text(
                        """
                        INSERT INTO objective_complex_mapping
                            (objective_complex_name, objective_project_id, domrf_obj_id,
                             objective_group, match_method, match_score, is_reviewed, note)
                        VALUES (
                            CAST(:name AS text),
                            CAST(:project_id AS bigint),
                            CAST(:obj_id AS bigint),
                            CAST(:group AS text),
                            CAST(:method AS text),
                            CAST(:score AS numeric),
                            CAST(:reviewed AS boolean),
                            CAST(:note AS text)
                        )
                        ON CONFLICT (objective_complex_name, objective_group) DO NOTHING
                        """
                    ),
                    {
                        "name": m.objective_project_name,
                        "project_id": m.objective_project_id,
                        "obj_id": m.domrf_obj_id,
                        "group": OBJECTIVE_GROUP,
                        "method": CORE_MATCH_METHOD,
                        "score": CORE_MATCH_SCORE,
                        "reviewed": False,
                        "note": CORE_MATCH_NOTE,
                    },
                )
            if result.rowcount > 0:
                inserted += 1
            else:
                conflict_skipped += 1
        except Exception as e:
            logger.warning(
                "core insert failed для %s ↔ %s: %s",
                m.domrf_comm_name,
                m.objective_project_name,
                e,
            )
            error_skipped += 1

    db.commit()
    logger.info(
        "apply_core_matches done: inserted=%d conflict_skipped=%d error_skipped=%d",
        inserted,
        conflict_skipped,
        error_skipped,
    )
    return {
        "inserted": inserted,
        "conflict_skipped": conflict_skipped,
        "error_skipped": error_skipped,
        "tier_a_total": counts["tier_a"],
    }


def run_core_pass(*, apply: bool = False) -> dict[str, int]:
    """CLI-обёртка core-pass: открыть Session, найти матчи, (опц.) применить.

    Дефолт — dry-run. Реальный insert только при apply=True.
    """
    from app.core.db import SessionLocal

    db = SessionLocal()
    try:
        report = find_core_matches(db)
        return apply_core_matches(db, report, dry_run=not apply)
    finally:
        db.close()


def _main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="objective_backfill core-pass (#2177 step2, auto_core_dev_v5)."
    )
    parser.add_argument(
        "--core-pass",
        action="store_true",
        help="Запустить core-name + dev-confirm проход (иначе no-op).",
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Реально писать tier_a в БД (по умолчанию — dry-run).",
    )
    args = parser.parse_args(argv)

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )

    if not args.core_pass:
        parser.print_help()
        return 0

    result = run_core_pass(apply=args.apply)
    logger.info("core-pass result: %s", result)
    return 0


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