362d903c31
The second review still failed on missing database and route coverage, an unmarked relationship structure, a stuck English lease, and a sidebar that fetched the product switch itself. Co-Authored-By: Grok (agent loop) Co-authored-by: Cursor <cursoragent@cursor.com>
341 lines
14 KiB
Python
341 lines
14 KiB
Python
"""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]
|
|
ZH_SECTION_MINIMUM = 120
|
|
ZH_TOTAL_MINIMUM = 1800
|
|
EN_SECTION_MINIMUM = 80
|
|
EN_TOTAL_MINIMUM = 1100
|
|
STATUS_ONLY_VALUES = {
|
|
"ready_for_comparison",
|
|
"executed",
|
|
"present",
|
|
"used",
|
|
"available",
|
|
"passed",
|
|
"parameter_sensitive",
|
|
"partial",
|
|
"weak",
|
|
"observation_only",
|
|
"not_applicable",
|
|
"not_used",
|
|
"not_used_for_date_intersection",
|
|
"blocked",
|
|
"missing",
|
|
"queued",
|
|
"running",
|
|
"ready",
|
|
"failed",
|
|
}
|
|
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",
|
|
"kuja", "nadi", "lagna", "ul", "dk",
|
|
)
|
|
ZH_NAKSHATRAS = (
|
|
"娄宿", "胃宿", "昴宿", "毕宿", "觜宿", "参宿", "井宿", "鬼宿", "柳宿",
|
|
"星宿", "张宿", "翼宿", "轸宿", "角宿", "亢宿", "氐宿", "房宿", "心宿",
|
|
"尾宿", "箕宿", "斗宿", "女宿", "虚宿", "危宿", "室宿", "壁宿", "奎宿",
|
|
)
|
|
EN_NAKSHATRAS = (
|
|
"ashwini", "bharani", "krittika", "rohini", "mrigashira", "ardra", "punarvasu",
|
|
"pushya", "ashlesha", "magha", "phalguni", "hasta", "chitra", "swati", "vishakha",
|
|
"anuradha", "jyeshtha", "mula", "ashadha", "shravana", "dhanishta", "shatabhisha",
|
|
"bhadrapada", "revati",
|
|
)
|
|
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"(?<![A-Za-z])D\d{1,2}(?![A-Za-z])", 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(term for term in ZH_SIGNS if term in basis)
|
|
tokens.extend(term for term in ZH_NAKSHATRAS 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, *EN_SIGNS, *EN_NAKSHATRAS) 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 _string_values(value: Any) -> set[str]:
|
|
"""Complete string field values. A one-character slice of a longer field does not count."""
|
|
found: set[str] = set()
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
if stripped and stripped.lower() not in STATUS_ONLY_VALUES:
|
|
found.add(stripped)
|
|
elif isinstance(value, dict):
|
|
for item in value.values():
|
|
found.update(_string_values(item))
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
found.update(_string_values(item))
|
|
return found
|
|
|
|
|
|
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 = ZH_SECTION_MINIMUM if language == "zh" else EN_SECTION_MINIMUM
|
|
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_values = _string_values(inputs.get(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_values:
|
|
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 "")
|
|
phase = str(segment.get("phase") or "")
|
|
failures.extend(_scan_prose(description, language, "timeline", opening=False))
|
|
failures.extend(_scan_prose(phase, 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_values = _string_values(inputs.get("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_values:
|
|
failures.append(_fail("practices", "evidence_missing", evidence or "item"))
|
|
|
|
total_minimum = ZH_TOTAL_MINIMUM if language == "zh" else EN_TOTAL_MINIMUM
|
|
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()
|