feat(synastry): write the full relationship report on demand
The pair page and reader replace the leftover chat draft, and the same pair is not charged again. Co-Authored-By: Grok (agent loop) Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jesse-ux
co-authored by
Grok (agent loop)
Cursor
Grok (agent loop)
Cursor
parent
2130682d43
commit
9ea5b3b0c0
@@ -10520,6 +10520,13 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, JyotishComputeService):
|
||||
except match_api.MarriageMatchInputError as exc:
|
||||
raise BadRequest(str(exc)) from exc
|
||||
self._json(result)
|
||||
elif path == '/api/synastry_full_reader_gate':
|
||||
match_api = _load_local_module('marriage_match_service.api')
|
||||
try:
|
||||
result = match_api.check_synastry_reader(body)
|
||||
except match_api.MarriageMatchInputError as exc:
|
||||
raise BadRequest(str(exc)) from exc
|
||||
self._json(result)
|
||||
elif path == '/api/dasha':
|
||||
result = self._compute_dasha_system(body)
|
||||
self._json(result)
|
||||
|
||||
@@ -78,6 +78,22 @@ def _commit_sha() -> str:
|
||||
return os.environ.get("GITHUB_SHA") or ""
|
||||
|
||||
|
||||
def check_synastry_reader(body: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Relationship-language gate. The body is the delivery plus the section inputs."""
|
||||
if not isinstance(body, dict):
|
||||
raise MarriageMatchInputError("request body must be an object")
|
||||
from marriage_match_service.synastry_full_reader_gate import validate_synastry_reader
|
||||
|
||||
delivery = body.get("delivery")
|
||||
section_inputs = body.get("sectionInputs")
|
||||
language = body.get("language") or "zh"
|
||||
if not isinstance(delivery, dict) or not isinstance(language, str):
|
||||
raise MarriageMatchInputError("delivery and language are required")
|
||||
if section_inputs is not None and not isinstance(section_inputs, dict):
|
||||
raise MarriageMatchInputError("sectionInputs must be an object")
|
||||
return validate_synastry_reader(delivery, section_inputs or {}, language)
|
||||
|
||||
|
||||
def compute_synastry_full_dossier(handler: Any, body: dict[str, Any]) -> dict[str, Any]:
|
||||
if not isinstance(body, dict):
|
||||
raise MarriageMatchInputError("request body must be an object")
|
||||
|
||||
@@ -9,6 +9,7 @@ from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
@@ -41,6 +42,60 @@ def _dossier_builder():
|
||||
return module.build_full_synastry_dossier
|
||||
|
||||
|
||||
def _parse_day(value: Any) -> datetime | None:
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return None
|
||||
text = value.strip()[:10]
|
||||
try:
|
||||
return datetime.strptime(text, "%Y-%m-%d")
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _clip_period(period: dict[str, Any], level: str, window_start: datetime, window_end: datetime) -> dict[str, Any] | None:
|
||||
start = _parse_day(period.get("start") or period.get("start_datetime"))
|
||||
end = _parse_day(period.get("end") or period.get("end_datetime"))
|
||||
if start is None or end is None or end <= window_start or start >= window_end:
|
||||
return None
|
||||
lord = period.get("lord")
|
||||
if not isinstance(lord, str) or not lord:
|
||||
return None
|
||||
lord_cn = period.get("lord_cn") if isinstance(period.get("lord_cn"), str) else lord
|
||||
return {
|
||||
"level": level,
|
||||
"lord": lord,
|
||||
"lordCn": lord_cn,
|
||||
"start": max(start, window_start).strftime("%Y-%m-%d"),
|
||||
"end": min(end, window_end).strftime("%Y-%m-%d"),
|
||||
}
|
||||
|
||||
|
||||
def upcoming_vimshottari(reading: dict[str, Any], today: str, years: int = 5) -> list[dict[str, Any]]:
|
||||
"""Mahadasha and antardasha slices that overlap the next few years."""
|
||||
packet = reading.get("raw_full_reading") if isinstance(reading.get("raw_full_reading"), dict) else reading
|
||||
modules = packet.get("modules") if isinstance(packet, dict) else {}
|
||||
dasha = modules.get("dasha") if isinstance(modules, dict) else {}
|
||||
timeline = dasha.get("timeline") if isinstance(dasha, dict) else []
|
||||
start = _parse_day(today) or datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
end = start + timedelta(days=365 * years)
|
||||
rows: list[dict[str, Any]] = []
|
||||
if not isinstance(timeline, list):
|
||||
return rows
|
||||
for mahadasha in timeline:
|
||||
if not isinstance(mahadasha, dict):
|
||||
continue
|
||||
clipped = _clip_period(mahadasha, "md", start, end)
|
||||
if clipped:
|
||||
rows.append(clipped)
|
||||
for antardasha in mahadasha.get("antardasha_timeline") or []:
|
||||
if not isinstance(antardasha, dict):
|
||||
continue
|
||||
child = _clip_period(antardasha, "ad", start, end)
|
||||
if child:
|
||||
rows.append(child)
|
||||
return rows
|
||||
|
||||
|
||||
def build_synastry_full_dossier(
|
||||
handler: Any,
|
||||
person_a: dict[str, Any],
|
||||
@@ -56,6 +111,7 @@ def build_synastry_full_dossier(
|
||||
reading_a = future_a.result()
|
||||
reading_b = future_b.result()
|
||||
dossier = _dossier_builder()(reading_a, reading_b)
|
||||
today = str(person_a.get("today") or person_b.get("today") or "")
|
||||
snapshots = dossier.get("chart_snapshots") if isinstance(dossier, dict) else {}
|
||||
snapshots = snapshots if isinstance(snapshots, dict) else {}
|
||||
raman = build_marriage_match_dossier(
|
||||
@@ -67,4 +123,8 @@ def build_synastry_full_dossier(
|
||||
"rules_version": RULES_VERSION,
|
||||
"dossier": dossier,
|
||||
"raman_marriage_match": raman,
|
||||
"timing_periods": {
|
||||
"person_a": upcoming_vimshottari(reading_a if isinstance(reading_a, dict) else {}, today),
|
||||
"person_b": upcoming_vimshottari(reading_b if isinstance(reading_b, dict) else {}, today),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -0,0 +1,290 @@
|
||||
"""Relationship-language gate for the full synastry report.
|
||||
|
||||
The upstream bilingual validator stays untouched. This gate only borrows its
|
||||
private-evidence and supplementary-method lists. It judges the reader text
|
||||
against the relationship-language guide, and it checks that each "盘面" line
|
||||
matches the evidence it cites.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
SKILL_SCRIPTS = ROOT / "skills" / "jyotish-full-synastry-reading" / "scripts"
|
||||
if str(SKILL_SCRIPTS) not in sys.path:
|
||||
sys.path.insert(0, str(SKILL_SCRIPTS))
|
||||
|
||||
from validate_bilingual_synastry_delivery import ( # noqa: E402
|
||||
PRIVATE_EVIDENCE_TERMS,
|
||||
SUPPLEMENTARY_METHOD_TERMS,
|
||||
)
|
||||
|
||||
SECTION_ORDER = (
|
||||
"climate",
|
||||
"emotion",
|
||||
"communication",
|
||||
"safety",
|
||||
"conflict",
|
||||
"daily",
|
||||
"direction",
|
||||
"timeline",
|
||||
"practices",
|
||||
)
|
||||
PROSE_SECTIONS = SECTION_ORDER[:7]
|
||||
EVIDENCE_LABELS = (
|
||||
"multi_system_consensus",
|
||||
"single_system_inference",
|
||||
"parameter_sensitive",
|
||||
"unclosed_divisional_chart",
|
||||
"user_history_verification_required",
|
||||
"blocked",
|
||||
)
|
||||
|
||||
ZH_PLANETS = ("太阳", "月亮", "水星", "金星", "火星", "木星", "土星", "罗睺", "计都")
|
||||
ZH_SIGNS = (
|
||||
"白羊", "金牛", "双子", "巨蟹", "狮子", "处女",
|
||||
"天秤", "天蝎", "射手", "摩羯", "水瓶", "双鱼",
|
||||
)
|
||||
ZH_TERMS = (
|
||||
*ZH_PLANETS,
|
||||
*ZH_SIGNS,
|
||||
"宫", "星宿", "上升", "分盘", "九分盘", "婚姻分盘",
|
||||
"大运", "小运", "行运", "Dasha", "Koota", "Kuja", "火星冲克",
|
||||
"Nadi", "Yoni", "Varna", "Ashtakoota", "Papasamya", "UL", "DK", "满分",
|
||||
)
|
||||
ZH_JUDGEMENTS = (
|
||||
"克", "煞", "凶", "吉", "注定", "必然", "命中", "一定会",
|
||||
"合不合", "不合", "相合", "不适合", "适合结婚", "不宜",
|
||||
"灵魂伴侣", "宇宙安排", "百分百", "会不会在一起",
|
||||
)
|
||||
EN_PLANETS = ("sun", "moon", "mercury", "venus", "mars", "jupiter", "saturn", "rahu", "ketu")
|
||||
EN_SIGNS = (
|
||||
"aries", "taurus", "gemini", "cancer", "leo", "virgo",
|
||||
"libra", "scorpio", "sagittarius", "capricorn", "aquarius", "pisces",
|
||||
)
|
||||
EN_TERMS = (
|
||||
*EN_PLANETS,
|
||||
*EN_SIGNS,
|
||||
"house", "nakshatra", "ascendant", "navamsa", "dasha", "transit", "koota",
|
||||
)
|
||||
EN_JUDGEMENTS = (
|
||||
"soulmate", "destined", "inevitable", "guaranteed", "the universe",
|
||||
"not suitable", "will marry", "will divorce",
|
||||
)
|
||||
ZH_OPENING = ("这段关系的核心不是", "不是有没有")
|
||||
EN_OPENING = ("the core of this relationship is not", "it is not whether")
|
||||
ZH_EVENT = re.compile(r"(19|20)\d{2}\s*年.{0,12}(结婚|领证|分手|离婚)")
|
||||
EN_EVENT = re.compile(r"\b(19|20)\d{2}\b.{0,24}\b(marry|married|marriage|divorce|break up)\b", re.I)
|
||||
ZH_SCORE = re.compile(r"\d+\s*分")
|
||||
EN_SCORE = re.compile(r"\b\d+\s*(points|point|score)\b", re.I)
|
||||
ZH_HOUSE = re.compile(r"(?:第)?[一二三四五六七八九十\d]{1,3}宫")
|
||||
EN_HOUSE = re.compile(r"\b(?:house\s*)?\d{1,2}(?:st|nd|rd|th)?\s+house\b", re.I)
|
||||
D_CHART = re.compile(r"\bD\d{1,2}\b", re.I)
|
||||
|
||||
|
||||
def _fail(section: str, code: str, detail: str) -> dict[str, str]:
|
||||
return {"section": section, "code": code, "detail": detail}
|
||||
|
||||
|
||||
def _contains_any(text: str, terms: tuple[str, ...]) -> str | None:
|
||||
for term in terms:
|
||||
if term and term in text:
|
||||
return term
|
||||
return None
|
||||
|
||||
|
||||
def _contains_word(text: str, words: tuple[str, ...]) -> str | None:
|
||||
lowered = text.lower()
|
||||
for word in words:
|
||||
if re.search(rf"\b{re.escape(word.lower())}\b", lowered):
|
||||
return word
|
||||
return None
|
||||
|
||||
|
||||
def _first_sentence(text: str) -> str:
|
||||
match = re.split(r"[。！？.!?]", text.strip(), maxsplit=1)
|
||||
return match[0] if match else text.strip()
|
||||
|
||||
|
||||
def _basis_tokens(basis: str, language: str) -> list[str]:
|
||||
tokens: list[str] = []
|
||||
if language == "zh":
|
||||
tokens.extend(term for term in ZH_PLANETS if term in basis)
|
||||
tokens.extend(match.group(0) for match in ZH_HOUSE.finditer(basis))
|
||||
if "大运" in basis:
|
||||
tokens.append("大运")
|
||||
if "小运" in basis:
|
||||
tokens.append("小运")
|
||||
else:
|
||||
tokens.extend(word for word in EN_PLANETS if re.search(rf"\b{word}\b", basis, re.I))
|
||||
tokens.extend(match.group(0) for match in EN_HOUSE.finditer(basis))
|
||||
if re.search(r"\bdasha\b", basis, re.I):
|
||||
tokens.append("dasha")
|
||||
tokens.extend(match.group(0) for match in D_CHART.finditer(basis))
|
||||
return tokens
|
||||
|
||||
|
||||
def _input_text(section_inputs: dict[str, Any], section_id: str) -> str:
|
||||
value = section_inputs.get(section_id, "")
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
return json.dumps(value, ensure_ascii=False, sort_keys=True)
|
||||
|
||||
|
||||
def _scan_prose(text: str, language: str, section: str, *, opening: bool) -> list[dict[str, str]]:
|
||||
failures: list[dict[str, str]] = []
|
||||
private = _contains_any(text, PRIVATE_EVIDENCE_TERMS)
|
||||
supplementary = _contains_any(text, SUPPLEMENTARY_METHOD_TERMS)
|
||||
if private:
|
||||
failures.append(_fail(section, "private_term", private))
|
||||
if supplementary:
|
||||
failures.append(_fail(section, "supplementary_method", supplementary))
|
||||
if language == "zh":
|
||||
term = _contains_any(text, ZH_TERMS)
|
||||
judgement = _contains_any(text, ZH_JUDGEMENTS)
|
||||
if term:
|
||||
failures.append(_fail(section, "banned_term", term))
|
||||
if judgement:
|
||||
failures.append(_fail(section, "banned_judgement", judgement))
|
||||
if ZH_SCORE.search(text):
|
||||
failures.append(_fail(section, "score", "数字分"))
|
||||
if ZH_EVENT.search(text):
|
||||
failures.append(_fail(section, "dated_event", "年份事件"))
|
||||
if D_CHART.search(text):
|
||||
failures.append(_fail(section, "banned_term", "D盘"))
|
||||
if opening and _contains_any(_first_sentence(text), ZH_OPENING):
|
||||
failures.append(_fail(section, "analytic_opening", _first_sentence(text)))
|
||||
else:
|
||||
term = _contains_word(text, EN_TERMS)
|
||||
judgement = _contains_word(text, EN_JUDGEMENTS)
|
||||
if term:
|
||||
failures.append(_fail(section, "banned_term", term or ""))
|
||||
if judgement:
|
||||
failures.append(_fail(section, "banned_judgement", judgement or ""))
|
||||
if EN_SCORE.search(text):
|
||||
failures.append(_fail(section, "score", "score"))
|
||||
if EN_EVENT.search(text):
|
||||
failures.append(_fail(section, "dated_event", "dated event"))
|
||||
if D_CHART.search(text):
|
||||
failures.append(_fail(section, "banned_term", "D chart"))
|
||||
if opening and _contains_word(_first_sentence(text), EN_OPENING):
|
||||
failures.append(_fail(section, "analytic_opening", _first_sentence(text)))
|
||||
return failures
|
||||
|
||||
|
||||
def validate_synastry_reader(
|
||||
delivery: dict[str, Any],
|
||||
section_inputs: dict[str, Any] | None = None,
|
||||
language: str = "zh",
|
||||
) -> dict[str, Any]:
|
||||
"""Return ``{ok, failures}``. ``section_inputs`` is what the model was shown."""
|
||||
if language not in {"zh", "en"}:
|
||||
return {"ok": False, "failures": [_fail("delivery", "language", language)]}
|
||||
inputs = section_inputs or {}
|
||||
failures: list[dict[str, str]] = []
|
||||
sections = delivery.get("sections")
|
||||
if not isinstance(sections, list):
|
||||
return {"ok": False, "failures": [_fail("delivery", "sections", "missing")]}
|
||||
by_id = {
|
||||
str(row.get("id")): row
|
||||
for row in sections
|
||||
if isinstance(row, dict) and row.get("id")
|
||||
}
|
||||
if [row for row in SECTION_ORDER if row not in by_id] or [str(row.get("id")) for row in sections] != list(SECTION_ORDER):
|
||||
failures.append(_fail("delivery", "section_order", "九节顺序不对"))
|
||||
|
||||
prose_characters = 0
|
||||
prose_words = 0
|
||||
for section_id in PROSE_SECTIONS:
|
||||
row = by_id.get(section_id)
|
||||
if not isinstance(row, dict):
|
||||
failures.append(_fail(section_id, "missing_section", section_id))
|
||||
continue
|
||||
text = str(row.get("text") or "")
|
||||
prose_characters += len(re.sub(r"\s+", "", text))
|
||||
prose_words += len(re.findall(r"\b[\w']+\b", text))
|
||||
minimum = 120 if language == "zh" else 80
|
||||
measured = len(re.sub(r"\s+", "", text)) if language == "zh" else len(re.findall(r"\b[\w']+\b", text))
|
||||
if measured < minimum:
|
||||
failures.append(_fail(section_id, "too_short", str(measured)))
|
||||
failures.extend(_scan_prose(text, language, section_id, opening=section_id == "climate"))
|
||||
reasons = row.get("reasons")
|
||||
if not isinstance(reasons, list) or not 2 <= len(reasons) <= 3:
|
||||
failures.append(_fail(section_id, "reason_count", "需要两到三条分析"))
|
||||
continue
|
||||
source = _input_text(inputs, section_id)
|
||||
for index, reason in enumerate(reasons):
|
||||
if not isinstance(reason, dict):
|
||||
failures.append(_fail(section_id, "reason_shape", str(index)))
|
||||
continue
|
||||
plain = str(reason.get("plain") or "")
|
||||
basis = str(reason.get("basis") or "")
|
||||
evidence = str(reason.get("evidence") or "")
|
||||
label = str(reason.get("label") or "")
|
||||
failures.extend(_scan_prose(plain, language, section_id, opening=False))
|
||||
if not evidence or evidence not in source:
|
||||
failures.append(_fail(section_id, "evidence_missing", evidence or str(index)))
|
||||
if label not in EVIDENCE_LABELS:
|
||||
failures.append(_fail(section_id, "evidence_label", label or str(index)))
|
||||
for token in _basis_tokens(basis, language):
|
||||
if token.lower() not in evidence.lower() and token not in evidence:
|
||||
failures.append(_fail(section_id, "basis_mismatch", token))
|
||||
|
||||
timeline = by_id.get("timeline") if isinstance(by_id.get("timeline"), dict) else {}
|
||||
segments = timeline.get("segments") if isinstance(timeline, dict) else None
|
||||
if not isinstance(segments, list) or not 3 <= len(segments) <= 5:
|
||||
failures.append(_fail("timeline", "segment_count", "需要三到五段"))
|
||||
else:
|
||||
for segment in segments:
|
||||
if not isinstance(segment, dict):
|
||||
failures.append(_fail("timeline", "segment_shape", "bad"))
|
||||
continue
|
||||
description = str(segment.get("text") or "")
|
||||
failures.extend(_scan_prose(description, language, "timeline", opening=False))
|
||||
if not str(segment.get("basis") or "").strip():
|
||||
failures.append(_fail("timeline", "basis_missing", str(segment.get("id") or "")))
|
||||
|
||||
practices = by_id.get("practices") if isinstance(by_id.get("practices"), dict) else {}
|
||||
items = practices.get("items") if isinstance(practices, dict) else None
|
||||
if not isinstance(items, list) or not items:
|
||||
failures.append(_fail("practices", "missing_items", "需要共同练习"))
|
||||
else:
|
||||
practice_source = _input_text(inputs, "practices")
|
||||
for item in items:
|
||||
if not isinstance(item, dict):
|
||||
failures.append(_fail("practices", "item_shape", "bad"))
|
||||
continue
|
||||
text = str(item.get("text") or "")
|
||||
evidence = str(item.get("evidence") or "")
|
||||
failures.extend(_scan_prose(text, language, "practices", opening=False))
|
||||
if not evidence or evidence not in practice_source:
|
||||
failures.append(_fail("practices", "evidence_missing", evidence or "item"))
|
||||
|
||||
total_minimum = 1800 if language == "zh" else 1100
|
||||
total = prose_characters if language == "zh" else prose_words
|
||||
if total < total_minimum:
|
||||
failures.append(_fail("delivery", "too_short", str(total)))
|
||||
return {"ok": not failures, "failures": failures}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
payload = json.load(sys.stdin)
|
||||
if not isinstance(payload, dict):
|
||||
raise SystemExit("payload must be an object")
|
||||
result = validate_synastry_reader(
|
||||
payload.get("delivery") if isinstance(payload.get("delivery"), dict) else {},
|
||||
payload.get("sectionInputs") if isinstance(payload.get("sectionInputs"), dict) else {},
|
||||
str(payload.get("language") or "zh"),
|
||||
)
|
||||
json.dump(result, sys.stdout, ensure_ascii=False)
|
||||
if not result["ok"]:
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user