from __future__ import annotations from typing import Any, Mapping, Sequence from uuid import NAMESPACE_URL, uuid5 from scripts.active_rectification_events import build_candidate_result_summary from scripts.rectification.candidate_feature_service import build_candidate_feature_snapshot from scripts.rectification.contracts import ( EVENT_CONTRACT_VERSION, RectificationRequest, is_primary_scoreable_event, ) from scripts.rectification.decision_policy import ( EXECUTION_LEDGER_VERSION, POLICY_VERSION, build_candidate_decisions, build_decision_receipt, build_execution_ledger, ) from scripts.rectification.diagnostics_service import run_diagnostics from scripts.rectification.scoring_service import ( ALGORITHM_VERSION, build_event_contribution_matrix, calculation_spec, score_from_matrix, scoreable_request, sha256, ) def _clock_minutes(value: str) -> int: hour, minute = value[:5].split(":", 1) return int(hour) * 60 + int(minute) def _window_width(start_time: str, end_time: str) -> int: return (_clock_minutes(end_time) - _clock_minutes(start_time)) % 1_440 + 1 def _report_candidate_range( request: RectificationRequest, candidate_scores: Sequence[dict[str, Any]], representative_time: str | None, ) -> dict[str, Any]: top_score = max((float(row.get("score") or 0) for row in candidate_scores), default=None) top_times = [ str(row.get("time"))[:5] for row in candidate_scores if top_score is not None and float(row.get("score") or 0) == top_score ] if not top_times: return { "start_time": request["start_time"], "end_time": request["end_time"], "representative_time": representative_time, "width_minutes": _window_width(request["start_time"], request["end_time"]), "representative_is_unique": False, } return { "start_time": top_times[0], "end_time": top_times[-1], "representative_time": representative_time or top_times[len(top_times) // 2], "width_minutes": len(top_times), "representative_is_unique": False, } def _report_evidence( request: RectificationRequest, built: dict[str, Any], representative_time: str | None, ) -> list[dict[str, Any]]: matrix = built.get("matrix") or {} rows: list[dict[str, Any]] = [] for event in request.get("events") or []: if not is_primary_scoreable_event(event): continue contribution = (matrix.get(event["id"]) or {}).get(representative_time or "") contribution = contribution if isinstance(contribution, dict) else {} points = float(contribution.get("points") or 0) status = "supporting" if points > 0 else "contradictory" if points < 0 else "unconfirmed" rows.append({ "event_id": event["id"], "summary": str(event.get("summary") or "").strip(), "domain": event["domain"], "date": { "start": event["date_start"], "end": event["date_end"], "precision": event["precision"], }, "status": status, "supports_candidate_time": representative_time if status == "supporting" else None, "methods": sorted({str(layer) for layer in contribution.get("technique_layers") or []}), }) return rows def _report_excluded_candidates( candidate_decisions: Sequence[dict[str, Any]], representative_time: str | None, ) -> list[dict[str, Any]]: return [ { "time": str(candidate.get("time") or "")[:5], "reason": "not_the_leading_candidate", "representative_time": representative_time, } for candidate in candidate_decisions if str(candidate.get("time") or "")[:5] != (representative_time or "") ] def _confirmation_blockers(receipt: dict[str, Any]) -> list[dict[str, str]]: allowed = {"VedAstro 分钟级校验", "唯一分钟确认"} return [ { "technique": str(row.get("technique")), "status": str(row.get("status")), "user_meaning": str(row.get("note") or ""), } for row in receipt.get("technique_audit_table") or [] if isinstance(row, dict) and str(row.get("technique")) in allowed and str(row.get("status")) != "executed" ] def _rectification_report( request: RectificationRequest, built: dict[str, Any], candidate_scores: Sequence[dict[str, Any]], candidate_decisions: Sequence[dict[str, Any]], receipt: dict[str, Any], ) -> dict[str, Any]: representative_time = str(receipt.get("representative_time") or "")[:5] or None blockers = _confirmation_blockers(receipt) candidate_range = _report_candidate_range(request, candidate_scores, representative_time) limitations = [item["user_meaning"] for item in blockers if item["user_meaning"]] if not limitations: limitations.append("本会话以代表性时间收口，不确认唯一分钟。") return { "candidate_range": candidate_range, "representative_time": representative_time, "representative_label": "代表性候选，不是唯一解", "confidence": receipt.get("overall_confidence", "low"), "evidence": _report_evidence(request, built, representative_time), "excluded_candidates": _report_excluded_candidates(candidate_decisions, representative_time), "next_step_codes": [], "confirmation_gate_blockers": blockers, "limitations": limitations, "claim_status": "candidate_range_not_birth_time_truth", } def engine_scoring_versions() -> dict[str, str]: """Identity fields also returned by `/api/rectification/v5/score`, without scoring.""" return { "algorithm_version": ALGORITHM_VERSION, "event_contract_version": EVENT_CONTRACT_VERSION, "decision_policy_version": POLICY_VERSION, } def candidate_features(request: RectificationRequest) -> dict[str, Any]: spec = calculation_spec(request) spec_hash = sha256(spec) scoring_request = scoreable_request(request) return { "algorithm_version": ALGORITHM_VERSION, "event_contract_version": EVENT_CONTRACT_VERSION, "decision_policy_version": POLICY_VERSION, "calculation_spec": spec, "calculation_spec_hash": spec_hash, "candidate_feature_snapshot": build_candidate_feature_snapshot(scoring_request, spec_hash), "can_confirm_exact_minute": False, } def score_candidates(request: RectificationRequest) -> dict[str, Any]: scoring_request = scoreable_request(request) built = build_event_contribution_matrix(scoring_request) rows = score_from_matrix(scoring_request, built) spec = calculation_spec(request) spec_hash = sha256(spec) diagnostic_values = run_diagnostics(scoring_request, rows, built) fingerprint = sha256({ key: value for key, value in request.items() if key != "asked_probe_keys" }) result_id = str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{fingerprint}")) candidate_decisions = build_candidate_decisions( rows, result_id=result_id, static_contexts=built.get("static_contexts") if isinstance(built.get("static_contexts"), list) else None, ) decision_receipt = build_decision_receipt(request, candidate_decisions, built, diagnostic_values) execution_ledger = build_execution_ledger(request, built, diagnostic_values, candidate_decisions) representative = candidate_decisions[0] if candidate_decisions else None representative_time = str(representative.get("time") or "")[:5] if representative else None report_range = _report_candidate_range(request, rows, representative_time) report_evidence = _report_evidence(request, built, representative_time) summary_evidence = [ { "event_id": item["event_id"], "domain": item["domain"], "candidate_time": representative_time or "", "rule_ids": item["methods"], "points": 1 if item["status"] == "supporting" else -1 if item["status"] == "contradictory" else 0, } for item in report_evidence ] candidate_summary = build_candidate_result_summary({ "winning_segment": report_range, "event_count": len(scoring_request.get("events", [])), "margin_percent": decision_receipt.get("margin_percent", 0), "reasons": decision_receipt.get("reasons", []), "evidence": summary_evidence, }) candidate_summary["stability"] = {"label": decision_receipt.get("overall_confidence", "low")} rectification_report = _rectification_report( request, built, [{ "time": row["time"], "score": row["score"], } for row in rows], candidate_decisions, decision_receipt, ) rectification_report["next_step_codes"] = candidate_summary["next_step_codes"] candidate_summary["report"] = rectification_report return { "result_id": result_id, "algorithm_version": ALGORITHM_VERSION, "event_contract_version": EVENT_CONTRACT_VERSION, "decision_policy_version": POLICY_VERSION, "calculation_spec": spec, "calculation_spec_hash": spec_hash, "candidate_scores": [{ "time": row["time"], "score": row["score"], "supporting_event_ids": [item["event_id"] for item in row["evidence"] if item["points"] > 0], "conflicting_event_ids": [item["event_id"] for item in row["evidence"] if item["points"] < 0], } for row in rows], "candidate_decisions": candidate_decisions, "candidate_decision_receipt": decision_receipt, "decision_receipt": decision_receipt, "execution_ledger_version": EXECUTION_LEDGER_VERSION, "execution_ledger": execution_ledger, "event_contribution_matrix": built["matrix"], "candidate_feature_snapshot": build_candidate_feature_snapshot( scoring_request, spec_hash, built.get("static_contexts") ), "diagnostics": diagnostic_values, "candidate_summary": candidate_summary, "next_step_codes": candidate_summary["next_step_codes"], "stability": candidate_summary["stability"], "rectification_report": rectification_report, "robustness": { "neighbor_support_minutes": diagnostic_values.get("neighbor_support_minutes", 0), "leave_one_out_retention_rate": diagnostic_values.get("leave_one_event_out_retention_rate", 0), "leave_one_domain_out_retention_rate": diagnostic_values.get("leave_one_domain_out_retention_rate", 0), "date_sensitivity_retention_rate": diagnostic_values.get("date_sensitivity_retention_rate", 0), }, "missing_layers": built["missing_layers"], "display_allowed": decision_receipt["display_allowed"], "selection_allowed": decision_receipt["selection_allowed"], "acceptance_allowed": decision_receipt["acceptance_allowed"], "propose_allowed": decision_receipt["propose_allowed"], "confirmation_allowed": decision_receipt["confirmation_allowed"], "representative_candidate_id": representative["candidate_id"] if representative else None, "representative_time": representative["time"] if representative else None, "overall_confidence": decision_receipt["overall_confidence"], "margin_percent": decision_receipt["margin_percent"], "can_confirm_exact_minute": decision_receipt["confirmation_allowed"], } def diagnostics(request: RectificationRequest) -> dict[str, Any]: scored = score_candidates(request) return { "result_id": scored["result_id"], "algorithm_version": scored["algorithm_version"], "event_contract_version": scored["event_contract_version"], "decision_policy_version": scored["decision_policy_version"], "calculation_spec_hash": scored["calculation_spec_hash"], "candidate_decisions": scored["candidate_decisions"], "candidate_decision_receipt": scored["candidate_decision_receipt"], "decision_receipt": scored["decision_receipt"], "execution_ledger_version": scored["execution_ledger_version"], "execution_ledger": scored["execution_ledger"], "diagnostics": scored["diagnostics"], "candidate_summary": scored.get("candidate_summary", {"next_step_codes": ["do_not_apply_as_birth_time_truth"]}), "next_step_codes": scored.get("next_step_codes", ["do_not_apply_as_birth_time_truth"]), "stability": scored.get("stability", {"label": scored.get("overall_confidence", "low")}), "rectification_report": scored.get("rectification_report", {}), "missing_layers": scored["missing_layers"], "display_allowed": scored["display_allowed"], "selection_allowed": scored["selection_allowed"], "acceptance_allowed": scored["acceptance_allowed"], "propose_allowed": scored["propose_allowed"], "confirmation_allowed": scored["confirmation_allowed"], "representative_candidate_id": scored["representative_candidate_id"], "representative_time": scored["representative_time"], "overall_confidence": scored["overall_confidence"], "margin_percent": scored["margin_percent"], "can_confirm_exact_minute": scored["confirmation_allowed"], } def range_reading(request: Mapping[str, Any]) -> dict[str, Any]: """Stable vs minute-sensitive themes for one unresolved clock window.""" from types import SimpleNamespace from scripts.jyotish_engine import _build_birth_time_sensitivity body = dict(request or {}) birth_date = str(body.get("birth_date") or "").strip() if birth_date: year_text, month_text, day_text = birth_date.split("-", 2) year, month, day = int(year_text), int(month_text), int(day_text) else: year, month, day = int(body["year"]), int(body["month"]), int(body["day"]) representative = str(body.get("representative_time") or "")[:5] if len(representative) == 5 and representative[2] == ":": hour, minute = int(representative[:2]), int(representative[3:]) else: hour = int(body.get("hour") or 12) minute = int(body.get("minute") or 0) representative = f"{hour:02d}:{minute:02d}" raw_range = body.get("candidate_range") if isinstance(raw_range, Mapping): start_time = str(raw_range.get("start_time") or "")[:5] end_time = str(raw_range.get("end_time") or "")[:5] representative = str(raw_range.get("representative_time") or representative)[:5] else: start_time = str(body.get("start_time") or "")[:5] end_time = str(body.get("end_time") or "")[:5] hour, minute = int(representative[:2]), int(representative[3:]) accuracy = str(body.get("birth_time_accuracy") or "provisional") args = SimpleNamespace( year=year, month=month, day=day, hour=hour, minute=minute, second=0, lat=float(body["lat"]), lon=float(body["lon"]), tz=float(body["tz"]), ayanamsa=body.get("ayanamsa") or "raman", node_mode=body.get("node_mode") or "mean", birth_time_accuracy=accuracy, candidate_range={ "start_time": start_time, "end_time": end_time, "representative_time": representative, }, representative_time=representative, declared_window_start=None, declared_window_end=None, uncertainty_before_minutes=None, uncertainty_after_minutes=None, ) sensitivity = _build_birth_time_sensitivity(args) themes = sensitivity.get("theme_sensitivity") themes = themes if isinstance(themes, dict) else {} stable = [ key for key, row in themes.items() if isinstance(row, dict) and row.get("status") == "stable" ] sensitive = [ key for key, row in themes.items() if isinstance(row, dict) and row.get("status") == "sensitive" ] return { "window": sensitivity.get("window"), "stable_themes": stable, "sensitive_themes": sensitive, "claim_boundary": sensitivity.get("claim_boundary"), "theme_sensitivity": themes, "accuracy": sensitivity.get("accuracy"), "status": sensitivity.get("status"), } BLOCK_SCAN_PERIODS: tuple[tuple[str, str, str], ...] = ( ("early_morning", "04:00", "07:59"), ("morning", "08:00", "11:59"), ("afternoon", "12:00", "17:59"), ("evening", "18:00", "22:59"), ("late_night", "23:00", "03:59"), ) def _clock_in_declared_period(clock: str, start_time: str, end_time: str) -> bool: current = _clock_minutes(clock[:5]) start = _clock_minutes(start_time) end = _clock_minutes(end_time) if start <= end: return start <= current <= end return current >= start or current <= end def _normalize_relative_support(raw: Sequence[float]) -> list[float]: floored = [max(0.0, float(value)) for value in raw] total = sum(floored) if total <= 0: return [20.0 for _ in floored] shares = [round(100.0 * value / total, 1) for value in floored] delta = round(100.0 - sum(shares), 1) if shares: shares[shares.index(max(shares))] = round(shares[shares.index(max(shares))] + delta, 1) return shares def block_scan(request: RectificationRequest) -> dict[str, Any]: """Aggregate 24h event scores into the five declared birth-time periods.""" step = int(request.get("minute_step") or 10) if step <= 1: step = 10 scoring_request = {**request, "minute_step": step} scored = score_candidates(scoring_request) events_by_id = { str(event.get("id") or ""): event for event in request.get("events") or [] if isinstance(event, dict) } rows = [ row for row in scored.get("candidate_scores") or [] if isinstance(row, dict) and str(row.get("time") or "")[:5] ] day_scores = [float(row.get("score") or 0) for row in rows] min_day = min(day_scores) if day_scores else 0.0 raw_support: list[float] = [] blocks: list[dict[str, Any]] = [] for period, start_time, end_time in BLOCK_SCAN_PERIODS: members = [ row for row in rows if _clock_in_declared_period(str(row.get("time") or "")[:5], start_time, end_time) ] counts: dict[str, int] = {} for row in members: for event_id in row.get("supporting_event_ids") or []: key = str(event_id) if key: counts[key] = counts.get(key, 0) + 1 top_events = [] for event_id, _count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))[:3]: event = events_by_id.get(event_id) or {} top_events.append({ "event_id": event_id, "domain": event.get("domain"), "summary": event.get("summary"), }) member_scores = [float(row.get("score") or 0) for row in members] mean = (sum(member_scores) / len(member_scores)) if member_scores else 0.0 raw_support.append(max(mean - min_day, 0.0)) blocks.append({ "period": period, "start_time": start_time, "end_time": end_time, "relative_support": 0, "top_events": top_events, "candidate_count": len(members), }) shares = _normalize_relative_support(raw_support) for block, share in zip(blocks, shares): block["relative_support"] = share receipt = scored.get("decision_receipt") if isinstance(scored.get("decision_receipt"), dict) else {} return { "result_id": scored.get("result_id"), "algorithm_version": scored.get("algorithm_version"), "calculation_spec": scored.get("calculation_spec"), "calculation_spec_hash": scored.get("calculation_spec_hash"), "minute_step": step, "candidate_count": len(rows), "precision_stage": {"current": "block_scan"}, "blocks": blocks, "discriminating_event_probes": [], "acceptance_allowed": False, "selection_allowed": False, "display_allowed": False, "decision_receipt": { **receipt, "precision_stage": {"current": "block_scan"}, "discriminating_event_probes": [], "acceptance_allowed": False, "selection_allowed": False, }, }