Jyotisha/scripts/tajika_year_lord.py

#!/usr/bin/env python3
"""Tajika year lord (Varshesha) by the Panchadhikari rule (BUG-1214).

Astrologer round-4 ruling 2026-10-03, Q3: five candidates (day/night new-year
lord, natal Lagna lord, Muntha lord, annual Lagna lord, Tri-Rasi lord) ->
those casting a benefic Tajika aspect on the annual Lagna, strongest by
Panchavargiya -> else those casting a malefic aspect, strongest -> else the
strongest of all (`panchadhikari_pvb_fallback_all_candidates`).

The selection itself is `tajika.calc_panchadhikari_year_lord`, ported from
upstream yinduzhanxing c7117f92. This module is the one place both annual
producers (`solar_return.solar_return_full_report` and
`jyotish_engine.cmd_tajika`) call, so `annual_tajika_pack`'s conflict gate
compares two runs of the same rule. Muntha stays natal Lagna + completed
years (BUG-1212); it is handed to the selector explicitly.

Panchavargiya engine (local divergence from upstream, BUG-1214 / BUG-1217):
upstream hands the annual JD to the selector, which then imports PyJHora
in-process when the interpreter has it, else falls back to a D1/D2/D3/D9/D12
dignity proxy. PyJHora is AGPL and stays an external reference engine
(requirements-reference-engines.txt); the API image and CI do not have it, so
the product path never hands it the JD. Since BUG-1217 (astrologer round-5
Q5) the candidates are compared by the Tajika five-fold Panchavargiya
(upstream `_calc_native_panchavargiya_bala`: Kshetra/graha, Uchcha, Hadda,
Drekkana, Navamsa; raw out of 80, score = raw / 4 out of 20;
`engine: tajika_native_five_fold`); the dignity proxy no longer selects a year
lord. Set JYOTISH_YEAR_LORD_PYJHORA_REFERENCE=1 to pass the annual JD and let
the upstream PyJHora lookup run (research comparison only).

No stand-in year lord (BUG-1223, product owner 2026-10-04, aligned with
upstream 03bea6ed `docs/research/tajika_shared_entrypoint_repair_2026_10_03.md`
item 3): when Panchadhikari is blocked (annual chart incomplete) the result is
`status: "blocked"` with the reason, `year_lord: None`, and the sentence
"年主暂无法确定（原因）"; the Muntha sign lord is no longer put in its place.
Mudda is independent of the year lord (BUG-1216) and is unaffected.
"""

from __future__ import annotations

import contextlib
import io
import os
from typing import Any, Dict, List, Optional, Tuple

try:  # scripts/ on sys.path
    from tajika import calc_panchadhikari_year_lord
except ImportError:  # pragma: no cover - package import
    from scripts.tajika import calc_panchadhikari_year_lord

PYJHORA_REFERENCE_ENV = "JYOTISH_YEAR_LORD_PYJHORA_REFERENCE"
SOURCE = "tajika.calc_panchadhikari_year_lord (upstream yinduzhanxing c7117f92)"
MUNTHA_BASIS = "natal_ascendant_plus_completed_years"
BLOCKED_BASIS = "panchadhikari_blocked_no_year_lord"
# Reason token -> (zh, en) for "年主暂无法确定（原因）".
BLOCKED_REASON_TEXT = {
    "sun_or_moon_missing_for_panchadhikari": ("年盘缺太阳或月亮位置", "the annual chart lacks the Sun or Moon position"),
    "tajika_five_fold_panchavargiya_unavailable": ("年盘缺行星位置，五分力量无法计算", "planet positions are missing from the annual chart, so the five-fold strength cannot be computed"),
    "no_panchadhikari_candidates_available": ("五类候选星都不在年盘上", "none of the five candidates is in the annual chart"),
    "solar_return_chart_unavailable": ("年盘无法计算", "the annual chart could not be computed"),
    "muntha_sign_required_for_year_lord": ("缺少 Muntha 星座", "the Muntha sign is missing"),
}
BLOCKED_REASON_DEFAULT = ("年盘数据不全", "the annual chart data are incomplete")

ENGINE_NOTE = {
    "tajika_native_five_fold": ("按 Tajika 五分力量", "by Tajika five-fold strength"),
    "native_proxy": ("Panchavargiya 为本地近似计算", "Panchavargiya is a local approximation"),
    "pyjhora": ("Panchavargiya 取 PyJHora 外部参照", "Panchavargiya from the PyJHora reference engine"),
    "pyjhora_sidecar": ("Panchavargiya 取 PyJHora 外部参照", "Panchavargiya from the PyJHora reference engine"),
}


def _names(rows: List[Dict[str, Any]], sep: str) -> str:
    return sep.join(str(row.get("planet")) for row in rows)


