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): upstream
hands the annual JD to the selector, which then imports PyJHora in-process
when the interpreter has it. PyJHora is AGPL and stays an external reference
engine (requirements-reference-engines.txt); the API image and CI do not have
it, so a developer machine that does would pick a different year lord than
production. The product path therefore always uses the native proxy
(`engine: native_proxy`); set JYOTISH_YEAR_LORD_PYJHORA_REFERENCE=1 to pass the
annual JD and let the upstream PyJHora lookup run (research comparison only).
"""

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 SIGNS, SIGN_LORDS, calc_panchadhikari_year_lord
except ImportError:  # pragma: no cover - package import
    from scripts.tajika import SIGNS, SIGN_LORDS, 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"
FALLBACK_BASIS = "fallback_muntha_sign_lord_after_panchadhikari_block"

ENGINE_NOTE = {
    "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 == FALLBACK_BASIS:
        return ("年盘不全，暂用 Muntha 主星", "Annual chart incomplete; Muntha sign lord used for now")
    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 _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_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":
        lord = SIGN_LORDS[SIGNS[muntha_sign_idx]]
        data = {
            "year_lord": lord,
            "year_lord_sign_idx": muntha_sign_idx,
            "year_lord_sign": SIGNS[muntha_sign_idx],
            "selection_basis": FALLBACK_BASIS,
            "fallback_reason": data.get("reason", "panchadhikari_blocked"),
        }
    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