Jesse_Chen
a3196584d2
Surface D9/D10 change minutes, event-dasha match copy, dual-dasha conflict, and a post-adopt consult handoff so users can keep narrowing or start a reading from a representative time. Co-authored-by: Cursor <cursoragent@cursor.com>
413 lines
17 KiB
Python
413 lines
17 KiB
Python
"""Server-owned P0/P1 refinement packet for birth-time rectification.
|
||
|
||
Produces candidate-narrowing structure only. Never grants a unique minute,
|
||
never emits D9/D10 type labels, and never copies raw scores into public copy.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Sequence
|
||
|
||
from scripts.rectification.house_table import PLANET_ZH, SIGN_LORDS, SIGNS, SIGNS_CN
|
||
|
||
NAKSHATRA_SPAN = 40.0 / 3.0
|
||
NAKSHATRA_BOUNDARY_DEGREES = 2.0
|
||
MATCH_LABELS = {
|
||
"strong": "强相关",
|
||
"medium": "有关联",
|
||
"weak": "弱关联",
|
||
"none": "未见对应",
|
||
}
|
||
# Everyday A/B traits only. No Sanskrit names and no D9/D10 personality tables.
|
||
NAKSHATRA_TRAITS: tuple[tuple[str, str], ...] = (
|
||
("起步快、敢先动手", "更愿意把第一步走完再看"),
|
||
("事情来了会立刻表态", "先把感受压一压再开口"),
|
||
("喜欢把节奏拉开、自己掌握步调", "更在意别人是否跟得上"),
|
||
("照顾身边的人会放在前面", "需要先把自己安顿好"),
|
||
("愿意站到台前把话说清楚", "更习惯在旁边把事情理顺"),
|
||
("对细节和次序很敏感", "更看重大方向有没有走偏"),
|
||
("希望两边都能说得过去", "必要时会直接选边"),
|
||
("碰到转折会往深处想", "更想尽快回到能做事的状态"),
|
||
("愿意把视野拉远一点再决定", "更盯着眼前能落地的一步"),
|
||
("愿意为长期结果多熬一阵", "更怕把时间耗在看不见的地方"),
|
||
("想法一多就想换条路试试", "更想把一条路走稳"),
|
||
("情绪来了会先自己消化", "更需要说出来才过得去"),
|
||
("新开始会让人兴奋", "新开始会让人先观察一阵"),
|
||
("承诺一旦出口就很难收回", "承诺前会反复确认自己是不是真想"),
|
||
("变化来时先问值不值得", "变化来时先问自己扛不扛得住"),
|
||
("家里的事会牵动判断", "更想把家里的事和工作分开"),
|
||
("被看见会更有劲", "被看见反而会先退半步"),
|
||
("计划乱了会先整理清单", "计划乱了会先找一个人商量"),
|
||
("两边关系都想维持住", "维持不住时会干脆拉开距离"),
|
||
("压力大时会往内部找原因", "压力大时会先改外部条件"),
|
||
("愿意把决定放到更大的时间尺度", "更相信眼前这一段就够判断"),
|
||
("愿意为结构稳定让步", "稳定如果太闷就会想拆掉重来"),
|
||
("对规则和例外都很敏感", "更想先有一个能用的规则"),
|
||
("说不清的感受会先放着", "说不清就会反复确认"),
|
||
("一有机会就想动手试", "会先把退路看清楚再动"),
|
||
("对人的反应比对事情本身更敏感", "对事情进度比对气氛更敏感"),
|
||
("收尾时会想把未完成的交代清", "收尾时会想尽快开始下一件"),
|
||
)
|
||
|
||
|
||
def _clock(value: str) -> int:
|
||
return int(value[:2]) * 60 + int(value[3:5])
|
||
|
||
|
||
def _feature_time(feature: dict[str, Any]) -> str | None:
|
||
raw = feature.get("time")
|
||
if isinstance(raw, str) and len(raw) >= 5:
|
||
return raw[:5]
|
||
return None
|
||
|
||
|
||
def _features(built: dict[str, Any]) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
for context in built.get("static_contexts") or []:
|
||
if not isinstance(context, dict):
|
||
continue
|
||
feature = context.get("feature")
|
||
if not isinstance(feature, dict):
|
||
continue
|
||
time = _feature_time(feature)
|
||
if time:
|
||
rows.append(feature)
|
||
rows.sort(key=lambda item: _clock(str(_feature_time(item))))
|
||
return rows
|
||
|
||
|
||
def match_level(rule_ids: Sequence[str]) -> str:
|
||
ids = [str(item) for item in rule_ids]
|
||
if not ids or ids == ["no_domain_activation"]:
|
||
return "none"
|
||
if any(
|
||
item.startswith("vim_md_domain") or item.startswith("narayana_md_domain")
|
||
for item in ids
|
||
):
|
||
return "strong"
|
||
if any(
|
||
item.startswith("vim_ad_") or item.startswith("narayana_ad_")
|
||
for item in ids
|
||
):
|
||
return "medium"
|
||
if any(item.startswith("vim_") or item.startswith("narayana_") for item in ids):
|
||
return "weak"
|
||
return "none"
|
||
|
||
|
||
def _tracks(rule_ids: Sequence[str]) -> list[str]:
|
||
tracks: list[str] = []
|
||
if any(str(item).startswith("vim_") for item in rule_ids):
|
||
tracks.append("vimshottari")
|
||
if any(str(item).startswith("narayana_") for item in rule_ids):
|
||
tracks.append("narayana")
|
||
return tracks
|
||
|
||
|
||
def _split_track_points(rule_ids: Sequence[str], points: float) -> tuple[float, float]:
|
||
vim = sum(str(item).startswith("vim_") for item in rule_ids)
|
||
narayana = sum(str(item).startswith("narayana_") for item in rule_ids)
|
||
total = vim + narayana
|
||
if total == 0:
|
||
return 0.0, 0.0
|
||
return points * vim / total, points * narayana / total
|
||
|
||
|
||
_LAYER_LABEL = {"d1": "本命上升", "d9": "D9", "d10": "D10", "d4": "D4"}
|
||
|
||
|
||
def window_scan(built: dict[str, Any]) -> dict[str, Any]:
|
||
"""D1/D9/D10/D4 diversity plus change minutes. Indices only; never sign names."""
|
||
counts: dict[str, set[int]] = {layer: set() for layer in _LAYER_LABEL}
|
||
transitions: list[dict[str, Any]] = []
|
||
previous: dict[str, int | None] | None = None
|
||
for feature in _features(built):
|
||
vargas = feature.get("varga_ascendants") if isinstance(feature.get("varga_ascendants"), dict) else {}
|
||
current = {
|
||
"d1": feature.get("ascendant_sign_index") if isinstance(feature.get("ascendant_sign_index"), int) else None,
|
||
"d9": vargas.get("D9") if isinstance(vargas.get("D9"), int) else None,
|
||
"d10": vargas.get("D10") if isinstance(vargas.get("D10"), int) else None,
|
||
"d4": vargas.get("D4") if isinstance(vargas.get("D4"), int) else None,
|
||
}
|
||
for layer, bucket in counts.items():
|
||
value = current[layer]
|
||
if isinstance(value, int):
|
||
bucket.add(value)
|
||
time = _feature_time(feature)
|
||
if previous and time:
|
||
for layer, label in _LAYER_LABEL.items():
|
||
before = previous[layer]
|
||
after = current[layer]
|
||
if isinstance(before, int) and isinstance(after, int) and before != after:
|
||
transitions.append({
|
||
"layer": layer,
|
||
"at": time,
|
||
"user_meaning": f"{label} 在 {time} 发生变化",
|
||
})
|
||
previous = current
|
||
return {
|
||
"scanned": True,
|
||
"confirmation_allowed": False,
|
||
"unique_minute_claim": False,
|
||
"d1_lagna_count": len(counts["d1"]),
|
||
"d9_lagna_count": len(counts["d9"]),
|
||
"d10_lagna_count": len(counts["d10"]),
|
||
"d4_lagna_count": len(counts["d4"]),
|
||
"d1_candidates_differ": len(counts["d1"]) > 1,
|
||
"d9_candidates_differ": len(counts["d9"]) > 1,
|
||
"d10_candidates_differ": len(counts["d10"]) > 1,
|
||
"d4_candidates_differ": len(counts["d4"]) > 1,
|
||
"transitions": transitions,
|
||
}
|
||
|
||
|
||
def event_dasha_ledger(
|
||
request: dict[str, Any],
|
||
built: dict[str, Any],
|
||
representative_time: str | None,
|
||
) -> list[dict[str, Any]]:
|
||
if not representative_time:
|
||
return []
|
||
matrix = built.get("matrix") or {}
|
||
rows: list[dict[str, Any]] = []
|
||
for event in request.get("events") or []:
|
||
if not isinstance(event, dict):
|
||
continue
|
||
contribution = (matrix.get(event.get("id")) or {}).get(representative_time)
|
||
if not isinstance(contribution, dict):
|
||
continue
|
||
rule_ids = contribution.get("rule_ids") or []
|
||
level = match_level(rule_ids)
|
||
summary = str(event.get("summary") or "").strip() or "这条经历"
|
||
tracks = _tracks(rule_ids)
|
||
track_text = "、".join(
|
||
"主限" if track == "vimshottari" else "分盘大运" for track in tracks
|
||
) or "现有大运层"
|
||
rows.append({
|
||
"summary": summary[:80],
|
||
"match": level,
|
||
"match_label": MATCH_LABELS[level],
|
||
"tracks": tracks,
|
||
"user_meaning": f"{summary[:40]}：{MATCH_LABELS[level]}（{track_text}）",
|
||
})
|
||
return rows
|
||
|
||
|
||
def dasha_agreement(built: dict[str, Any], candidate_times: Sequence[str]) -> dict[str, Any]:
|
||
times = [str(item)[:5] for item in candidate_times if isinstance(item, str) and len(str(item)) >= 5]
|
||
if not times:
|
||
return {
|
||
"status": "unavailable",
|
||
"vimshottari_top": None,
|
||
"narayana_top": None,
|
||
"user_meaning": "还没有足够的大运对照。",
|
||
}
|
||
vim_scores = {time: 0.0 for time in times}
|
||
narayana_scores = {time: 0.0 for time in times}
|
||
for contributions in (built.get("matrix") or {}).values():
|
||
if not isinstance(contributions, dict):
|
||
continue
|
||
for time in times:
|
||
cell = contributions.get(time)
|
||
if not isinstance(cell, dict):
|
||
continue
|
||
vim_points, narayana_points = _split_track_points(
|
||
cell.get("rule_ids") or [],
|
||
float(cell.get("points") or 0),
|
||
)
|
||
vim_scores[time] += vim_points
|
||
narayana_scores[time] += narayana_points
|
||
if all(value == 0 for value in vim_scores.values()) or all(value == 0 for value in narayana_scores.values()):
|
||
return {
|
||
"status": "partial",
|
||
"vimshottari_top": max(times, key=lambda time: vim_scores[time]) if any(vim_scores.values()) else None,
|
||
"narayana_top": max(times, key=lambda time: narayana_scores[time]) if any(narayana_scores.values()) else None,
|
||
"user_meaning": "主限和分盘大运还不能做成完整对照，只作观察。",
|
||
}
|
||
vim_top = max(times, key=lambda time: (vim_scores[time], -_clock(time)))
|
||
narayana_top = max(times, key=lambda time: (narayana_scores[time], -_clock(time)))
|
||
if vim_top == narayana_top:
|
||
return {
|
||
"status": "agree",
|
||
"vimshottari_top": vim_top,
|
||
"narayana_top": narayana_top,
|
||
"user_meaning": "主限和分盘大运都更支持同一段代表性时间。这仍不是唯一分钟确认。",
|
||
}
|
||
return {
|
||
"status": "conflict",
|
||
"vimshottari_top": vim_top,
|
||
"narayana_top": narayana_top,
|
||
"user_meaning": f"主限更偏向 {vim_top}，分盘大运更偏向 {narayana_top}。冲突时不能按更高把握收口。",
|
||
}
|
||
|
||
|
||
def _house_lord_zh(asc_idx: int, house: int) -> str | None:
|
||
sign = SIGNS[(asc_idx + house - 1) % 12]
|
||
lord = SIGN_LORDS.get(sign)
|
||
return PLANET_ZH.get(lord) if lord else None
|
||
|
||
|
||
def lagna_contrast(built: dict[str, Any]) -> dict[str, Any] | None:
|
||
features = _features(built)
|
||
if not features:
|
||
return None
|
||
intervals: list[dict[str, Any]] = []
|
||
current: dict[str, Any] | None = None
|
||
for feature in features:
|
||
time = _feature_time(feature)
|
||
index = feature.get("ascendant_sign_index")
|
||
if not time or not isinstance(index, int) or index < 0 or index > 11:
|
||
continue
|
||
if current and current["d1_lagna_index"] == index:
|
||
current["end"] = time
|
||
continue
|
||
if current:
|
||
intervals.append(current)
|
||
sign = SIGNS_CN.get(SIGNS[index])
|
||
current = {
|
||
"start": time,
|
||
"end": time,
|
||
"d1_lagna_index": index,
|
||
"lagna": sign,
|
||
"lords": {
|
||
"l1": _house_lord_zh(index, 1),
|
||
"l4": _house_lord_zh(index, 4),
|
||
"l7": _house_lord_zh(index, 7),
|
||
"l10": _house_lord_zh(index, 10),
|
||
},
|
||
}
|
||
if current:
|
||
intervals.append(current)
|
||
if len(intervals) < 2:
|
||
return None
|
||
left, right = intervals[0], intervals[1]
|
||
return {
|
||
"intervals": intervals[:3],
|
||
"user_meaning": (
|
||
f"窗口里出现两段本命上升：{left['start']}-{left['end']} 为{left['lagna']}，"
|
||
f"{right['start']}-{right['end']} 为{right['lagna']}。"
|
||
"只比较宫主结构，不给性格或类型标签。"
|
||
),
|
||
"unique_minute_claim": False,
|
||
}
|
||
|
||
|
||
def nakshatra_boundary(built: dict[str, Any], representative_time: str | None) -> dict[str, Any] | None:
|
||
features = {
|
||
_feature_time(feature): feature
|
||
for feature in _features(built)
|
||
if _feature_time(feature)
|
||
}
|
||
feature = features.get(representative_time or "") or (list(features.values())[0] if features else None)
|
||
if not isinstance(feature, dict):
|
||
return None
|
||
longitude = feature.get("ascendant_degree")
|
||
if not isinstance(longitude, (int, float)):
|
||
return None
|
||
wrapped = float(longitude) % 360.0
|
||
index = int(wrapped / NAKSHATRA_SPAN) % 27
|
||
position = wrapped % NAKSHATRA_SPAN
|
||
distance = min(position, NAKSHATRA_SPAN - position)
|
||
if distance > NAKSHATRA_BOUNDARY_DEGREES:
|
||
return {
|
||
"near_boundary": False,
|
||
"distance_degrees": round(distance, 4),
|
||
"user_meaning": None,
|
||
"options": [],
|
||
}
|
||
earlier_index = index if position <= NAKSHATRA_SPAN / 2 else (index - 1) % 27
|
||
later_index = (earlier_index + 1) % 27
|
||
earlier = NAKSHATRA_TRAITS[earlier_index]
|
||
later = NAKSHATRA_TRAITS[later_index]
|
||
return {
|
||
"near_boundary": True,
|
||
"distance_degrees": round(distance, 4),
|
||
"options": [
|
||
{
|
||
"key": "A",
|
||
"time_bias": "earlier",
|
||
"traits": list(earlier),
|
||
},
|
||
{
|
||
"key": "B",
|
||
"time_bias": "later",
|
||
"traits": list(later),
|
||
},
|
||
],
|
||
"user_meaning": (
|
||
"升点靠近两段日常节奏的交界。哪一组更像你近年的处事方式？"
|
||
f"A：{earlier[0]}；{earlier[1]}。"
|
||
f"B：{later[0]}；{later[1]}。"
|
||
"这只用来偏置时间窗，不能确认唯一分钟。"
|
||
),
|
||
}
|
||
|
||
|
||
def precision_stage(scan: dict[str, Any], event_count: int) -> dict[str, Any]:
|
||
if event_count <= 0:
|
||
current = "collect_events"
|
||
meaning = "还需要带大概时间的经历，才能开始缩小窗口。"
|
||
elif scan.get("d1_candidates_differ"):
|
||
current = "lagna_frame"
|
||
meaning = "本命上升还可能落在两段里。先补能分开这两段的带日期经历。"
|
||
elif scan.get("d9_candidates_differ"):
|
||
current = "d9_refine"
|
||
meaning = "本命上升已较稳，关系盘仍会换升。可再补一件记得时间的感情或关系变化。"
|
||
elif scan.get("d10_candidates_differ"):
|
||
current = "d10_refine"
|
||
meaning = "关系盘已较稳，事业盘仍会换升。可再补一件记得时间的工作变化。"
|
||
elif scan.get("d4_candidates_differ"):
|
||
current = "theme_refine"
|
||
meaning = "核心分盘已较稳。若还想收窄，可再补一件记得时间的家人或住处变化；也可以先采用代表性时间。"
|
||
else:
|
||
current = "ready_to_adopt"
|
||
meaning = "核心分盘已不再换升。可以采用代表性时间看盘，也可以再补主题经历。"
|
||
return {
|
||
"current": current,
|
||
"can_stop": current in {"d9_refine", "d10_refine", "theme_refine", "ready_to_adopt"},
|
||
"user_meaning": meaning,
|
||
"unique_minute_claim": False,
|
||
}
|
||
|
||
|
||
def oos_blind_prompts(request: dict[str, Any]) -> list[dict[str, Any]]:
|
||
covered = {
|
||
str(event.get("domain"))
|
||
for event in request.get("events") or []
|
||
if isinstance(event, dict) and event.get("domain")
|
||
}
|
||
catalog = (
|
||
("relationship", "校时还没用过感情这条线。有没有一件没提过、但记得大概时间的关系变化？"),
|
||
("career", "校时还没用过事业这条线。有没有一件没提过、但记得大概时间的工作变化？"),
|
||
("family", "校时还没用过家人这条线。有没有一件没提过、但记得大概时间的家人变化？"),
|
||
("education", "校时还没用过学习这条线。有没有一件没提过、但记得大概时间的学业变化？"),
|
||
)
|
||
prompts = [
|
||
{"domain": domain, "user_meaning": meaning, "used_for_scoring": False}
|
||
for domain, meaning in catalog
|
||
if domain not in covered
|
||
]
|
||
return prompts[:3]
|
||
|
||
|
||
def build_refinement_packet(
|
||
request: dict[str, Any],
|
||
built: dict[str, Any],
|
||
*,
|
||
representative_time: str | None,
|
||
candidate_times: Sequence[str],
|
||
) -> dict[str, Any]:
|
||
scan = window_scan(built)
|
||
agreement = dasha_agreement(built, candidate_times)
|
||
return {
|
||
"window_scan": scan,
|
||
"event_dasha_ledger": event_dasha_ledger(request, built, representative_time),
|
||
"dasha_agreement": agreement,
|
||
"lagna_contrast": lagna_contrast(built),
|
||
"nakshatra_boundary": nakshatra_boundary(built, representative_time),
|
||
"precision_stage": precision_stage(scan, len(request.get("events") or [])),
|
||
"oos_blind_prompts": oos_blind_prompts(request),
|
||
"unique_minute_claim": False,
|
||
"confirmation_allowed": False,
|
||
}
|