jesse-ux
d8d03b5a69
Measure the model-visible consultation payload on three public charts, draft per-domain cards, and record the Narayana/pratyantar projection gap as BUG-1054. No runtime behavior change.
812 lines
31 KiB
Python
812 lines
31 KiB
Python
"""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
|