# =============================================================================
# Job·Ka — Groupe KA
# Auteur  : Simon-Pierre Boucher
# Contact : contact@spboucher.ai
# Fichier : jobka/dedup.py
# Rôle    : Déduplication inter-sources — blocage par (titre, employeur, ville)
#           + union-find, offres canoniques par autorité de la source
# Créé    : 2026-08-17   Modifié : 2026-08-17
# =============================================================================
"""Déduplication des offres publiées par plusieurs sources.

Même squelette que Lou·Ka (clé exacte + union-find, AUCUN fuzzy matching) :
1. blocage par clé « titre normalisé | employeur normalisé | ville » ;
2. confirmation de paire : clé identique obligatoire, refus sur signal
   contradictoire (type d'emploi différent, salaires connus s'écartant de
   plus de SALARY_TOL), et exigence d'un second signal concordant ;
3. union-find avec contrainte : jamais deux offres de la même source dans
   une composante (et jamais de dédup intra-source) ;
4. canonique = source la plus autoritaire (page carrière de l'employeur
   avant agrégat), puis richesse du contenu. Les doublons sont MASQUÉS
   (dup_of), jamais supprimés. Recalcul intégral idempotent à chaque passe.
"""
from __future__ import annotations

import json
import re
import sqlite3

from . import db
from .normalize import strip_accents

SALARY_TOL = 0.04      # ±4 % d'écart toléré entre salaires « identiques »
MAX_BLOCK = 200        # bloc anormalement gros = clé trop générique, ignorer

# Sources agrégatrices (guichets, portails) — moins autoritaires que la page
# carrière de l'employeur : en cas de doublon, l'offre DIRECTE gagne la
# canonique et la copie portail est masquée.
AGGREGATORS: set[str] = {"guichet_emplois", "jobillico", "espresso_jobs"}

# Mots de bruit RH retirés du titre pour le blocage (bilinguisme, urgence…)
_TITLE_NOISE = re.compile(
    r"\b(h/f|f/h|m/f|f/m|h-f|\(h/f\)|bilingue|bilingual|urgent|nouveau|new|"
    r"junior|senior|sr|jr|intermediaire|intermediate|principal[e]?|lead|"
    r"temps plein|temps partiel|full[- ]?time|part[- ]?time|permanent[e]?|"
    r"contractuel(?:le)?|stage|stagiaire|remote|teletravail|hybride|hybrid)\b",
    re.I)


def _norm(s: str) -> str:
    s = strip_accents((s or "").lower())
    s = re.sub(r"[^a-z0-9]+", " ", s)
    return re.sub(r"\s+", " ", s).strip()


def _block_key(title: str, employer: str, city: str) -> str:
    t = _TITLE_NOISE.sub(" ", strip_accents((title or "").lower()))
    t = re.sub(r"[^a-z0-9]+", " ", t)
    t = re.sub(r"\s+", " ", t).strip()
    e = _norm(employer)
    c = _norm(city)
    if not t or not e:
        return ""     # sans titre ou employeur, pas de blocage fiable
    return f"{t}|{e}|{c}"


def _salary_close(a_min, a_max, b_min, b_max) -> bool | None:
    """True/False si comparables, None si l'un des deux est inconnu."""
    if a_min is None or b_min is None:
        return None
    a, b = float(a_min), float(b_min)
    if max(a, b) == 0:
        return None
    if abs(a - b) / max(a, b) > SALARY_TOL:
        return False
    if a_max is not None and b_max is not None:
        a2, b2 = float(a_max), float(b_max)
        if max(a2, b2) and abs(a2 - b2) / max(a2, b2) > SALARY_TOL:
            return False
    return True


