Jyotisha/scripts/rectification/event_probes.py

"""Public-safe biographical probes from candidate dasha / varga differences.

Inverts event scoring: pick two representative minutes, find calendar years
where Vimshottari + Narayana activation (or true period-start years) differ,
and emit a yes/no life-event question. Never grants a unique minute.
"""

from __future__ import annotations

from datetime import date, datetime, timedelta
from typing import Any, Sequence

from scripts.active_rectification_event_engine import (
    DOMAIN_CONFIG,
    _active_narayana,
    _active_vimshottari,
    _score_event,
)
import dasha_analyzer
import narayana_dasha
from scripts.rectification.refinement_packet import match_level

MAX_PROBES = 3
LEVEL_RANK = {"none": 0, "weak": 1, "medium": 2, "strong": 3}
SCORING_LAYERS = ("d1", "d9", "d10", "d4", "d5", "d24", "d7", "d12", "d2", "d11", "d30")
LAYER_DOMAIN = {
    "d9": "relationship",
    "d10": "career",
    "d4": "relocation",
    "d5": "education",
    "d24": "education",
    "d7": "family",
    "d12": "family",
    "d2": "finance",
    "d11": "finance",
    "d30": "health_pressure",
}
STAGE_DOMAIN = {
    "d9_refine": "relationship",
    "d10_refine": "career",
    "d4_refine": "relocation",
    "theme_refine": "relocation",
    "d5_refine": "education",
}
VOLUNTEER_ONLY = frozenset({"finance", "health_pressure"})
DOMAIN_CATALOG: dict[str, dict[str, Any]] = {
    "education": {
        "event_family": "升学、高考、转学或学习环境变化",
        "quality_family": "高考或重要考试发挥明显失常、压力很大",
        "kind": "education_milestone",
        "age_lo": 16,
        "age_hi": 18,
        "varga": "D5 / D24",
    },
    "relocation": {
        "event_family": "搬家、离乡或长期异地",
        "quality_family": "搬家、离乡或住宿结构明显变化",
        "kind": "home_change",
        "age_lo": 18,
        "age_hi": 24,
        "varga": "D4",
    },
    "relationship": {
        "event_family": "认真关系进入、结束或关系观明显转变",
        "quality_family": "认真关系进入、结束或关系观明显转变",
        "kind": "relationship_change",
        "age_lo": 21,
        "age_hi": 26,
        "varga": "D9",
    },
    "career": {
        "event_family": "入职、升职或职责明显加重",
        "quality_family": "入职、升职或职责明显加重",
        "kind": "career_change",
        "age_lo": 22,
        "age_hi": 30,
        "varga": "D10",
    },
    "family": {
        "event_family": "家人相关的明显变化",
        "quality_family": "家人相关的明显变化",
        "kind": "family_event",
        "age_lo": 18,
        "age_hi": 30,
        "varga": "D12 / D7 / D3",
    },
    "finance": {
        "event_family": "收入、资产或财务明显变化",
        "quality_family": "收入、资产或财务明显变化",
        "kind": "finance_change",
        "age_lo": 22,
        "age_hi": 32,
        "varga": "D2 / D11",
    },
    "health_pressure": {
        "event_family": "健康、事故或持续压力明显变化",
        "quality_family": "健康、事故或持续压力明显变化",
        "kind": "self_health_event",
        "age_lo": 16,
        "age_hi": 40,
        "varga": "D30",
    },
}


def _clock(value: str) -> int:
    return int(value[:2]) * 60 + int(value[3:5])


def _context_time(context: dict[str, Any]) -> str | None:
    feature = context.get("feature") if isinstance(context.get("feature"), dict) else {}
    raw = feature.get("time")
    if isinstance(raw, str) and len(raw) >= 5:
        return raw[:5]
    at = context.get("candidate_at")
    if isinstance(at, datetime):
        return at.strftime("%H:%M")
    return None


def _birth_year(value: object) -> int | None:
    text = str(value or "").strip()
    if len(text) < 4 or not text[:4].isdigit():
        return None
    year = int(text[:4])
    return year if 1900 <= year <= 2100 else None


def _event_year(event: dict[str, Any]) -> int | None:
    for key in ("date", "date_start", "occurred_from"):
        year = _birth_year(event.get(key))
        if year is not None:
            return year
    return None


def _year_label(year: int) -> str:
    return f"{year} 年前后"


def _age_band_year(birth_year: int, domain: str, today: date) -> int | None:
    catalog = DOMAIN_CATALOG.get(domain)
    if not catalog:
        return None
    age = (int(catalog["age_lo"]) + int(catalog["age_hi"])) // 2
    year = birth_year + age
    latest = min(today.year, birth_year + 80)
    earliest = birth_year + 5
    if year < earliest or year > latest:
        return None
    return year


