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

PLANET_ABBR = {
    "Sun": "Su", "Moon": "Mo", "Mercury": "Me", "Venus": "Ve", "Mars": "Ma",
    "Jupiter": "Ju", "Saturn": "Sa", "Uranus": "Ur", "Neptune": "Ne", "Pluto": "Pl",
}
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 build_report(
    positions: list[dict],
    data: DataSource,
    aspects: list[dict] | None = None,
    per_object_limit: int = 5000,
) -> 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
            facets.append({
                "type": "aspect", "label": f"{ASP_NAME[asp['aspect']]} z {other}",
                "token": f"{asp_tok} + {other_tok}",
                "count": len(asp_samples), "samples": asp_samples,
            })

        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}