Jyotisha/scripts/technique_source_adapters.py

#!/usr/bin/env python3
"""Refresh named-technique candidates from public catalogues and local book index."""
from __future__ import annotations

import argparse
import json
import re
import urllib.request
from html.parser import HTMLParser
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
REF = ROOT / "references"
PYJHORA_DASHA_RAW = "https://raw.githubusercontent.com/naturalstupid/PyJHora/main/src/jhora/horoscope/dhasa/README.md"
JHORA_FEATURES = "https://vedicastrologer.org/jh/features.htm"


def fetch_text(url: str) -> str:
    request = urllib.request.Request(url, headers={"User-Agent": "jyotish-technique-source-adapter/1.0"})
    with urllib.request.urlopen(request, timeout=15) as response:
        return response.read().decode("utf-8-sig", errors="ignore")


def slug(value: str) -> str:
    value = value.lower().replace("dhasa", "dasha").replace("dasa", "dasha")
    value = value.replace("prasna", "prashna").replace("sudharsana", "sudarsana")
    value = value.replace("chathuraaseethi", "chaturashiti").replace("chaturaaseethi", "chaturashiti")
    return re.sub(r"[^a-z0-9]+", "_", value).strip("_")


class _TextLines(HTMLParser):
    def __init__(self) -> None:
        super().__init__()
        self.parts: list[str] = []

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag in {"br", "p", "div", "li", "h1", "h2", "h3", "h4", "tr"}:
            self.parts.append("\n")

    def handle_data(self, data: str) -> None:
        self.parts.append(data)

    def lines(self) -> list[str]:
        return [" ".join(line.split()) for line in "".join(self.parts).splitlines() if " ".join(line.split())]


def parse_pyjhora_dashas(markdown: str) -> list[dict[str, Any]]:
    rows = []
    headings = list(re.finditer(r"^####\s+(.+)$", markdown, flags=re.MULTILINE))
    for index, match in enumerate(headings):
        heading = match.group(1)
        if "jhora.horoscope.dhasa" not in heading and "dhasa" not in heading.lower():
            continue
        block_end = headings[index + 1].start() if index + 1 < len(headings) else len(markdown)
        block = markdown[match.end():block_end]
        clean = re.split(r"\s+jhora\.horoscope|:\s*jhora\.horoscope", heading, maxsplit=1)[0].strip(" :")
        clean = clean.replace("Dhasa", "Dasha").replace("Dasa", "Dasha")
        module_match = re.search(r"(jhora\.horoscope\.dhasa[\w.]+)", heading)
        module = module_match.group(1) if module_match else ""
        skill = "tajika-annual" if ".annual." in module.lower() or clean.lower().startswith(("mudda", "patyayini", "varsha ")) else ("jaimini-timing" if ".raasi." in module.lower() else "specialized-jyotish")
        methods = []
        parameters = []
        for signature in re.findall(r"^#####\s+(.+)$", block, flags=re.MULTILINE):
            signature = signature.replace(r"\_", "_").strip().rstrip(":")
            methods.append(signature)
            argument_match = re.search(r"\((.*)\)", signature)
            if argument_match:
                for argument in argument_match.group(1).split(","):
                    name = argument.strip().lstrip("*").split("=", 1)[0].strip()
                    if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) and name not in parameters:
                        parameters.append(name)
        for name in re.findall(r"@param\s+([A-Za-z_][A-Za-z0-9_]*)", block.replace(r"\_", "_")):
            if name not in parameters:
                parameters.append(name)
        applicability_hints = []
        for line in block.splitlines():
            compact = " ".join(line.strip().strip('"').split())
            if compact and re.search(r"\b(applicab(?:le|ility)|condition(?:al)?|only when|automatically determine)\b", compact, re.IGNORECASE):
                applicability_hints.append(compact[:240])
        rows.append({
            "candidate_id": slug(clean),
            "name": clean,
            "family": "dasha",
            "skill_family": skill,
            "source_id": "pyjhora_dasha_raw",
            "documented_module": module,
            "documented_methods": methods,
            "documented_parameters": parameters,
            "applicability_hints": applicability_hints[:4],
        })
    return _dedupe(rows)


