"""Pure helpers for the consultation evidence-card research. No network, no engine mutation. Classification partitions a JSON value so the five category sizes sum to the canonical serialization length. """ from __future__ import annotations import json import re from collections import Counter from pathlib import Path from typing import Any CATEGORIES = ("core", "status", "not_applicable", "research", "western") TOKEN_CHARS_PER_TOKEN = 3.5 TOKEN_METHOD = ( "chars/3.5. The 2026-09-27 diagnosis measured about 144000 projected " "characters as about 40000 tokens (3.6 chars/token). This estimate uses " "3.5 so the research tables stay comparable. It is not a vendor tokenizer." ) REFERENCE_DATE = "2026-09-27" AYANAMSA = "raman" NODE_MODE = "mean" QUESTIONS = ( {"id": "career", "domain": "career", "question": "未来一年，事业和收入该关注什么？"}, {"id": "marriage", "domain": "marriage", "question": "我的关系模式是什么？"}, {"id": "wealth", "domain": "wealth", "question": "我的财富增长方式和风险点是什么？"}, {"id": "health", "domain": "health", "question": "我近期的身心压力模式和调节重点是什么？"}, {"id": "education", "domain": "education", "question": "我的学习优势、瓶颈和进阶方向是什么？"}, {"id": "migration", "domain": "migration", "question": "迁居、置业或海外发展更适合怎样规划？"}, {"id": "family", "domain": "family", "question": "我的家庭关系与责任边界该如何理解？"}, {"id": "general", "domain": "general", "question": "请综合说明我当前最值得关注的主题。"}, {"id": "parents", "domain": "family", "question": "我和父母关系如何"}, {"id": "children", "domain": "family", "question": "我和子女的关系如何"}, ) CHART_SOURCES = ( { "id": "steve_jobs", "label": "Steve Jobs", "path": "references/real_case_calibration/minute_rectification_development_v1.json", "case_id": "steve_jobs_1955_development", }, { "id": "barack_obama", "label": "Barack Obama", "path": "references/real_case_calibration/minute_rectification_holdout_v4.json", "case_id": "barack_obama_1961_aa_v4_holdout", }, { "id": "elizabeth_taylor", "label": "Elizabeth Taylor", "path": "references/real_case_calibration/minute_rectification_holdout_v4.json", "case_id": "elizabeth_taylor_1932_aa_v4_holdout", }, ) STATUS_TECHNIQUES = { "mevg / global web evidence", "real case calibration", "vedastro main entry overview", "vedastro cloud state", "external oracle progress", "interpretation source pack", "module execution audit", "exact cusp oracle", "blind technical mode", "timing precision gate", "evidence packet", "shadbala boundary", } RESEARCH_TECHNIQUES = { "research d-n through d60", "extended vargas d81/d108/d144", "ashtottari dasha", } NOT_APPLICABLE_TECHNIQUES = { "prashna chart", "ashtakoota matching", "adhana / niseka", "muhurta election", } WESTERN_PREFIX = "western" STATUS_KEYS = { "methodology", "presentation", "rectification", "answer_policy", "user_facing_limitation", "hard_blockers", "missing_route_layers", "available_layers", "must_use_layers", "technique_truth", "reference_transparency", "shadbala_boundary", "vedastro_cross_check", "vedastro_gateway", "commercial_evidence_status", "success", "omitted_domains", "status", "packet_version", "schema_error", } # Domain card draft. Values are copied from the projection; nothing is rewritten. CARD_SPECS: dict[str, dict[str, Any]] = { "career": { "label": "事业", "vargas": ["D10"], "houses": [10], "arudha": ["A10"], "karakas": ["AmK"], "planets": [], "yogas": True, "transits": True, "ashtakavarga": False, "shadbala": True, }, "marriage": { "label": "婚恋", "vargas": ["D9"], "houses": [7], "arudha": ["UL", "A7"], "karakas": ["DK"], "planets": ["venus", "jupiter"], "yogas": False, "transits": True, "ashtakavarga": False, "shadbala": False, }, "wealth": { "label": "财富", "vargas": ["D2", "D11"], "houses": [2, 11, 9, 5], "arudha": [], "karakas": [], "planets": [], "yogas": True, "transits": False, "ashtakavarga": True, "shadbala": True, }, "health": { "label": "健康", "vargas": ["D6", "D8", "D30"], "houses": [6, 8], "arudha": [], "karakas": [], "planets": [], "yogas": False, "transits": True, "ashtakavarga": False, "shadbala": True, }, "education": { "label": "学习", "vargas": ["D5", "D24"], "houses": [5, 9], "arudha": [], "karakas": [], "planets": ["mercury", "jupiter"], "yogas": False, "transits": False, "ashtakavarga": False, "shadbala": False, }, "migration": { "label": "迁居", "vargas": ["D4", "D12"], "houses": [4, 12], "arudha": [], "karakas": [], "planets": [], "yogas": False, "transits": False, "ashtakavarga": False, "shadbala": False, }, "family": { "label": "家庭（现路由，父母与子女混在一起）", "vargas": ["D7", "D12"], "houses": [4, 5, 9], "arudha": [], "karakas": [], "planets": [], "yogas": False, "transits": False, "ashtakavarga": False, "shadbala": False, }, "parents": { "label": "父母", "vargas": ["D12"], "houses": [4, 9], "arudha": [], "karakas": [], "planets": ["sun", "moon"], "yogas": False, "transits": False, "ashtakavarga": False, "shadbala": False, "basis": [ "skills/jyotish-vedic-astrology/references/strict-workflow-router.md shared baseline: D12 for family/ancestral themes", "references/birth-time-rectification-decision-tree.md section 4: 父母/家族 -> D12", ], }, "children": { "label": "子女", "vargas": ["D7"], "houses": [5], "arudha": [], "karakas": ["PK"], "planets": ["jupiter"], "yogas": False, "transits": False, "ashtakavarga": False, "shadbala": False, "basis": [ "skills/jyotish-vedic-astrology/references/strict-workflow-router.md shared baseline: D7 for children", "references/birth-time-rectification-decision-tree.md section 4: 子女/生育 -> D7", ], }, "general": { "label": "综合", "vargas": ["D9", "D10", "D2"], "houses": [1, 10, 7, 2], "arudha": ["A10"], "karakas": [], "planets": [], "yogas": True, "transits": True, "ashtakavarga": True, "shadbala": True, }, } PART_B_BASE = ( "functional_benefic_malefic", "vimshottari_md_ad_pd", "narayana_current", "ascendant_degree", ) def dumps(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":")) def estimate_tokens(text: str) -> dict[str, Any]: chars = len(text) return { "chars": chars, "token_estimate": int(round(chars / TOKEN_CHARS_PER_TOKEN)), "token_method": TOKEN_METHOD, } def _norm(value: str) -> str: return re.sub(r"\s+", " ", value).strip().lower() def classify_audit_row(row: dict[str, Any]) -> str: technique = _norm(str(row.get("technique") or "")) system = _norm(str(row.get("system") or "")) status = _norm(str(row.get("status") or "")) if system == "western" or technique.startswith(WESTERN_PREFIX): return "western" if technique in RESEARCH_TECHNIQUES: return "research" if status == "not_applicable" or technique in NOT_APPLICABLE_TECHNIQUES: return "not_applicable" if technique in STATUS_TECHNIQUES or "vedastro" in technique or "mevg" in technique: return "status" return "core" def classify(path: tuple[str, ...], value: Any) -> str: joined = ".".join(path).lower() if "western_spectrum" in joined or joined.startswith("western"): return "western" if "research_dn" in joined or "extended" in joined and "varga" in joined: return "research" if isinstance(value, dict) and "technique" in value and "status" in value and "technique_audit" in joined: return classify_audit_row(value) leaf = path[-1].lower() if path else "" if leaf in STATUS_KEYS or leaf.endswith("_boundary") or leaf.endswith("_status"): return "status" if isinstance(value, str) and _norm(value) == "not_applicable": return "not_applicable" return "core" def _whole(path: tuple[str, ...], value: Any) -> str | None: joined = ".".join(path).lower() if joined.endswith("western_spectrum"): return "western" if joined.endswith("research_dn") or joined.endswith("extended") and "varga_spectrum" in joined: return "research" if joined.endswith("methodology") or joined.endswith("presentation") or joined.endswith("rectification"): return "status" if ( isinstance(value, dict) and "technique" in value and "status" in value and any(part == "technique_audit_table" for part in path) ): return classify_audit_row(value) return None def _dominant(sizes: Counter[str]) -> str: if not sizes: return "status" return max(CATEGORIES, key=lambda name: (sizes.get(name, 0), -CATEGORIES.index(name))) def account(value: Any, path: tuple[str, ...] = ()) -> Counter[str]: """Partition dumps(value) across the five categories. Sum equals len(dumps(value)).""" whole = _whole(path, value) if whole is not None: return Counter({whole: len(dumps(value))}) if isinstance(value, dict): sizes: Counter[str] = Counter() if not value: sizes[classify(path, value)] = 2 return sizes items = list(value.items()) for index, (key, child) in enumerate(items): child_path = path + (str(key),) child_sizes = account(child, child_path) sizes.update(child_sizes) overhead = len(dumps(str(key))) + 1 if index: overhead += 1 sizes[_dominant(child_sizes)] += overhead sizes[_dominant(sizes)] += 2 return sizes if isinstance(value, list): sizes = Counter() if not value: sizes[classify(path, value)] = 2 return sizes for index, child in enumerate(value): child_sizes = account(child, path + (str(index),)) sizes.update(child_sizes) if index: sizes[_dominant(child_sizes)] += 1 sizes[_dominant(sizes)] += 2 return sizes return Counter({classify(path, value): len(dumps(value))}) def category_report(value: Any) -> dict[str, Any]: sizes = account(value) total = len(dumps(value)) covered = sum(sizes[name] for name in CATEGORIES) return { "categories": {name: int(sizes.get(name, 0)) for name in CATEGORIES}, "total_chars": total, "category_sum": covered, "sum_error_ratio": 0.0 if total == 0 else abs(covered - total) / total, "tokens": estimate_tokens(dumps(value)), } def duplicate_report(model_context: dict[str, Any]) -> dict[str, Any]: consultations = model_context.get("consultations") first = consultations[0] if isinstance(consultations, list) and consultations else None spread_keys = [ key for key in (first or {}) if key in model_context and key != "consultations" ] matching = [ key for key in spread_keys if dumps(model_context.get(key)) == dumps(first.get(key)) ] named_copies: dict[str, int] = {} def walk(node: Any, path: tuple[str, ...] = ()) -> None: if isinstance(node, dict): for key, child in node.items(): leaf = str(key) if leaf in {"varga_spectrum", "western_spectrum", "technique_audit_table", "methodology"}: named_copies[leaf] = named_copies.get(leaf, 0) + 1 walk(child, path + (leaf,)) elif isinstance(node, list): for index, child in enumerate(node): walk(child, path + (str(index),)) walk(model_context) consultation_chars = len(dumps(consultations)) if consultations is not None else 0 return { "consultation_copies": 1 + (1 if isinstance(first, dict) else 0), "spread_keys_matching_consultations_0": matching, "consultations_chars": consultation_chars, "named_subtree_copies": named_copies, } def load_public_charts(root: Path) -> list[dict[str, Any]]: charts = [] for spec in CHART_SOURCES: payload = json.loads((root / spec["path"]).read_text(encoding="utf-8")) case = next(item for item in payload["cases"] if item["case_id"] == spec["case_id"]) birth = case["birth"] year, month, day = (int(part) for part in birth["date"].split("-")) hour, minute = (int(part) for part in birth["time"].split(":")[:2]) charts.append({ "id": spec["id"], "label": spec["label"], "source": spec["path"], "case_id": spec["case_id"], "rodden_rating": (birth.get("source") or {}).get("rodden_rating"), "place": birth.get("place"), "body": { "year": year, "month": month, "day": day, "hour": hour, "minute": minute, "second": 0, "lat": birth["latitude"], "lon": birth["longitude"], "tz": birth["timezone_offset"], "city": birth.get("place") or spec["label"], "ayanamsa": AYANAMSA, "node_mode": NODE_MODE, "today": REFERENCE_DATE, "entry_mode": "direct_chart", "defer_optional_external_evidence": True, "declared_accuracy": "minute", "birth_time_accuracy": "confirmed", }, }) return charts def _dig(value: Any, *keys: str) -> Any: current = value for key in keys: if not isinstance(current, dict) or key not in current: return None current = current[key] return current def claim_evidence(model_context: dict[str, Any], category: str) -> dict[str, Any]: cards = model_context.get("claim_cards") if not isinstance(cards, list): return {} for card in cards: if isinstance(card, dict) and card.get("category") == category and isinstance(card.get("evidence"), dict): return card["evidence"] return {} def _house_entry(houses: Any, number: int) -> Any: if isinstance(houses, dict): for key in (str(number), f"house_{number}", f"h{number}"): if key in houses: return houses[key] for value in houses.values(): if isinstance(value, dict) and value.get("number") == number: return value if isinstance(houses, list): for value in houses: if isinstance(value, dict) and value.get("number") == number: return value if 1 <= number <= len(houses): return houses[number - 1] return None def _pick(mapping: Any, names: list[str]) -> dict[str, Any]: if not isinstance(mapping, dict): return {} lowered = {str(key).lower(): key for key in mapping} picked = {} for name in names: key = lowered.get(name.lower()) if key is not None: picked[str(key)] = mapping[key] return picked def cut_evidence_card(model_context: dict[str, Any], question_id: str) -> dict[str, Any]: """Copy card fields from the projected model context. Missing paths stay gaps.""" spec = CARD_SPECS[question_id] natal = claim_evidence(model_context, "natal_foundation") timing = claim_evidence(model_context, "timing") spectrum = natal.get("varga_spectrum") if isinstance(natal.get("varga_spectrum"), dict) else {} formal = spectrum.get("formal") if isinstance(spectrum.get("formal"), dict) else {} houses = natal.get("houses") planets = natal.get("planets") if isinstance(natal.get("planets"), dict) else {} functional = natal.get("functional_benefic_malefic") dasha = timing.get("dasha_sub_periods") if isinstance(timing.get("dasha_sub_periods"), dict) else timing.get("dasha") narayana = timing.get("narayana_dasha") gaps = [] vargas = {} for code in spec["vargas"]: chart = formal.get(code) if chart is None: gaps.append(f"varga:{code}") else: vargas[code] = chart house_rows = {} for number in spec["houses"]: row = _house_entry(houses, number) if row is None: gaps.append(f"house:{number}") else: house_rows[str(number)] = row picked_planets = _pick(planets, spec["planets"]) for name in spec["planets"]: if name.lower() not in {key.lower() for key in picked_planets}: gaps.append(f"planet:{name}") card = { "card_id": question_id, "label": spec["label"], "question": model_context.get("question"), "route": model_context.get("route"), "base": { "ascendant": natal.get("ascendant"), "houses": houses, "planet_placements": _planet_placements(planets), "functional_benefic_malefic": functional, "vimshottari": dasha, "narayana_dasha": narayana, }, "domain": { "vargas": vargas, "houses": house_rows, "planets": picked_planets, "arudha": _pick(natal.get("arudha_padas"), spec["arudha"]), "karakas": _pick(_dig(natal, "jaimini", "chara_karakas") or natal.get("chara_karakas"), spec["karakas"]), "yogas": natal.get("yogas") if spec["yogas"] else None, "transits": timing.get("transits") if spec["transits"] else None, "ashtakavarga": natal.get("ashtakavarga") if spec["ashtakavarga"] else None, "shadbala": natal.get("shadbala") if spec["shadbala"] else None, }, "gaps": gaps, "basis": spec.get("basis", []), } if card["base"]["ascendant"] is None: gaps.append("ascendant") if functional is None: gaps.append("functional_benefic_malefic") if dasha is None: gaps.append("vimshottari") if narayana is None: gaps.append("narayana_dasha") text = dumps(card) return {"card": card, "text": text, "size": estimate_tokens(text), "gaps": gaps} def _ci_get(mapping: Any, name: str) -> Any: if not isinstance(mapping, dict): return None for key, value in mapping.items(): if str(key).lower() == name.lower(): return value return None def _planet_placements(planets: Any) -> dict[str, Any]: if not isinstance(planets, dict): return {} placed = {} for name, row in planets.items(): if isinstance(row, dict): placed[str(name)] = { key: row[key] for key in ("sign", "house") if key in row } elif isinstance(row, str): placed[str(name)] = {"sign": row} return placed def engine_facts(workflow: dict[str, Any]) -> dict[str, Any]: chart = workflow.get("chart") if isinstance(workflow.get("chart"), dict) else {} modules = chart.get("modules") if isinstance(chart.get("modules"), dict) else {} ascendant = chart.get("ascendant") if isinstance(chart.get("ascendant"), dict) else {} planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {} moon = _ci_get(planets, "moon") moon = moon if isinstance(moon, dict) else {} sub = modules.get("dasha_sub_periods") if isinstance(modules.get("dasha_sub_periods"), dict) else {} current = sub.get("current") if isinstance(sub.get("current"), dict) else {} mahadasha = current.get("mahadasha") if isinstance(current.get("mahadasha"), dict) else {} antardasha = current.get("antardasha") if isinstance(current.get("antardasha"), dict) else {} pratyantar = sub.get("pratyantar_dasha_timeline") if isinstance(sub.get("pratyantar_dasha_timeline"), dict) else {} pratyantar_current = pratyantar.get("current") if isinstance(pratyantar.get("current"), dict) else {} narayana = modules.get("narayana_dasha") if isinstance(modules.get("narayana_dasha"), dict) else {} narayana_current = narayana.get("current_dasha") if isinstance(narayana.get("current_dasha"), dict) else {} narayana_md = narayana_current.get("md") if isinstance(narayana_current.get("md"), dict) else {} spectrum = modules.get("varga_spectrum") if isinstance(modules.get("varga_spectrum"), dict) else {} formal = spectrum.get("formal") if isinstance(spectrum.get("formal"), dict) else {} d12 = formal.get("D12") if isinstance(formal.get("D12"), dict) else {} return { "ascendant_sign": ascendant.get("sign"), "moon_sign": moon.get("sign"), "moon_house": moon.get("house"), "vimshottari_lord": mahadasha.get("lord"), "vimshottari_start": mahadasha.get("start"), "vimshottari_end": mahadasha.get("end"), "antardasha_start": antardasha.get("start"), "antardasha_end": antardasha.get("end"), "pratyantar_start": pratyantar_current.get("start"), "pratyantar_end": pratyantar_current.get("end"), "narayana_sign": narayana_md.get("sign"), "d12_lagna": d12.get("lagna") if isinstance(d12.get("lagna"), str) else None, "d12_moon": _ci_get(d12.get("planets"), "moon") if isinstance(d12.get("planets"), dict) else None, } def card_facts(card: dict[str, Any]) -> dict[str, Any]: base = card.get("base") if isinstance(card.get("base"), dict) else {} domain = card.get("domain") if isinstance(card.get("domain"), dict) else {} ascendant = base.get("ascendant") if isinstance(base.get("ascendant"), dict) else {} moon = _ci_get(base.get("planet_placements"), "moon") if not isinstance(moon, dict): moon = _ci_get(domain.get("planets"), "moon") moon = moon if isinstance(moon, dict) else {} sub = base.get("vimshottari") if isinstance(base.get("vimshottari"), dict) else {} current = sub.get("current") if isinstance(sub.get("current"), dict) else {} mahadasha = current.get("mahadasha") if isinstance(current.get("mahadasha"), dict) else {} antardasha = current.get("antardasha") if isinstance(current.get("antardasha"), dict) else {} pratyantar = sub.get("pratyantar_dasha_timeline") if isinstance(sub.get("pratyantar_dasha_timeline"), dict) else {} pratyantar_current = pratyantar.get("current") if isinstance(pratyantar.get("current"), dict) else {} narayana = base.get("narayana_dasha") if isinstance(base.get("narayana_dasha"), dict) else {} narayana_current = narayana.get("current_dasha") if isinstance(narayana.get("current_dasha"), dict) else {} narayana_md = narayana_current.get("md") if isinstance(narayana_current.get("md"), dict) else {} d12 = (domain.get("vargas") or {}).get("D12") if isinstance(domain.get("vargas"), dict) else None d12 = d12 if isinstance(d12, dict) else {} return { "ascendant_sign": ascendant.get("sign"), "moon_sign": moon.get("sign"), "moon_house": moon.get("house"), "vimshottari_lord": mahadasha.get("lord"), "vimshottari_start": mahadasha.get("start"), "vimshottari_end": mahadasha.get("end"), "antardasha_start": antardasha.get("start"), "antardasha_end": antardasha.get("end"), "pratyantar_start": pratyantar_current.get("start"), "pratyantar_end": pratyantar_current.get("end"), "narayana_sign": narayana_md.get("sign"), "d12_lagna": d12.get("lagna"), "d12_moon": _ci_get(d12.get("planets"), "moon"), } def verbatim_check(card: dict[str, Any], facts: dict[str, Any]) -> dict[str, Any]: actual = card_facts(card) has_d12 = isinstance((card.get("domain") or {}).get("vargas"), dict) and "D12" in card["domain"]["vargas"] checks = {} for key, expected in facts.items(): if key.startswith("d12_") and not has_d12: checks[key] = {"status": "not_on_card", "expected": expected, "actual": None} continue got = actual.get(key) if expected is None or expected == "": checks[key] = {"status": "engine_missing", "expected": None, "actual": got} elif got == expected: checks[key] = {"status": "match", "expected": expected, "actual": got} else: checks[key] = {"status": "mismatch", "expected": expected, "actual": got} mismatched = [key for key, row in checks.items() if row["status"] == "mismatch"] return {"ok": not mismatched, "checks": checks, "mismatched": mismatched} def parse_registry_layers(source: str) -> dict[str, list[str]]: found = {} for match in re.finditer(r'id: "(\w+)".*?requiredLayers: \[([^\]]*)\]', source): layers = re.findall(r'"([^"]+)"', match.group(2)) found[match.group(1)] = layers return found def parse_methodology_routes(source: str) -> dict[str, str | None]: found = {} for match in re.finditer(r'(\w+): \{ strictRoute: (?:"([^"]+)"|null)', source): found[match.group(1)] = match.group(2) return found def parse_must_use_layers(source: str) -> dict[str, list[str]]: block = re.search(r"DOMAIN_MUST_USE_LAYERS[\s\S]*?};", source) if not block: return {} found = {} for match in re.finditer(r'(\w+): \[([^\]]*)\]', block.group(0)): found[match.group(1)] = re.findall(r'"([^"]+)"', match.group(2)) return found def source_matrix(root: Path) -> dict[str, Any]: registry = parse_registry_layers((root / "frontend/src/lib/consultation-domain-registry.ts").read_text(encoding="utf-8")) methodology = parse_methodology_routes((root / "frontend/src/lib/consultation-methodology.ts").read_text(encoding="utf-8")) must_use = parse_must_use_layers((root / "frontend/src/mastra/consultation-workflow.ts").read_text(encoding="utf-8")) from unified_consultation_orchestrator import UnifiedConsultationOrchestrator python_routes = { name: list(route.focus_techniques) for name, route in UnifiedConsultationOrchestrator._ROUTE_DEFINITIONS.items() } domains = sorted(set(registry) | set(methodology) | set(python_routes) | set(CARD_SPECS)) rows = [] for domain in domains: card_id = domain if domain in CARD_SPECS else None rows.append({ "domain": domain, "registry_required_layers": registry.get(domain, []), "methodology_strict_route": methodology.get(domain), "ts_must_use_layers": must_use.get(domain, []), "python_focus_techniques": python_routes.get(domain, []), "card_vargas": CARD_SPECS.get(card_id, {}).get("vargas", []) if card_id else [], "parents_children_split": domain == "family", }) return {"domains": rows, "skill_citations": { "d12_family": "references/strict-workflow-router.md: D12 for family/ancestral themes", "d7_children": "references/strict-workflow-router.md: D7 for children", "decision_tree_parents": "references/birth-time-rectification-decision-tree.md: 父母/家族 -> D12", "decision_tree_children": "references/birth-time-rectification-decision-tree.md: 子女/生育 -> D7", }} def part_b_coverage(card: dict[str, Any], question_id: str) -> dict[str, Any]: base = card.get("base") if isinstance(card.get("base"), dict) else {} domain = card.get("domain") if isinstance(card.get("domain"), dict) else {} spec = CARD_SPECS[question_id] present = { "functional_benefic_malefic": base.get("functional_benefic_malefic") is not None, "vimshottari": base.get("vimshottari") is not None, "narayana": base.get("narayana_dasha") is not None, "ascendant_raw": base.get("ascendant") is not None, "domain_vargas": sorted(domain.get("vargas") or {}), "expected_vargas": spec["vargas"], } missing_vargas = [code for code in spec["vargas"] if code not in (domain.get("vargas") or {})] return { "dual_dasha": present["vimshottari"] and present["narayana"], "domain_divisional": not missing_vargas, "functional_benefic_malefic": present["functional_benefic_malefic"], "raw_data": present["ascendant_raw"] and present["vimshottari"], "mevg": "kept_backstage", "real_case_calibration": "kept_backstage", "missing_vargas": missing_vargas, "present": present, } TELEMETRY_SPEC = { "stores": "numbers, enums, and field names only", "does_not_store": [ "question text", "answer text", "birth date", "birth time", "birth place", "name", "email", "user id", "session id", "matched substring", ], "fields": [ {"name": "domain", "type": "enum", "values": [item["id"] for item in QUESTIONS]}, {"name": "card_version", "type": "enum", "values": ["evidence-card-draft-20260927"]}, {"name": "card_chars", "type": "int"}, {"name": "card_token_estimate", "type": "int"}, {"name": "cited_field_ids", "type": "enum[]", "note": "ids from the card schema, not free text"}, {"name": "feedback", "type": "enum", "values": ["up", "down", "none"]}, ], "citation_method": { "rule": "A field is cited when its scalar value appears in the answer and the value is an ISO date, a degree token, or a compound planet-in-sign phrase of at least 8 characters. Bare planet or sign names are not counted.", "false_positive": "A date or degree that also appears in the question, or a repeated stock phrase, can be counted without the model using that field.", "false_negative": "A paraphrase that names the period without copying the date is not counted.", "stored": "Only the field id. The matched text is discarded.", }, "aggregates": [ "count of answers by domain and card_version", "fields whose cited rate stays under 5 percent after 30 answers in that domain", "down rate by domain and card_version", ], } def card_dependencies(question_id: str) -> list[str]: spec = CARD_SPECS[question_id] modules = ["compute_chart", "functional_benefic_malefic", "dasha_sub_periods", "narayana_dasha"] if spec["vargas"]: modules.append("varga_requested_only") if spec["arudha"] or spec["karakas"]: modules.append("arudha_padas") if spec["karakas"]: modules.append("chara_karakas") if spec["transits"]: modules.append("transits") if spec["ashtakavarga"]: modules.append("ashtakavarga") if spec["shadbala"]: modules.append("shadbala") if spec["yogas"]: modules.append("yogas") return modules