def _probe_domains(
    scan: dict[str, Any],
    precision_current: str | None,
    events: Sequence[dict[str, Any]],
) -> list[str]:
    volunteered = {
        str(event.get("domain"))
        for event in events
        if isinstance(event, dict) and event.get("domain")
    }
    ordered: list[str] = []
    stage_domain = STAGE_DOMAIN.get(str(precision_current or ""))
    if stage_domain:
        ordered.append(stage_domain)
    for layer in SCORING_LAYERS:
        domain = LAYER_DOMAIN.get(layer)
        if not domain or domain in ordered:
            continue
        if not scan.get(f"{layer}_candidates_differ"):
            continue
        if domain in VOLUNTEER_ONLY and domain not in volunteered:
            continue
        ordered.append(domain)
    if not ordered and scan.get("d1_candidates_differ"):
        ordered.append("education")
    return ordered


def _static_contexts(built: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for context in built.get("static_contexts") or []:
        if isinstance(context, dict) and _context_time(context):
            rows.append(context)
    rows.sort(key=lambda item: _clock(str(_context_time(item))))
    return rows


def _pick_representatives(
    built: dict[str, Any],
    scan: dict[str, Any],
    candidate_times: Sequence[str],
    representative_time: str | None,
) -> tuple[dict[str, Any], dict[str, Any]] | None:
    contexts = _static_contexts(built)
    by_time = {_context_time(item): item for item in contexts}
    times = [str(_context_time(item)) for item in contexts]
    for transition in scan.get("transitions") or []:
        if not isinstance(transition, dict):
            continue
        layer = transition.get("layer")
        at = str(transition.get("at") or "")[:5]
        if layer not in SCORING_LAYERS or at not in by_time:
            continue
        index = times.index(at)
        left = times[index - 1] if index > 0 else at
        if left != at:
            return by_time[left], by_time[at]
    picked: list[str] = []
    for raw in [*candidate_times, representative_time]:
        time = str(raw or "")[:5]
        if len(time) >= 5 and time in by_time and time not in picked:
            picked.append(time)
    if len(picked) >= 2:
        return by_time[picked[0]], by_time[picked[-1]]
    if len(times) >= 2:
        return by_time[times[0]], by_time[times[-1]]
    return None


def _scoreable(context: dict[str, Any]) -> bool:
    chart = context.get("chart")
    planets = context.get("planet_longitudes")
    vargas = context.get("varga_charts")
    return (
        isinstance(chart, dict)
        and isinstance(planets, dict)
        and isinstance(vargas, dict)
        and isinstance(planets.get("Moon"), (int, float))
        and isinstance(context.get("ascendant_index"), int)
    )


def _candidate_at(context: dict[str, Any], birth_date: str) -> datetime | None:
    at = context.get("candidate_at")
    if isinstance(at, datetime):
        return at
    time = _context_time(context)
    if not time:
        return None
    try:
        return datetime.strptime(f"{birth_date} {time}", "%Y-%m-%d %H:%M")
    except ValueError:
        return None


def _has_domain_activation(rule_ids: Sequence[str]) -> bool:
    return any(
        ("_domain_" in str(item) or str(item).endswith("_domain_house") or str(item).endswith("_domain_lord") or str(item).endswith("_domain_varga"))
        and not str(item).startswith("no_")
        for item in rule_ids
    )


def _discriminates(left: Sequence[str], right: Sequence[str]) -> bool:
    left_rank = LEVEL_RANK.get(match_level(left), 0)
    right_rank = LEVEL_RANK.get(match_level(right), 0)
    if abs(left_rank - right_rank) >= 2:
        return True
    return _has_domain_activation(left) != _has_domain_activation(right)


def _tracks_present(rule_ids: Sequence[str]) -> tuple[bool, bool]:
    text = [str(item) for item in rule_ids]
    return (
        any(item.startswith("vim_") for item in text),
        any(item.startswith("narayana_") for item in text),
    )


def _vim_start_years(birth_date: str, moon_longitude: float, lo: int, hi: int) -> list[int]:
    nakshatra, progress, _ = dasha_analyzer.lon_to_nakshatra(float(moon_longitude))
    timeline, _, _, _ = dasha_analyzer.build_dasha_timeline(birth_date, nakshatra, progress)
    years: list[int] = []
    for major in timeline:
        start = major.get("start")
        if isinstance(start, datetime) and lo <= start.year <= hi:
            years.append(start.year)
        for minor in dasha_analyzer.build_antardasha(major):
            minor_start = minor.get("start")
            if isinstance(minor_start, datetime) and lo <= minor_start.year <= hi:
                years.append(minor_start.year)
    return years


def _narayana_start_years(
    ascendant_index: int,
    planet_longitudes: dict[str, float],
    birth_date: str,
    lo: int,
    hi: int,
) -> list[int] | None:
    periods = narayana_dasha.calc_narayana_mahadasha(ascendant_index, planet_longitudes)
    if not periods:
        return None
    birth = datetime.strptime(birth_date, "%Y-%m-%d")
    years: list[int] = []
    for major in periods:
        start_age = major.get("start_age")
        if not isinstance(start_age, (int, float)):
            return None
        year = (birth + timedelta(days=float(start_age) * 365.2425)).year
        if lo <= year <= hi:
            years.append(year)
        antars = narayana_dasha.calc_narayana_antardasha(periods, int(major["sign_idx"]))
        for minor in antars:
            minor_age = minor.get("start_age")
            if not isinstance(minor_age, (int, float)):
                continue
            minor_year = (birth + timedelta(days=float(minor_age) * 365.2425)).year
            if lo <= minor_year <= hi:
                years.append(minor_year)
    return years


def _boundary_years(left: list[int], right: list[int]) -> set[int]:
    years: set[int] = set()
    for one, two in zip(left, right):
        if abs(one - two) >= 1:
            years.add(one)
            years.add(two)
    return years


def _score_year(
    context: dict[str, Any],
    *,
    birth_date: str,
    domain: str,
    year: int,
) -> dict[str, Any] | None:
    catalog = DOMAIN_CATALOG[domain]
    prefixes, _ = DOMAIN_CONFIG[domain]
    varga_charts = context.get("varga_charts") or {}
    domain_vargas = [varga_charts.get(prefix) for prefix in prefixes]
    if any(item is None for item in domain_vargas):
        return None
    candidate_at = _candidate_at(context, birth_date)
    moon = (context.get("planet_longitudes") or {}).get("Moon")
    if candidate_at is None or not isinstance(moon, (int, float)):
        return None
    event_at = datetime(year, 7, 1)
    event = {
        "id": f"probe-{domain}-{year}",
        "domain": domain,
        "event_kind": catalog["kind"],
        "date": f"{year}-07-01",
        "precision": "year",
        "summary": catalog["event_family"],
    }
    try:
        vimshottari = _active_vimshottari(birth_date, float(moon), event_at)
        narayana = _active_narayana(
            int(context["ascendant_index"]),
            context["planet_longitudes"],
            candidate_at,
            event_at,
        )
    except (KeyError, TypeError, ValueError):
        return None
    if narayana[0] is None or narayana[1] is None:
        return None
    return _score_event(
        candidate_time=str(_context_time(context)),
        event=event,
        natal_chart=context["chart"],
        varga_charts=[item for item in domain_vargas if item is not None],
        vimshottari=vimshottari,
        narayana=narayana,
        arudha_padas=context.get("arudha_padas") or {},
    )


def _agent_brief(
    *,
    year_label: str,
    family: str,
    quality: bool = False,
    exam: bool = False,
) -> str:
    if exam:
        return (
            f"年份锁定 {year_label}。已有高考或考试经历。"
            "请写成一句自然语言，问那次是否发挥失常或压力特别大。"
            "不得改年份。"
        )
    if quality:
        return (
            f"年份锁定 {year_label}。已有相关经历。"
            f"请写成一句自然语言，问{family}有没有发生过。"
            "不得改年份。"
        )
    return (
        f"年份锁定 {year_label}。事件家族：{family}。"
        "请写成一句自然语言是/否题。"
        "不得改年份。"
    )


def _public_probe(
    *,
    year: int,
    domain: str,
    source: str,
    tracks: Sequence[str],
    tracks_agree: bool,
    user_meaning: str,
    event_family: str,
) -> dict[str, Any]:
    return {
        "year": year,
        "year_label": _year_label(year),
        "domain": domain,
        "event_family": event_family,
        "source": source,
        "tracks": list(tracks),
        "tracks_agree": tracks_agree,
        "unique_minute_claim": False,
        "user_meaning": user_meaning,
        "role": "distinguish" if source == "known_event_quality" else "reverse_verify",
    }


def _quality_probes(
    events: Sequence[dict[str, Any]],
    domains: Sequence[str],
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    seen: set[tuple[str, int]] = set()
    for event in events:
        if not isinstance(event, dict):
            continue
        domain = str(event.get("domain") or "")
        year = _event_year(event)
        if domain not in domains or year is None or (domain, year) in seen:
            continue
        seen.add((domain, year))
        summary = str(event.get("summary") or "")
        family = str(DOMAIN_CATALOG[domain]["quality_family"])
        exam = domain == "education" and ("高考" in summary or "考试" in summary)
        rows.append(_public_probe(
            year=year,
            domain=domain,
            source="known_event_quality",
            tracks=("vimshottari", "narayana"),
            tracks_agree=True,
            user_meaning=_agent_brief(
                year_label=_year_label(year),
                family=family,
                quality=True,
                exam=exam,
            ),
            event_family=family,
        ))
    return rows


def _evaluate_year(
    left: dict[str, Any],
    right: dict[str, Any],
    *,
    birth_date: str,
    domain: str,
    year: int,
    source: str,
) -> dict[str, Any] | None:
    scored_left = _score_year(left, birth_date=birth_date, domain=domain, year=year)
    scored_right = _score_year(right, birth_date=birth_date, domain=domain, year=year)
    if scored_left is None or scored_right is None:
        return None
    left_rules = scored_left.get("rule_ids") or []
    right_rules = scored_right.get("rule_ids") or []
    if not _discriminates(left_rules, right_rules):
        return None
    stronger = left_rules if LEVEL_RANK[match_level(left_rules)] >= LEVEL_RANK[match_level(right_rules)] else right_rules
    vim_hit, narayana_hit = _tracks_present(stronger)
    return _public_probe(
        year=year,
        domain=domain,
        source=source,
        tracks=("vimshottari", "narayana"),
        tracks_agree=vim_hit and narayana_hit,
        user_meaning=_agent_brief(
            year_label=_year_label(year),
            family=str(DOMAIN_CATALOG[domain]["event_family"]),
        ),
        event_family=str(DOMAIN_CATALOG[domain]["event_family"]),
    )


def discriminating_event_probes(
    request: dict[str, Any],
    built: dict[str, Any],
    *,
    scan: dict[str, Any],
    candidate_times: Sequence[str],
    representative_time: str | None,
    precision_current: str | None = None,
    today: date | None = None,
) -> list[dict[str, Any]]:
    birth_date = str(request.get("birth_date") or "").strip()
    birth_year = _birth_year(birth_date)
    if birth_year is None:
        return []
    try:
        datetime.strptime(birth_date, "%Y-%m-%d")
    except ValueError:
        return []
    now = today or date.today()
    events = [item for item in (request.get("events") or []) if isinstance(item, dict)]
    domains = _probe_domains(scan, precision_current, events)
    if not domains:
        return []
    probes = _quality_probes(events, domains)
    covered_domains = {row["domain"] for row in probes}
    pair = _pick_representatives(built, scan, candidate_times, representative_time)
    lo, hi = birth_year + 5, min(now.year, birth_year + 80)
    can_score = (
        pair is not None
        and _scoreable(pair[0])
        and _scoreable(pair[1])
    )
    dasha_domains: set[str] = set()
    if can_score and pair is not None:
        left, right = pair
        left_moon = float(left["planet_longitudes"]["Moon"])
        right_moon = float(right["planet_longitudes"]["Moon"])
        vim_years = _boundary_years(
            _vim_start_years(birth_date, left_moon, lo, hi),
            _vim_start_years(birth_date, right_moon, lo, hi),
        )
        left_narayana = _narayana_start_years(int(left["ascendant_index"]), left["planet_longitudes"], birth_date, lo, hi)
        right_narayana = _narayana_start_years(int(right["ascendant_index"]), right["planet_longitudes"], birth_date, lo, hi)
        narayana_years: set[int] = set()
        if left_narayana is not None and right_narayana is not None:
            narayana_years = _boundary_years(left_narayana, right_narayana)
        for domain in domains:
            if domain in covered_domains:
                continue
            boundary = sorted((vim_years | narayana_years) & set(range(lo, hi + 1)))
            found = None
            for year in boundary:
                found = _evaluate_year(
                    left, right, birth_date=birth_date, domain=domain, year=year, source="dasha_boundary",
                )
                if found:
                    break
            if found is None:
                midpoint = _age_band_year(birth_year, domain, now)
                if midpoint is not None:
                    found = _evaluate_year(
                        left, right, birth_date=birth_date, domain=domain, year=midpoint, source="dasha_activation",
                    )
            if found:
                probes.append(found)
                dasha_domains.add(domain)
                covered_domains.add(domain)
    for domain in domains:
        if domain in covered_domains or domain in dasha_domains:
            continue
        year = _age_band_year(birth_year, domain, now)
        if year is None:
            continue
        probes.append(_public_probe(
            year=year,
            domain=domain,
            source="age_band",
            tracks=("vimshottari", "narayana"),
            tracks_agree=False,
            user_meaning=_agent_brief(
            year_label=_year_label(year),
            family=str(DOMAIN_CATALOG[domain]["event_family"]),
        ),
            event_family=str(DOMAIN_CATALOG[domain]["event_family"]),
        ))
        covered_domains.add(domain)
    public: list[dict[str, Any]] = []
    seen: set[tuple[str, int, str]] = set()
    for row in probes:
        key = (str(row["domain"]), int(row["year"]), str(row["source"]))
        encoded = str(row)
        if key in seen or "points" in encoded:
            continue
        seen.add(key)
        public.append(row)
        if len(public) >= MAX_PROBES:
            break
    return public