def parse_jhora_features(html: str) -> list[dict[str, Any]]:
    parser = _TextLines()
    parser.feed(html)
    sections = {"Planets and mathematical points", "Chakras", "Annual Charts", "Krishnamoorthy Paddhati (KP)", "Panchanga"}
    section = ""
    rows = []
    for line in parser.lines():
        if line in sections:
            section = line
            continue
        if line in {"Basic Calculations", "Strengths", "Dasas", "Transits", "Mundane", "Miscellaneous", "Learning Aids", "Conveniences"}:
            section = ""
            continue
        if not line.startswith("-"):
            continue
        text = line.lstrip("- ")
        if section == "Chakras" and "chakra" in text.lower():
            name = re.split(r"\s+from\s+|\s*\(", text, maxsplit=1, flags=re.IGNORECASE)[0].strip()
            rows.append({"candidate_id": slug(name), "name": name, "family": "chakra", "skill_family": "specialized-jyotish", "source_id": "jhora_features_live"})
        elif section == "Annual Charts" and "pravesha" in text.lower():
            for name in re.findall(r"(?:Tithi|Yoga|Nakshatra) Pravesha", text, flags=re.IGNORECASE):
                proper = " ".join(word.capitalize() for word in name.split())
                rows.append({"candidate_id": slug(proper), "name": proper, "family": "pravesha", "skill_family": "tajika-annual", "source_id": "jhora_features_live"})
        elif section == "Planets and mathematical points" and "sphuta" in text.lower():
            name = re.split(r"\s+of\s+", text, maxsplit=1, flags=re.IGNORECASE)[0].strip()
            rows.append({"candidate_id": slug(name), "name": name, "family": "sphuta", "skill_family": "prashna-muhurta", "source_id": "jhora_features_live"})
        elif section == "Krishnamoorthy Paddhati (KP)":
            if "sub lords" in text.lower() and "cusp" in text.lower():
                candidate_id, name = "kp_cusp_planet_sub_chain", "KP cusp and planet star/sub chain"
            else:
                candidate_id, name = "kp_significator_occupancy_lookup", "KP star/sub occupancy and significator lookup"
            rows.append({"candidate_id": candidate_id, "name": name, "family": "kp", "skill_family": "kp-event", "source_id": "jhora_features_live"})
        elif section == "Panchanga":
            lower = text.lower()
            if lower.startswith("show the panchanga"):
                candidate_id, name = "daily_panchanga_with_end_times", "Daily Panchanga with limb end times"
            elif "panchanga for a month" in lower:
                candidate_id, name = "monthly_panchanga", "Monthly Panchanga"
            elif "ephemeris for a month" in lower:
                candidate_id, name = "monthly_ephemeris", "Monthly ephemeris"
            else:
                candidate_id, name = "monthly_exact_transit_events", "Monthly exact transit and aspect events"
            rows.append({"candidate_id": candidate_id, "name": name, "family": "panchanga", "skill_family": "prashna-muhurta", "source_id": "jhora_features_live"})
    return _dedupe(rows)


def parse_local_book_topics(data: dict[str, Any]) -> list[dict[str, Any]]:
    return [{
        "topic_id": topic_id,
        "aliases": topic.get("aliases", []),
        "source_role": topic.get("source_role"),
        "priority_books": topic.get("priority_books", []),
        "status": "local_topic_candidate",
    } for topic_id, topic in data.get("topics", {}).items()]


def _dedupe(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    unique = {}
    for row in rows:
        unique.setdefault(row["candidate_id"], row)
    return list(unique.values())


def _canonical_lookup() -> dict[str, str]:
    catalog = json.loads((REF / "external_technique_catalog.json").read_text(encoding="utf-8"))
    ontology = json.loads((REF / "technique_ontology.json").read_text(encoding="utf-8"))
    lookup = {}
    for row in catalog["techniques"]:
        technique_id = row["canonical_id"]
        for name in [technique_id, row["name"], *ontology.get("aliases", {}).get(technique_id, [])]:
            lookup[slug(name)] = technique_id
    for technique_id, names in ontology.get("aliases", {}).items():
        lookup[slug(technique_id)] = technique_id
        for name in names:
            lookup[slug(name)] = technique_id
    return lookup


def resolve_candidates(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    lookup = _canonical_lookup()
    result = []
    for row in rows:
        candidate = row["candidate_id"]
        canonical = lookup.get(candidate)
        result.append({**row, "canonical_id": canonical, "resolution_status": "resolved" if canonical else "needs_canonicalization"})
    return result


def refresh() -> tuple[dict[str, Any], dict[str, Any]]:
    public_rows = resolve_candidates(parse_pyjhora_dashas(fetch_text(PYJHORA_DASHA_RAW)) + parse_jhora_features(fetch_text(JHORA_FEATURES)))
    public = {
        "schema": "jyotish.external_technique_discovery_snapshot.v1",
        "sources": {"pyjhora_dasha_raw": PYJHORA_DASHA_RAW, "jhora_features_live": JHORA_FEATURES},
        "summary": {"candidate_count": len(public_rows), "resolved_count": sum(row["resolution_status"] == "resolved" for row in public_rows), "unresolved_count": sum(row["resolution_status"] != "resolved" for row in public_rows)},
        "candidates": public_rows,
    }
    books = parse_local_book_topics(json.loads((REF / "jyotish_book_topic_index.json").read_text(encoding="utf-8")))
    local = {"schema": "jyotish.local_book_technique_candidates.v1", "summary": {"topic_count": len(books), "priority_book_links": sum(len(row["priority_books"]) for row in books)}, "topics": books}
    return public, local


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--write", action="store_true")
    args = parser.parse_args()
    public, local = refresh()
    if args.write:
        (REF / "external_technique_discovery_snapshot.json").write_text(json.dumps(public, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        (REF / "local_book_technique_candidates.json").write_text(json.dumps(local, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({"public": public["summary"], "local_books": local["summary"]}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()