astrololo/services/logic/app/significators.py

"""Most: policzony horoskop → tokeny sygnifikatorów → wyszukiwanie w bazie.

Zalążek LOG-15/16: z pozycji obiektów (wraz z domami) generujemy tokeny w składni
bazy i wyszukujemy pasujące rekordy. Dla każdego obiektu tworzymy kilka **faset**:
- „w znaku"  — planeta + token znaku (`[Su` + `[Tau`)
- „w domu"   — planeta + token domu (`[Su` + `11th H.`)
Aspekty (`[conj`,`[sq`,`[opp`) wymagają policzenia aspektów (LOG-06) — na później.

Format skrótów odczytany z realnej bazy: planety `[Su`,`[Mo`,…; znaki
`[Ari`,`[Tau`,…; domy `12th H.`; np. `[Sa [conj [Su in 6th H.`.
"""
from __future__ import annotations

from typing import Any, Protocol

from app.abbreviations import expand
from app.engine.formats import SIGN_ABBR, SIGNS
from app.scoring import score_facet

PLANET_ABBR = {
    "Sun": "Su", "Moon": "Mo", "Mercury": "Me", "Venus": "Ve", "Mars": "Ma",
    "Jupiter": "Ju", "Saturn": "Sa", "Uranus": "Ur", "Neptune": "Ne", "Pluto": "Pl",
    # punkty wirtualne — tokeny wg SIGNIFICATORS KEY ([NN, [SN, [Lilith)
    "North Node": "NN", "South Node": "SN", "Lilith": "Lilith",
}
SIGN_TO_ABBR = dict(zip(SIGNS, SIGN_ABBR))


class DataSource(Protocol):
    def search(
        self, key: str, value: str, exact: bool, limit: int, fields: list[str] | None = None
    ) -> dict[str, Any]: ...


def _effect(row: dict) -> str:
    for col in ("actioneffect", "topicresult", "bodypart"):
        v = row.get(col)
        if v and str(v).strip().lower() not in ("", "nan"):
            return str(v).strip()
    return ""


def _is_noise(sig: str, effect: str) -> bool:
    s = sig.strip().lower()
    e = effect.strip().lower()
    return (
        e in ("", "nan", "x", "x?", "?", "-")     # efekt pusty/zastępczy
        or s.startswith("significator")            # wiersz-legenda/nagłówek
        or "header" in s
        or s.startswith("*")                       # *MARKER / *header
        or s in ("x", "x?", "nan")                 # znacznik „pomiń rekord"
    )


def _ordinal(n: int) -> str:
    if 10 <= n % 100 <= 20:
        suffix = "th"
    else:
        suffix = {1: "st", 2: "nd", 3: "rd"}.get(n % 10, "th")
    return f"{n}{suffix}"


def _norm(s: str) -> str:
    """Normalizacja do porównań duplikatów: bez skrajnych spacji, jedna spacja, lower."""
    return " ".join(str(s).strip().lower().split())


def _facet_samples(rows: list[dict], tokens: list[str]) -> list[dict]:
    """Rekordy, których sygnifikator zawiera WSZYSTKIE tokeny — bez szumu i bez duplikatów.

    Duplikat = ten sam sygnifikator ORAZ ten sam opis (po normalizacji). Dedup
    działa na zagregowanym wyniku, więc odsiewa też powtórki między wieloma bazami.
    """
    toks = [t.lower() for t in tokens if t]
    out: list[dict] = []
    seen: set[tuple[str, str]] = set()
    for r in rows:
        sig = str(r.get("significator") or "").strip()
        low = sig.lower()
        if not all(t in low for t in toks):
            continue
        eff = _effect(r)
        if _is_noise(sig, eff):
            continue
        key = (_norm(sig), _norm(eff))
        if key in seen:
            continue
        seen.add(key)
        out.append({"significator": sig, "expanded": expand(sig), "effect": eff})
    return out


def _group_by_effect(samples: list[dict]) -> list[dict]:
    """Grupuje próbki po opisie: ten sam efekt = jedna grupa z listą sygnifikatorów."""
    groups: dict[str, dict] = {}
    order: list[str] = []
    for s in samples:
        key = _norm(s["effect"])
        g = groups.get(key)
        if g is None:
            g = {"effect": s["effect"], "count": 0, "significators": []}
            groups[key] = g
            order.append(key)
        g["count"] += 1
        g["significators"].append(s["expanded"])
    result = [groups[k] for k in order]
    result.sort(key=lambda g: g["count"], reverse=True)
    return result


