#!/usr/bin/env python3 """Compute-only mixins lifted verbatim out of ``JyotishAPIHandler``. These methods never touch HTTP context (``headers`` / ``wfile`` / ``rfile`` / ``path`` / ``client_address``), so offline callers (MCP, reporting scripts) can instantiate the mixins directly instead of forging a ``JyotishAPIHandler`` instance that never ran its constructor. Hard constraint: this module must never import ``jyotish_api_server`` -- that would merely move the backdoor. See tests/test_api_server_growth_contract.py. """ from __future__ import annotations import math import os class BadRequest(ValueError): """Client-side request validation failed.""" def _build_vedastro_official_full_snapshot_payload_from_chart(chart): modules = chart.get('modules') if isinstance(chart, dict) else {} snapshot = modules.get('vedastro_official_full_snapshot') if isinstance(modules, dict) else {} strict_workflow_contracts = snapshot.get('strict_workflow_contracts') if isinstance(snapshot, dict) else {} if not isinstance(strict_workflow_contracts, dict): strict_workflow_contracts = {} if not isinstance(snapshot, dict) or not snapshot: return { 'status': 'blocked', 'available': False, 'operation': 'official_full_snapshot', 'primary_source': 'vedastro_official', 'strict_workflow_primary_route': None, 'strict_workflow_routes_available': [], 'strict_workflow_contracts': {}, 'boundary_note': 'VedAstro official full snapshot is not attached.', } manifest = snapshot.get('request_manifest') if isinstance(snapshot.get('request_manifest'), dict) else {} requests = manifest.get('requests') if isinstance(manifest.get('requests'), list) else [] sections = snapshot.get('snapshot_sections') if isinstance(snapshot.get('snapshot_sections'), dict) else {} metadata = snapshot.get('source_metadata') if isinstance(snapshot.get('source_metadata'), dict) else {} official_bundle = metadata.get('official_python_bundle') if isinstance(metadata.get('official_python_bundle'), dict) else {} full_catalog = metadata.get('official_full_capability_catalog') if isinstance(metadata.get('official_full_capability_catalog'), dict) else {} coverage = official_bundle.get('coverage') if isinstance(official_bundle.get('coverage'), dict) else {} official_chart = snapshot.get('official_chart') if isinstance(snapshot.get('official_chart'), dict) else {} dynamic_selection = full_catalog.get('dynamic_selection') if isinstance(full_catalog.get('dynamic_selection'), dict) else {} report_references = { theme: selection.get('report_reference') for theme, selection in dynamic_selection.items() if isinstance(selection, dict) and isinstance(selection.get('report_reference'), dict) } return { 'status': snapshot.get('status') or 'blocked', 'available': bool(snapshot.get('available')), 'operation': snapshot.get('operation') or 'official_full_snapshot', 'primary_source': snapshot.get('primary_source') or 'vedastro_official', 'official_python_path': metadata.get('official_python_path'), 'official_bundle_status': official_bundle.get('status'), 'official_primary_sections_ok': coverage.get('filled_sections') or [], 'official_chart_available': bool(official_chart.get('planets')) and bool(official_chart.get('ascendant')), 'official_full_capability_catalog_status': full_catalog.get('status'), 'official_full_capability_catalog_summary': full_catalog.get('summary') or {}, 'official_full_capability_catalog_coverage': full_catalog.get('coverage') or {}, 'official_full_capability_domain_routing': full_catalog.get('domain_routing') or {}, 'official_full_capability_dynamic_selection': dynamic_selection, 'official_report_references': report_references, 'strict_workflow_primary_route': snapshot.get('strict_workflow_primary_route'), 'strict_workflow_routes_available': snapshot.get('strict_workflow_routes_available') or list(strict_workflow_contracts.keys()), 'strict_workflow_contracts': strict_workflow_contracts, 'section_statuses': snapshot.get('section_statuses') or {}, 'snapshot_section_keys': sorted(sections.keys()), 'request_section_count': len(requests), 'request_sections': [item.get('section') for item in requests if isinstance(item, dict)], 'method_catalog': manifest.get('method_catalog') or {}, 'user_visibility': snapshot.get('user_visibility') or 'backend_raw_evidence_not_direct_user_report', 'source_metadata': snapshot.get('source_metadata') or {}, 'boundary_note': ( snapshot.get('reason') or 'VedAstro official full snapshot is the primary raw evidence layer; user reports consume selected slices only.' ), } def _free_tier_queue_enabled_env() -> bool: raw_values = [ str(os.environ.get("VEDASTRO_FREE_TIER_QUEUE", "")).strip().lower(), str(os.environ.get("VEDASTRO_FREE_TIER_QUEUE_ENABLED", "")).strip().lower(), str(os.environ.get("VEDASTRO_ENABLE_FREE_TIER_QUEUE", "")).strip().lower(), ] return any(value in {"1", "true", "yes", "on"} for value in raw_values) def _preferred_strict_contract(strict_workflow_contracts, primary_route=None): if not isinstance(strict_workflow_contracts, dict) or not strict_workflow_contracts: return None, {} route = primary_route if primary_route in strict_workflow_contracts else next(iter(strict_workflow_contracts.keys())) contract = strict_workflow_contracts.get(route) return route, contract if isinstance(contract, dict) else {} class RequestParamMixin: """Request-parameter coercion helpers (raise :class:`BadRequest`).""" def _get_float(self, body, key, default, min_value=None, max_value=None): value = body.get(key, default) try: number = float(value) except (TypeError, ValueError) as e: raise BadRequest(f'{key} must be a number') from e if not math.isfinite(number): raise BadRequest(f'{key} must be finite') self._check_range(key, number, min_value, max_value) return number def _check_range(self, key, number, min_value, max_value): if min_value is not None and number < min_value: raise BadRequest(f'{key} must be >= {min_value}') if max_value is not None and number > max_value: raise BadRequest(f'{key} must be <= {max_value}') def _normalize_degree(self, body, key, default): return self._get_float(body, key, default, 0, 360) % 360 class VedastroEvidenceMixin: """VedAstro runtime-evidence summaries used by the consultation boundary.""" def _high_rigor_vedastro_official_summary(self, chart): prompt_pack = chart.get('ai_prompt_pack') if isinstance(chart, dict) else {} evidence_snapshot = prompt_pack.get('evidence_snapshot') if isinstance(prompt_pack, dict) else {} prompt_official = evidence_snapshot.get('vedastro_official_snapshot') if isinstance(evidence_snapshot, dict) else {} if not isinstance(prompt_official, dict): prompt_official = {} prompt_full_snapshot = evidence_snapshot.get('vedastro_official_full_snapshot') if isinstance(evidence_snapshot, dict) else {} if not isinstance(prompt_full_snapshot, dict): prompt_full_snapshot = {} modules = chart.get('modules') if isinstance(chart, dict) else {} if not isinstance(modules, dict): modules = {} range_scan = modules.get('vedastro_range_scan_result') if isinstance(modules, dict) else {} if not isinstance(range_scan, dict): range_scan = {} full_snapshot_payload = _build_vedastro_official_full_snapshot_payload_from_chart(chart) official_snapshot = range_scan.get('official_full_snapshot') if isinstance(range_scan, dict) else {} if not isinstance(official_snapshot, dict): official_snapshot = {} if not official_snapshot and isinstance(modules.get('vedastro_official_full_snapshot'), dict): official_snapshot = modules.get('vedastro_official_full_snapshot') or {} metadata = official_snapshot.get('source_metadata') if isinstance(official_snapshot, dict) else {} catalog = metadata.get('official_full_capability_catalog') if isinstance(metadata, dict) else {} if not isinstance(catalog, dict): catalog = {} range_metadata = range_scan.get('source_metadata') if isinstance(range_scan, dict) else {} if not isinstance(range_metadata, dict): range_metadata = {} strict_workflow_contracts = ( prompt_full_snapshot.get('strict_workflow_contracts') or full_snapshot_payload.get('strict_workflow_contracts') or {} ) if not isinstance(strict_workflow_contracts, dict): strict_workflow_contracts = {} strict_workflow_primary_route = ( prompt_full_snapshot.get('strict_workflow_primary_route') or full_snapshot_payload.get('strict_workflow_primary_route') ) strict_workflow_routes_available = ( prompt_full_snapshot.get('strict_workflow_routes_available') or full_snapshot_payload.get('strict_workflow_routes_available') or list(strict_workflow_contracts.keys()) ) if not isinstance(strict_workflow_routes_available, list): strict_workflow_routes_available = list(strict_workflow_contracts.keys()) _selected_route, primary_contract = _preferred_strict_contract( strict_workflow_contracts, strict_workflow_primary_route, ) dynamic_selection = ( prompt_official.get('official_full_capability_dynamic_selection') or catalog.get('dynamic_selection') or prompt_full_snapshot.get('official_full_capability_dynamic_selection') or full_snapshot_payload.get('official_full_capability_dynamic_selection') or range_metadata.get('official_full_capability_dynamic_selection') or {} ) report_references = ( prompt_official.get('official_report_references') or prompt_full_snapshot.get('official_report_references') or full_snapshot_payload.get('official_report_references') or range_metadata.get('official_report_references') or { theme: selection.get('report_reference') for theme, selection in dynamic_selection.items() if isinstance(selection, dict) and isinstance(selection.get('report_reference'), dict) } ) status = ( prompt_official.get('status') or official_snapshot.get('status') or range_scan.get('status') or 'blocked' ) chart_core_status = 'blocked' official_primary_evidence = ( primary_contract.get('official_primary_evidence') or prompt_official.get('official_primary_evidence') or {} ) if not isinstance(official_primary_evidence, dict): official_primary_evidence = {} chart_core = official_primary_evidence.get('chart_core') if isinstance(chart_core, dict) and chart_core.get('status'): chart_core_status = chart_core.get('status') elif full_snapshot_payload.get('available'): chart_core_status = 'ok' event_radar_status = 'blocked' if ( prompt_official.get('blocked_items') or prompt_official.get('fallback_used') or prompt_official.get('conflicts') ): event_radar_status = 'partial' elif range_scan.get('status') == 'ok': event_radar_status = 'ok' elif range_scan.get('status'): event_radar_status = 'partial' runtime_truth = { 'status': status, 'catalog_boundary': 'catalog_recognized_not_full_runtime_execution', 'primary_route': strict_workflow_primary_route or _selected_route, 'routes_available': strict_workflow_routes_available, 'official_execution_layers': { 'chart_core': chart_core_status, 'event_radar': event_radar_status, 'catalog_status': ( prompt_official.get('official_full_capability_catalog_status') or catalog.get('status') or range_metadata.get('official_full_capability_catalog_status') or official_snapshot.get('status') or 'blocked' ), }, 'fallback_active': bool( primary_contract.get('fallback_used') or prompt_official.get('fallback_used') ), 'blocked_items': ( primary_contract.get('blocked_items') or prompt_official.get('blocked_items') or [] ), 'conflicts': ( primary_contract.get('conflicts') or prompt_official.get('conflicts') or [] ), 'free_tier_strategy': { 'using_free_tier': not bool(os.environ.get('VEDASTRO_API_KEY', '').strip()), 'queue_enabled': _free_tier_queue_enabled_env(), 'cache_hit': bool( (((official_snapshot.get('source_metadata') or {}).get('semantic_cache') or {}).get('cache_hit')) if isinstance(official_snapshot, dict) else False ), 'guard_status': ( 'degraded_or_partial' if status in {'partial', 'blocked', 'official_snapshot_budget_exhausted'} or bool(prompt_official.get('blocked_items')) else 'within_free_tier_strategy' ), }, } raw_response = ( official_snapshot.get('raw_response') or official_snapshot.get('official_raw_response') or official_snapshot.get('raw_payload') or official_snapshot.get('raw') or prompt_full_snapshot.get('raw_response') or prompt_full_snapshot.get('official_raw_response') or prompt_full_snapshot.get('raw_payload') or prompt_full_snapshot.get('raw') ) return { 'status': status, 'range_scan_status': range_scan.get('status') if isinstance(range_scan, dict) else None, 'event_count': int(range_scan.get('event_count', 0) or 0) if isinstance(range_scan, dict) else 0, 'official_full_capability_catalog_status': ( prompt_official.get('official_full_capability_catalog_status') or catalog.get('status') or range_metadata.get('official_full_capability_catalog_status') ), 'official_full_capability_catalog_summary': ( prompt_official.get('official_full_capability_catalog_summary') or prompt_full_snapshot.get('official_full_capability_catalog_summary') or full_snapshot_payload.get('official_full_capability_catalog_summary') or catalog.get('summary') or range_metadata.get('official_full_capability_catalog_summary') or {} ), 'official_full_capability_domain_routing': ( prompt_official.get('official_full_capability_domain_routing') or prompt_full_snapshot.get('official_full_capability_domain_routing') or full_snapshot_payload.get('official_full_capability_domain_routing') or catalog.get('domain_routing') or range_metadata.get('official_full_capability_domain_routing') or {} ), 'official_full_capability_dynamic_selection': dynamic_selection, 'official_report_references': report_references, 'strict_workflow_primary_route': strict_workflow_primary_route, 'strict_workflow_routes_available': strict_workflow_routes_available, 'strict_workflow_contracts': strict_workflow_contracts, 'official_primary_evidence': ( primary_contract.get('official_primary_evidence') or prompt_official.get('official_primary_evidence') or {} ), 'local_supplemental_evidence': ( primary_contract.get('local_supplemental_evidence') or prompt_official.get('local_supplemental_evidence') or {} ), 'fallback_used': ( primary_contract.get('fallback_used') or prompt_official.get('fallback_used') or [] ), 'blocked_items': ( primary_contract.get('blocked_items') or prompt_official.get('blocked_items') or [] ), 'conflicts': ( primary_contract.get('conflicts') or prompt_official.get('conflicts') or [] ), 'technique_audit_summary': primary_contract.get('technique_audit_summary') or {}, 'adjudication_stages': primary_contract.get('adjudication_stages') or {}, 'multi_reference_reading_summary': primary_contract.get('multi_reference_reading_summary') or {}, 'verdict': primary_contract.get('verdict'), 'dominant_label': primary_contract.get('dominant_label'), 'main_conflicts': primary_contract.get('main_conflicts') or primary_contract.get('conflicts') or [], 'runtime_truth': runtime_truth, 'raw_response': raw_response, 'boundary': 'VedAstro official snapshot and capability catalog are consumed as primary evidence metadata; execution breadth depends on configured network and sample limits.', } def _compute_vedastro_gateway_archives(self): from scripts.vedastro_gateway import list_official_raw_response_archives return list_official_raw_response_archives() def _interpretation_source_runtime_coverage(self, chart): modules = chart.get('modules') if isinstance(chart, dict) else {} if not isinstance(modules, dict): modules = {} prompt_pack = chart.get('ai_prompt_pack') if isinstance(chart, dict) else {} evidence_snapshot = prompt_pack.get('evidence_snapshot') if isinstance(prompt_pack, dict) else {} interpretation_pack = evidence_snapshot.get('interpretation_source_pack') if isinstance(evidence_snapshot.get('interpretation_source_pack'), dict) else {} candidates = { 'dasha_timing_layer_used', 'varga_strength_layer_used', 'annual_special_layer_context', 'modifier_obstacle_layer_used', } proven_markers = [] guided_topics = modules.get('guided_topics') if isinstance(modules.get('guided_topics'), list) else [] for topic in guided_topics: if not isinstance(topic, dict): continue strict_gate = topic.get('strict_audit_gate') if not isinstance(strict_gate, dict): continue secondary = strict_gate.get('secondary_context') if not isinstance(secondary, list): continue for item in secondary: if isinstance(item, str) and item in candidates and item not in proven_markers: proven_markers.append(item) return { 'source_pack_status': interpretation_pack.get('status') or 'used', 'proven_runtime_markers': proven_markers, 'runtime_visibility_status': 'partial' if proven_markers else 'blocked', 'not_fully_closed': [ 'references/open_source_sources/jyotishganit', 'references/open_source_sources/jaimini-tropical', 'references/open_source_sources/VedicAstro', 'references/open_source_sources/rishi-ai-mcp', 'references/open_source_sources/vedic-astro-skills', 'references/open_source_sources/dashaflow', ], 'boundary': 'Inventory/grading exists, but full runtime invocation is only proven for surfaced strict-workflow markers, not every local source asset.', } class SynastryMixin(RequestParamMixin): """Ashtakoot synastry scoring.""" def _compute_synastry(self, body): from ashtakoot import calculate_ashtakoot result = calculate_ashtakoot( self._normalize_degree(body, 'male_moon', 0), self._normalize_degree(body, 'female_moon', 0), ) # Backward-compatible aliases for older frontend/report consumers. result['is_approved'] = result.get('is_match_approved', False) result['assessment'] = ( '优秀' if result.get('total_score', 0) >= 28 else '良好' if result.get('total_score', 0) >= 21 else '一般' if result.get('total_score', 0) >= 18 else '不推荐' ) result['male'] = result.get('male_details', {}) result['female'] = result.get('female_details', {}) return result