def _same_posting(a: sqlite3.Row, b: sqlite3.Row) -> bool:
    """Deux offres de la même clé de blocage sont-elles le même poste ?

    Clé identique obligatoire (déjà garanti par le blocage). Ensuite :
    refus sur signal contradictoire, puis exigence d'un second signal
    concordant — la clé seule ne suffit pas (deux affichages distincts du
    même titre chez le même employeur restent deux postes).
    """
    # signaux contradictoires -> refus
    ta, tb = a["employment_type"], b["employment_type"]
    if ta and tb and ta != tb:
        return False
    sal = _salary_close(a["salary_year_min"], a["salary_year_max"],
                        b["salary_year_min"], b["salary_year_max"])
    if sal is False:
        return False

    # second signal concordant requis
    if sal is True:
        return True
    if ta and tb and ta == tb:
        return True
    da, db_ = a["date_posted"], b["date_posted"]
    if da and db_:
        try:
            import datetime as _dt
            delta = abs((_dt.date.fromisoformat(da) - _dt.date.fromisoformat(db_)).days)
            if delta <= 7:
                return True
        except ValueError:
            pass
    pa, pb = a["postal_code"], b["postal_code"]
    if pa and pb and pa == pb:
        return True
    return False


def _authority(source: str) -> int:
    """0 = page carrière directe (le plus autoritaire), 10 = agrégateur."""
    return 10 if source in AGGREGATORS else 0


def run(con: sqlite3.Connection | None = None) -> dict:
    """Recalcule dup_of / dup_sources pour toutes les offres actives."""
    own = con is None
    if own:
        con = db.connect()

    rows = con.execute(
        """SELECT uid, source, title, employer, city, employment_type,
                  salary_year_min, salary_year_max, date_posted, postal_code,
                  LENGTH(description) desc_len
           FROM jobs WHERE active=1""").fetchall()

    by_key: dict[str, list[int]] = {}
    for i, r in enumerate(rows):
        k = _block_key(r["title"], r["employer"], r["city"])
        if k:
            by_key.setdefault(k, []).append(i)

    # union-find avec contrainte anti-transitive : une composante ne peut
    # contenir deux offres de la même source
    parent = list(range(len(rows)))
    srcset: dict[int, set[str]] = {i: {rows[i]["source"]} for i in range(len(rows))}

    def find(i: int) -> int:
        while parent[i] != i:
            parent[i] = parent[parent[i]]
            i = parent[i]
        return i

    def union(i: int, j: int) -> bool:
        ri, rj = find(i), find(j)
        if ri == rj:
            return True
        if srcset[ri] & srcset[rj]:
            return False
        parent[rj] = ri
        srcset[ri] |= srcset.pop(rj)
        return True

    for key, members in by_key.items():
        if len(members) < 2 or len(members) > MAX_BLOCK:
            continue
        for x in range(len(members)):
            for y in range(x + 1, len(members)):
                i, j = members[x], members[y]
                if rows[i]["source"] == rows[j]["source"]:
                    continue   # jamais de dédup intra-source
                if _same_posting(rows[i], rows[j]):
                    union(i, j)

    # résolution : canonique par (autorité, richesse, uid)
    groups: dict[int, list[int]] = {}
    for i in range(len(rows)):
        groups.setdefault(find(i), []).append(i)

    n_groups = hidden = 0
    con.execute("UPDATE jobs SET dup_of=NULL, dup_sources=NULL")
    for members in groups.values():
        if len(members) < 2:
            continue
        n_groups += 1
        members.sort(key=lambda i: (_authority(rows[i]["source"]),
                                    -(rows[i]["desc_len"] or 0), rows[i]["uid"]))
        canon = rows[members[0]]["uid"]
        others = [rows[i]["source"] for i in members[1:]]
        con.execute("UPDATE jobs SET dup_sources=? WHERE uid=?",
                    (json.dumps(sorted(set(others)), ensure_ascii=False), canon))
        for i in members[1:]:
            con.execute("UPDATE jobs SET dup_of=? WHERE uid=?",
                        (canon, rows[i]["uid"]))
            hidden += 1
    con.commit()
    if own:
        con.close()
    return {"groups": n_groups, "duplicates_hidden": hidden,
            "jobs_scanned": len(rows)}


if __name__ == "__main__":
    print(run())