def build_report(
    positions: list[dict],
    data: DataSource,
    aspects: list[dict] | None = None,
    per_object_limit: int = 5000,
    group: bool = False,
) -> dict:
    """positions: pozycje z build_chart (name, sign, direction, house).

    Dla każdego obiektu fasety: „w znaku", „w domu" oraz „w aspekcie" (dla każdego
    aspektu głównego z listy `aspects`, jeśli w bazie są dopasowania). Duplikaty
    (ten sam sygnifikator i opis) są odsiewane wewnątrz każdej fasety.
    """
    from app.engine.aspects import DB_TOKEN as ASP_TOKEN, PL_NAME as ASP_NAME

    items: list[dict] = []
    provider = None
    for p in positions:
        name = p.get("name")
        if name not in PLANET_ABBR:
            continue
        planet_tok = "[" + PLANET_ABBR[name]
        raw = data.search(
            key="significator",
            value=planet_tok,
            exact=False,
            limit=per_object_limit,
            fields=["significator", "actioneffect", "topicresult", "bodypart"],
        )
        provider = raw.get("provider", provider)
        rows = raw.get("rows", [])

        facets: list[dict] = []
        sign = p.get("sign")
        sign_tok = "[" + SIGN_TO_ABBR.get(sign, "")
        sign_samples = _facet_samples(rows, [sign_tok])
        facets.append({
            "type": "sign", "label": f"w znaku {sign}", "token": sign_tok,
            "count": len(sign_samples), "samples": sign_samples,
        })

        house = p.get("house")
        if house:
            ordn = _ordinal(int(house))
            house_samples = _facet_samples(rows, [f"{ordn} h"])   # matcuje '12th H.'
            facets.append({
                "type": "house", "label": f"w {ordn} domu", "token": f"{ordn} H.",
                "count": len(house_samples), "samples": house_samples,
            })

        for asp in (aspects or []):
            if name not in (asp.get("obj1"), asp.get("obj2")):
                continue
            other = asp["obj2"] if asp["obj1"] == name else asp["obj1"]
            asp_tok = ASP_TOKEN.get(asp["aspect"])
            if other not in PLANET_ABBR or not asp_tok:
                continue
            other_tok = "[" + PLANET_ABBR[other]
            asp_samples = _facet_samples(rows, [asp_tok, other_tok])
            if not asp_samples:                       # pokazujemy tylko aspekty z trafieniami
                continue
            as_suffix = f" ({asp['as']})" if asp.get("as") else ""
            facets.append({
                "type": "aspect", "label": f"{ASP_NAME[asp['aspect']]} z {other}{as_suffix}",
                "token": f"{asp_tok} + {other_tok}",
                "aspect": asp["aspect"], "orb": asp.get("orb"), "allowed": asp.get("allowed"),
                "applying": asp.get("applying"),
                "count": len(asp_samples), "samples": asp_samples,
            })

        # punktacja siły (LOG-21), opcjonalne grupowanie po opisie, ranking faset
        for f in facets:
            f["score"] = score_facet(f)
            if group:
                f["groups"] = _group_by_effect(f["samples"])
        facets.sort(key=lambda f: f["score"], reverse=True)

        items.append({
            "object": name,
            "sign": sign,
            "house": house,
            "direction": p.get("direction"),
            "planet_token": planet_tok,
            "planet_total": raw.get("total", 0),
            "facets": facets,
        })
    return {"provider": provider, "objects": items}


def _event_tokens(event: dict) -> list[str]:
    """Tokeny bazy dla zdarzenia osi czasu (spięcie 1B→2B). Pierwszy = planeta."""
    from app.engine.aspects import DB_TOKEN

    if event.get("technique") == "solar_arc":
        p = PLANET_ABBR.get(event.get("directed"))
        a = DB_TOKEN.get(event.get("aspect"))
        q = PLANET_ABBR.get(event.get("target"))     # None dla Asc/MC (nie ma tokenu)
        toks = []
        if p:
            toks.append("[" + p)
        if a:
            toks.append(a)
        if q:
            toks.append("[" + q)
        return toks if p else []
    if event.get("technique") == "profection":
        lord = PLANET_ABBR.get(event.get("lord"))
        sign = SIGN_TO_ABBR.get(event.get("sign"))
        toks = []
        if lord:
            toks.append("[" + lord)
        if sign:
            toks.append("[" + sign)
        return toks if lord else []
    return []


def interpret_events(events: list[dict], data: DataSource, limit: int = 4,
                     per_object_limit: int = 5000) -> list[dict]:
    """Dopina interpretacje z bazy do zdarzeń osi czasu (predykcyjne 1B → 2B).

    Wyszukuje po tokenie planety zdarzenia i zawęża do wszystkich tokenów
    (aspekt/druga planeta lub znak), z odsiewaniem szumu i duplikatów.
    """
    for ev in events:
        tokens = _event_tokens(ev)
        if not tokens:
            continue
        raw = data.search(
            key="significator", value=tokens[0], exact=False, limit=per_object_limit,
            fields=["significator", "actioneffect", "topicresult", "bodypart"],
        )
        samples = _facet_samples(raw.get("rows", []), tokens)
        ev["interpretations"] = samples[:limit]
        ev["interpretations_count"] = len(samples)
    return events