"""Public-safe biographical probes from candidate dasha / varga differences. Inverts event scoring: pick two representative minutes, find calendar years where Vimshottari + Narayana activation (or true period-start years) differ, and emit a yes/no life-event question. Never grants a unique minute. """ from __future__ import annotations from datetime import date, datetime, timedelta from typing import Any, Sequence from scripts.active_rectification_event_engine import ( DOMAIN_CONFIG, _active_narayana, _active_vimshottari, _score_event, ) import dasha_analyzer import narayana_dasha from scripts.rectification.refinement_packet import match_level MAX_PROBES = 3 LEVEL_RANK = {"none": 0, "weak": 1, "medium": 2, "strong": 3} SCORING_LAYERS = ("d1", "d9", "d10", "d4", "d5", "d24", "d7", "d12", "d2", "d11", "d30") LAYER_DOMAIN = { "d9": "relationship", "d10": "career", "d4": "relocation", "d5": "education", "d24": "education", "d7": "family", "d12": "family", "d2": "finance", "d11": "finance", "d30": "health_pressure", } STAGE_DOMAIN = { "d9_refine": "relationship", "d10_refine": "career", "d4_refine": "relocation", "theme_refine": "relocation", "d5_refine": "education", } VOLUNTEER_ONLY = frozenset({"finance", "health_pressure"}) DOMAIN_CATALOG: dict[str, dict[str, Any]] = { "education": { "event_family": "升学、高考、转学或学习环境变化", "quality_family": "高考或重要考试发挥明显失常、压力很大", "kind": "education_milestone", "age_lo": 16, "age_hi": 18, "varga": "D5 / D24", }, "relocation": { "event_family": "搬家、离乡或长期异地", "quality_family": "搬家、离乡或住宿结构明显变化", "kind": "home_change", "age_lo": 18, "age_hi": 24, "varga": "D4", }, "relationship": { "event_family": "认真关系进入、结束或关系观明显转变", "quality_family": "认真关系进入、结束或关系观明显转变", "kind": "relationship_change", "age_lo": 21, "age_hi": 26, "varga": "D9", }, "career": { "event_family": "入职、升职或职责明显加重", "quality_family": "入职、升职或职责明显加重", "kind": "career_change", "age_lo": 22, "age_hi": 30, "varga": "D10", }, "family": { "event_family": "家人相关的明显变化", "quality_family": "家人相关的明显变化", "kind": "family_event", "age_lo": 18, "age_hi": 30, "varga": "D12 / D7 / D3", }, "finance": { "event_family": "收入、资产或财务明显变化", "quality_family": "收入、资产或财务明显变化", "kind": "finance_change", "age_lo": 22, "age_hi": 32, "varga": "D2 / D11", }, "health_pressure": { "event_family": "健康、事故或持续压力明显变化", "quality_family": "健康、事故或持续压力明显变化", "kind": "self_health_event", "age_lo": 16, "age_hi": 40, "varga": "D30", }, } def _clock(value: str) -> int: return int(value[:2]) * 60 + int(value[3:5]) def _context_time(context: dict[str, Any]) -> str | None: feature = context.get("feature") if isinstance(context.get("feature"), dict) else {} raw = feature.get("time") if isinstance(raw, str) and len(raw) >= 5: return raw[:5] at = context.get("candidate_at") if isinstance(at, datetime): return at.strftime("%H:%M") return None def _birth_year(value: object) -> int | None: text = str(value or "").strip() if len(text) < 4 or not text[:4].isdigit(): return None year = int(text[:4]) return year if 1900 <= year <= 2100 else None def _event_year(event: dict[str, Any]) -> int | None: for key in ("date", "date_start", "occurred_from"): year = _birth_year(event.get(key)) if year is not None: return year return None def _year_label(year: int) -> str: return f"{year} 年前后" def _age_band_year(birth_year: int, domain: str, today: date) -> int | None: catalog = DOMAIN_CATALOG.get(domain) if not catalog: return None age = (int(catalog["age_lo"]) + int(catalog["age_hi"])) // 2 year = birth_year + age latest = min(today.year, birth_year + 80) earliest = birth_year + 5 if year < earliest or year > latest: return None return year def _probe_domains( scan: dict[str, Any], precision_current: str | None, events: Sequence[dict[str, Any]], ) -> list[str]: volunteered = { str(event.get("domain")) for event in events if isinstance(event, dict) and event.get("domain") } ordered: list[str] = [] stage_domain = STAGE_DOMAIN.get(str(precision_current or "")) if stage_domain: ordered.append(stage_domain) for layer in SCORING_LAYERS: domain = LAYER_DOMAIN.get(layer) if not domain or domain in ordered: continue if not scan.get(f"{layer}_candidates_differ"): continue if domain in VOLUNTEER_ONLY and domain not in volunteered: continue ordered.append(domain) if not ordered and scan.get("d1_candidates_differ"): ordered.append("education") return ordered def _static_contexts(built: dict[str, Any]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for context in built.get("static_contexts") or []: if isinstance(context, dict) and _context_time(context): rows.append(context) rows.sort(key=lambda item: _clock(str(_context_time(item)))) return rows def _pick_representatives( built: dict[str, Any], scan: dict[str, Any], candidate_times: Sequence[str], representative_time: str | None, ) -> tuple[dict[str, Any], dict[str, Any]] | None: contexts = _static_contexts(built) by_time = {_context_time(item): item for item in contexts} times = [str(_context_time(item)) for item in contexts] for transition in scan.get("transitions") or []: if not isinstance(transition, dict): continue layer = transition.get("layer") at = str(transition.get("at") or "")[:5] if layer not in SCORING_LAYERS or at not in by_time: continue index = times.index(at) left = times[index - 1] if index > 0 else at if left != at: return by_time[left], by_time[at] picked: list[str] = [] for raw in [*candidate_times, representative_time]: time = str(raw or "")[:5] if len(time) >= 5 and time in by_time and time not in picked: picked.append(time) if len(picked) >= 2: return by_time[picked[0]], by_time[picked[-1]] if len(times) >= 2: return by_time[times[0]], by_time[times[-1]] return None def _scoreable(context: dict[str, Any]) -> bool: chart = context.get("chart") planets = context.get("planet_longitudes") vargas = context.get("varga_charts") return ( isinstance(chart, dict) and isinstance(planets, dict) and isinstance(vargas, dict) and isinstance(planets.get("Moon"), (int, float)) and isinstance(context.get("ascendant_index"), int) ) def _candidate_at(context: dict[str, Any], birth_date: str) -> datetime | None: at = context.get("candidate_at") if isinstance(at, datetime): return at time = _context_time(context) if not time: return None try: return datetime.strptime(f"{birth_date} {time}", "%Y-%m-%d %H:%M") except ValueError: return None def _has_domain_activation(rule_ids: Sequence[str]) -> bool: return any( ("_domain_" in str(item) or str(item).endswith("_domain_house") or str(item).endswith("_domain_lord") or str(item).endswith("_domain_varga")) and not str(item).startswith("no_") for item in rule_ids ) def _discriminates(left: Sequence[str], right: Sequence[str]) -> bool: left_rank = LEVEL_RANK.get(match_level(left), 0) right_rank = LEVEL_RANK.get(match_level(right), 0) if abs(left_rank - right_rank) >= 2: return True return _has_domain_activation(left) != _has_domain_activation(right) def _tracks_present(rule_ids: Sequence[str]) -> tuple[bool, bool]: text = [str(item) for item in rule_ids] return ( any(item.startswith("vim_") for item in text), any(item.startswith("narayana_") for item in text), ) def _vim_start_years(birth_date: str, moon_longitude: float, lo: int, hi: int) -> list[int]: nakshatra, progress, _ = dasha_analyzer.lon_to_nakshatra(float(moon_longitude)) timeline, _, _, _ = dasha_analyzer.build_dasha_timeline(birth_date, nakshatra, progress) years: list[int] = [] for major in timeline: start = major.get("start") if isinstance(start, datetime) and lo <= start.year <= hi: years.append(start.year) for minor in dasha_analyzer.build_antardasha(major): minor_start = minor.get("start") if isinstance(minor_start, datetime) and lo <= minor_start.year <= hi: years.append(minor_start.year) return years def _narayana_start_years( ascendant_index: int, planet_longitudes: dict[str, float], birth_date: str, lo: int, hi: int, ) -> list[int] | None: periods = narayana_dasha.calc_narayana_mahadasha(ascendant_index, planet_longitudes) if not periods: return None birth = datetime.strptime(birth_date, "%Y-%m-%d") years: list[int] = [] for major in periods: start_age = major.get("start_age") if not isinstance(start_age, (int, float)): return None year = (birth + timedelta(days=float(start_age) * 365.2425)).year if lo <= year <= hi: years.append(year) antars = narayana_dasha.calc_narayana_antardasha(periods, int(major["sign_idx"])) for minor in antars: minor_age = minor.get("start_age") if not isinstance(minor_age, (int, float)): continue minor_year = (birth + timedelta(days=float(minor_age) * 365.2425)).year if lo <= minor_year <= hi: years.append(minor_year) return years def _boundary_years(left: list[int], right: list[int]) -> set[int]: years: set[int] = set() for one, two in zip(left, right): if abs(one - two) >= 1: years.add(one) years.add(two) return years def _score_year( context: dict[str, Any], *, birth_date: str, domain: str, year: int, ) -> dict[str, Any] | None: catalog = DOMAIN_CATALOG[domain] prefixes, _ = DOMAIN_CONFIG[domain] varga_charts = context.get("varga_charts") or {} domain_vargas = [varga_charts.get(prefix) for prefix in prefixes] if any(item is None for item in domain_vargas): return None candidate_at = _candidate_at(context, birth_date) moon = (context.get("planet_longitudes") or {}).get("Moon") if candidate_at is None or not isinstance(moon, (int, float)): return None event_at = datetime(year, 7, 1) event = { "id": f"probe-{domain}-{year}", "domain": domain, "event_kind": catalog["kind"], "date": f"{year}-07-01", "precision": "year", "summary": catalog["event_family"], } try: vimshottari = _active_vimshottari(birth_date, float(moon), event_at) narayana = _active_narayana( int(context["ascendant_index"]), context["planet_longitudes"], candidate_at, event_at, ) except (KeyError, TypeError, ValueError): return None if narayana[0] is None or narayana[1] is None: return None return _score_event( candidate_time=str(_context_time(context)), event=event, natal_chart=context["chart"], varga_charts=[item for item in domain_vargas if item is not None], vimshottari=vimshottari, narayana=narayana, arudha_padas=context.get("arudha_padas") or {}, ) def _agent_brief( *, year_label: str, family: str, quality: bool = False, exam: bool = False, ) -> str: if exam: return ( f"年份锁定 {year_label}。已有高考或考试经历。" "请写成一句自然语言，问那次是否发挥失常或压力特别大。" "不得改年份。" ) if quality: return ( f"年份锁定 {year_label}。已有相关经历。" f"请写成一句自然语言，问{family}有没有发生过。" "不得改年份。" ) return ( f"年份锁定 {year_label}。事件家族：{family}。" "请写成一句自然语言是/否题。" "不得改年份。" ) def _public_probe( *, year: int, domain: str, source: str, tracks: Sequence[str], tracks_agree: bool, user_meaning: str, event_family: str, ) -> dict[str, Any]: return { "year": year, "year_label": _year_label(year), "domain": domain, "event_family": event_family, "source": source, "tracks": list(tracks), "tracks_agree": tracks_agree, "unique_minute_claim": False, "user_meaning": user_meaning, "role": "distinguish" if source == "known_event_quality" else "reverse_verify", } def _quality_probes( events: Sequence[dict[str, Any]], domains: Sequence[str], ) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] seen: set[tuple[str, int]] = set() for event in events: if not isinstance(event, dict): continue domain = str(event.get("domain") or "") year = _event_year(event) if domain not in domains or year is None or (domain, year) in seen: continue seen.add((domain, year)) summary = str(event.get("summary") or "") family = str(DOMAIN_CATALOG[domain]["quality_family"]) exam = domain == "education" and ("高考" in summary or "考试" in summary) rows.append(_public_probe( year=year, domain=domain, source="known_event_quality", tracks=("vimshottari", "narayana"), tracks_agree=True, user_meaning=_agent_brief( year_label=_year_label(year), family=family, quality=True, exam=exam, ), event_family=family, )) return rows def _evaluate_year( left: dict[str, Any], right: dict[str, Any], *, birth_date: str, domain: str, year: int, source: str, ) -> dict[str, Any] | None: scored_left = _score_year(left, birth_date=birth_date, domain=domain, year=year) scored_right = _score_year(right, birth_date=birth_date, domain=domain, year=year) if scored_left is None or scored_right is None: return None left_rules = scored_left.get("rule_ids") or [] right_rules = scored_right.get("rule_ids") or [] if not _discriminates(left_rules, right_rules): return None stronger = left_rules if LEVEL_RANK[match_level(left_rules)] >= LEVEL_RANK[match_level(right_rules)] else right_rules vim_hit, narayana_hit = _tracks_present(stronger) return _public_probe( year=year, domain=domain, source=source, tracks=("vimshottari", "narayana"), tracks_agree=vim_hit and narayana_hit, user_meaning=_agent_brief( year_label=_year_label(year), family=str(DOMAIN_CATALOG[domain]["event_family"]), ), event_family=str(DOMAIN_CATALOG[domain]["event_family"]), ) def discriminating_event_probes( request: dict[str, Any], built: dict[str, Any], *, scan: dict[str, Any], candidate_times: Sequence[str], representative_time: str | None, precision_current: str | None = None, today: date | None = None, ) -> list[dict[str, Any]]: birth_date = str(request.get("birth_date") or "").strip() birth_year = _birth_year(birth_date) if birth_year is None: return [] try: datetime.strptime(birth_date, "%Y-%m-%d") except ValueError: return [] now = today or date.today() events = [item for item in (request.get("events") or []) if isinstance(item, dict)] domains = _probe_domains(scan, precision_current, events) if not domains: return [] probes = _quality_probes(events, domains) covered_domains = {row["domain"] for row in probes} pair = _pick_representatives(built, scan, candidate_times, representative_time) lo, hi = birth_year + 5, min(now.year, birth_year + 80) can_score = ( pair is not None and _scoreable(pair[0]) and _scoreable(pair[1]) ) dasha_domains: set[str] = set() if can_score and pair is not None: left, right = pair left_moon = float(left["planet_longitudes"]["Moon"]) right_moon = float(right["planet_longitudes"]["Moon"]) vim_years = _boundary_years( _vim_start_years(birth_date, left_moon, lo, hi), _vim_start_years(birth_date, right_moon, lo, hi), ) left_narayana = _narayana_start_years(int(left["ascendant_index"]), left["planet_longitudes"], birth_date, lo, hi) right_narayana = _narayana_start_years(int(right["ascendant_index"]), right["planet_longitudes"], birth_date, lo, hi) narayana_years: set[int] = set() if left_narayana is not None and right_narayana is not None: narayana_years = _boundary_years(left_narayana, right_narayana) for domain in domains: if domain in covered_domains: continue boundary = sorted((vim_years | narayana_years) & set(range(lo, hi + 1))) found = None for year in boundary: found = _evaluate_year( left, right, birth_date=birth_date, domain=domain, year=year, source="dasha_boundary", ) if found: break if found is None: midpoint = _age_band_year(birth_year, domain, now) if midpoint is not None: found = _evaluate_year( left, right, birth_date=birth_date, domain=domain, year=midpoint, source="dasha_activation", ) if found: probes.append(found) dasha_domains.add(domain) covered_domains.add(domain) for domain in domains: if domain in covered_domains or domain in dasha_domains: continue year = _age_band_year(birth_year, domain, now) if year is None: continue probes.append(_public_probe( year=year, domain=domain, source="age_band", tracks=("vimshottari", "narayana"), tracks_agree=False, user_meaning=_agent_brief( year_label=_year_label(year), family=str(DOMAIN_CATALOG[domain]["event_family"]), ), event_family=str(DOMAIN_CATALOG[domain]["event_family"]), )) covered_domains.add(domain) public: list[dict[str, Any]] = [] seen: set[tuple[str, int, str]] = set() for row in probes: key = (str(row["domain"]), int(row["year"]), str(row["source"])) encoded = str(row) if key in seen or "points" in encoded: continue seen.add(key) public.append(row) if len(public) >= MAX_PROBES: break return public