def selection_basis_text(data: Dict[str, Any]) -> Tuple[str, str]:
    """One sentence (zh, en) saying why this planet is the year lord."""
    basis = data.get("selection_basis")
    lord = data.get("year_lord")
    if basis == BLOCKED_BASIS or data.get("status") == "blocked":
        return blocked_text(data.get("reason"))
    candidates = [row for row in data.get("candidates") or [] if isinstance(row, dict)]
    benefic = [row for row in candidates if row.get("benefic_aspects_asc")]
    malefic = [row for row in candidates if row.get("malefic_aspects_asc")]
    if basis == "panchadhikari_single_benefic":
        zh = f"五类候选中只有{lord}以吉相位照年盘上升，取为年主"
        en = f"Of the five candidates only {lord} casts a benefic aspect on the annual Lagna, so it is the year lord"
    elif basis == "panchadhikari_benefic_shortlist_pvb":
        zh = f"{_names(benefic, '、')}以吉相位照年盘上升，取其中 Panchavargiya 最强的{lord}"
        en = f"{_names(benefic, ', ')} cast a benefic aspect on the annual Lagna; the strongest by Panchavargiya, {lord}, is the year lord"
    elif basis == "panchadhikari_single_malefic":
        zh = f"没有候选以吉相位照年盘上升；只有{lord}以凶相位照上升，取为年主"
        en = f"No candidate casts a benefic aspect on the annual Lagna; only {lord} casts a malefic one, so it is the year lord"
    elif basis == "panchadhikari_malefic_shortlist_pvb":
        zh = f"没有候选以吉相位照年盘上升；{_names(malefic, '、')}以凶相位照上升，取其中 Panchavargiya 最强的{lord}"
        en = (f"No candidate casts a benefic aspect on the annual Lagna; {_names(malefic, ', ')} cast a malefic one; "
              f"the strongest by Panchavargiya, {lord}, is the year lord")
    elif basis == "panchadhikari_pvb_fallback_all_candidates":
        zh = f"五类候选都不照年盘上升，取全体 Panchavargiya 最强的{lord}"
        en = f"None of the five candidates aspects the annual Lagna; the strongest of all by Panchavargiya, {lord}, is the year lord"
    else:
        return ("-", "-")
    note = ENGINE_NOTE.get(str(data.get("panchavargiya_engine")))
    if note:
        zh += f"（{note[0]}）"
        en += f" ({note[1]})"
    return zh, en


def blocked_text(reason: Any) -> Tuple[str, str]:
    """「年主暂无法确定（原因）」 and its English twin."""
    zh, en = BLOCKED_REASON_TEXT.get(str(reason), BLOCKED_REASON_DEFAULT)
    return f"年主暂无法确定（{zh}）", f"Year lord cannot be determined yet ({en})"


def blocked_year_lord(reason: Any, detail: Any = None) -> Dict[str, Any]:
    """Blocked year lord: no planet, reason and sentence only (BUG-1223)."""
    data: Dict[str, Any] = {
        "status": "blocked",
        "reason": str(reason or "panchadhikari_blocked"),
        "year_lord": None,
        "selection_basis": BLOCKED_BASIS,
        "muntha_basis": MUNTHA_BASIS,
        "candidate_list": [],
        "source": SOURCE,
    }
    if detail is not None:
        data["detail"] = detail
    data["selection_basis_text"], data["selection_basis_text_en"] = blocked_text(data["reason"])
    return data


def _candidate_list(candidates: List[Dict[str, Any]], winner: Optional[str]) -> List[Dict[str, Any]]:
    return [{
        "source": row.get("source"),
        "planet": row.get("planet"),
        "planet_sign": row.get("planet_sign"),
        "benefic_aspects_asc": bool(row.get("benefic_aspects_asc")),
        "malefic_aspects_asc": bool(row.get("malefic_aspects_asc")),
        "panchavargiya_score": row.get("panchavargiya_score"),
        "panchavargiya_raw_total": row.get("panchavargiya_raw_total"),
        "panchavargiya_engine": row.get("panchavargiya_engine"),
        "selected": row.get("planet") == winner,
    } for row in candidates if isinstance(row, dict)]


def _select(sr, natal_asc_sign_idx, completed_years, muntha_sign_idx,
            ayanamsa_name, birth_lat, birth_lon, birth_tz) -> Dict[str, Any]:
    info = sr.get("solar_return") or {}
    location = sr.get("annual_location") or {}
    return calc_panchadhikari_year_lord(
        sr.get("planet_lons") or {},
        annual_asc_sign_idx=sr["asc_sign_idx"],
        natal_lagna_sign_idx=int(natal_asc_sign_idx),
        years_from_dob=int(completed_years),
        annual_jd=info.get("jd_ut") if os.environ.get(PYJHORA_REFERENCE_ENV) == "1" else None,
        annual_dt=info.get("dt_local") or info.get("dt_ut"),
        lat=location.get("latitude", birth_lat),
        lon=location.get("longitude", birth_lon),
        tz=location.get("utc_offset", birth_tz),
        ayanamsa_name=ayanamsa_name,
        muntha_sign_idx=muntha_sign_idx,
    )


def select_annual_year_lord(
    sr: Dict[str, Any],
    *,
    natal_asc_sign_idx: int,
    completed_years: int,
    ayanamsa_name: Optional[str],
    birth_lat: float,
    birth_lon: float,
    birth_tz: float,
) -> Dict[str, Any]:
    """Year lord for one solar-return chart (`solar_return.calc_solar_return_chart`)."""
    muntha_sign_idx = (int(natal_asc_sign_idx) + int(completed_years)) % 12
    # PyJHora (optional reference engine, absent from the API image) prints
    # on import; keep stdout clean for the CLI JSON, as upstream cmd_tajika does.
    with contextlib.redirect_stdout(io.StringIO()):
        data = _select(sr, natal_asc_sign_idx, completed_years, muntha_sign_idx,
                       ayanamsa_name, birth_lat, birth_lon, birth_tz)
    if data.get("status") != "partial_verified":
        return blocked_year_lord(data.get("reason", "panchadhikari_blocked"), data.get("detail"))
    data["muntha_basis"] = MUNTHA_BASIS
    data["candidate_list"] = _candidate_list(data.get("candidates") or [], data.get("year_lord"))
    data["source"] = SOURCE
    zh, en = selection_basis_text(data)
    data["selection_basis_text"] = zh
    data["selection_basis_text_en"] = en
    return data