#!/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()