From 09817b96fdf120dd5ee63359cf966cd0204a3dfb Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Sun, 28 Jun 2026 18:18:33 +0800 Subject: [PATCH] Bridge functional benefic malefic layer --- ...benefic_malefic_bridge_audit_2026_06_28.md | 42 +++++++ mcp_server.py | 80 ++----------- scripts/functional_benefics.py | 112 ++++++++++++++++++ scripts/jyotish_api_server.py | 63 +++++++--- scripts/jyotish_engine.py | 22 +--- scripts/oracle_functional_benefics.py | 52 ++++++++ tests/test_api_server_security.py | 25 ++++ ...st_mcp_strict_workflow_functional_layer.py | 26 ++++ 8 files changed, 317 insertions(+), 105 deletions(-) create mode 100644 docs/research/functional_benefic_malefic_bridge_audit_2026_06_28.md create mode 100644 scripts/functional_benefics.py create mode 100644 scripts/oracle_functional_benefics.py diff --git a/docs/research/functional_benefic_malefic_bridge_audit_2026_06_28.md b/docs/research/functional_benefic_malefic_bridge_audit_2026_06_28.md new file mode 100644 index 00000000..18f509eb --- /dev/null +++ b/docs/research/functional_benefic_malefic_bridge_audit_2026_06_28.md @@ -0,0 +1,42 @@ +# Functional Benefic/Malefic Bridge Audit - 2026-06-28 + +## Scope + +This pass promotes the previously untracked `scripts/oracle_functional_benefics.py` +fragment into a reusable strict-workflow layer. + +## What Changed + +- Added `/Users/wuyongnaren/Documents/印度占星/scripts/functional_benefics.py` as the single source for functional benefic/malefic classification by Lagna. +- Converted `/Users/wuyongnaren/Documents/印度占星/scripts/oracle_functional_benefics.py` into a CLI wrapper around that module. +- Routed MCP strict workflows through the shared module. +- Routed full-reading prompt-pack and API prompt-pack snapshots through the same shared module. + +## Contract + +The bridge requires only a valid ascendant sign. + +It returns: + +- `status` +- `ascendant` +- `functional_benefics` +- `functional_malefics` +- `functional_neutrals` +- `yogakarakas` +- `owned_houses` +- `effect_on_confidence` +- `source` + +## Boundary + +- This layer classifies functional house-lord roles. +- It does not replace natural benefic/malefic assessment. +- It does not directly force event labels. +- It must appear in strict workflow evidence and Technique Audit Table outputs for high-rigor readings. + +## Regression Coverage + +- `/Users/wuyongnaren/Documents/印度占星/tests/test_mcp_strict_workflow_functional_layer.py` +- `/Users/wuyongnaren/Documents/印度占星/tests/test_cli_smoke.py` +- `/Users/wuyongnaren/Documents/印度占星/tests/test_api_server_security.py::test_chart_ai_prompt_pack_exposes_functional_benefic_malefic_layer` diff --git a/mcp_server.py b/mcp_server.py index ef879784..d7f7f303 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -34,6 +34,7 @@ SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(SCRIPT_DIR, "scripts")) from mcp.server.fastmcp import FastMCP +from functional_benefics import derive_functional_benefic_malefic # ============================================================================ # MCP Server @@ -452,86 +453,19 @@ def _derive_external_activation_support(modules: Dict[str, Any], domain: str) -> def _derive_functional_benefic_malefic(modules: Dict[str, Any]) -> Dict[str, Any]: chart = _safe_get(modules, "chart") ascendant = _safe_get(chart, "ascendant") if isinstance(chart, dict) else None - planets = _safe_get(chart, "planets") if isinstance(chart, dict) else None - if not isinstance(ascendant, dict) or not isinstance(planets, dict): + if not isinstance(ascendant, dict): return { "status": "blocked", "ascendant": ascendant.get("sign") if isinstance(ascendant, dict) else None, "functional_benefics": [], "functional_malefics": [], - "effect_on_confidence": "Missing chart.ascendant or chart.planets; functional layer blocked.", + "functional_neutrals": [], + "yogakarakas": [], + "owned_houses": {}, + "effect_on_confidence": "Missing chart.ascendant; functional layer blocked.", "source": "strict_functional_benefic_malefic_v1", } - asc_sign = ascendant.get("sign") - asc_idx = _SIGN_TO_INDEX.get(asc_sign) - if asc_idx is None: - return { - "status": "blocked", - "ascendant": asc_sign, - "functional_benefics": [], - "functional_malefics": [], - "effect_on_confidence": "Functional layer blocked: unknown ascendant sign.", - "source": "strict_functional_benefic_malefic_v1", - } - - owned_houses: Dict[str, List[int]] = {} - for house_num in range(1, 13): - sign = _SIGNS[(asc_idx + house_num - 1) % 12] - lord = _SIGN_LORDS.get(sign) - if lord: - owned_houses.setdefault(lord, []).append(house_num) - - trines = {1, 5, 9} - kendras = {1, 4, 7, 10} - challenging = {3, 6, 8, 11, 12} - benefics: set[str] = set() - malefics: set[str] = set() - yogakarakas: set[str] = set() - neutrals: set[str] = set() - - for planet in ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"): - houses = owned_houses.get(planet, []) - if not houses: - continue - owns_trine = any(house in trines for house in houses) - owns_kendra = any(house in kendras for house in houses) - owns_challenge = any(house in challenging for house in houses) - - if owns_trine and owns_kendra and planet not in {"Sun", "Moon"}: - yogakarakas.add(planet) - benefics.add(planet) - elif owns_trine: - benefics.add(planet) - elif owns_challenge and 1 not in houses: - malefics.add(planet) - elif owns_kendra and planet in {"Jupiter", "Venus", "Mercury", "Moon"}: - neutrals.add(planet) - else: - neutrals.add(planet) - - # Classical softening: Sun/Moon as 8L are not treated as harshly as other 8L. - eighth_lord = owned_houses and next( - (planet for planet, houses in owned_houses.items() if 8 in houses), - None, - ) - if eighth_lord in {"Sun", "Moon"} and eighth_lord in malefics: - malefics.remove(eighth_lord) - neutrals.add(eighth_lord) - - return { - "status": "used", - "ascendant": asc_sign, - "functional_benefics": sorted(benefics), - "functional_malefics": sorted(malefics), - "functional_neutrals": sorted(neutrals - benefics - malefics), - "yogakarakas": sorted(yogakarakas), - "owned_houses": {planet: houses for planet, houses in sorted(owned_houses.items())}, - "effect_on_confidence": ( - "High-rigor outputs must combine functional house-lord roles with natural roles; " - "conflicts should cap confidence or be explicitly noted." - ), - "source": "strict_functional_benefic_malefic_v1", - } + return derive_functional_benefic_malefic(ascendant.get("sign")) def _derive_synastry_relationship_support(modules: Dict[str, Any]) -> Dict[str, Any]: diff --git a/scripts/functional_benefics.py b/scripts/functional_benefics.py new file mode 100644 index 00000000..eaf973e6 --- /dev/null +++ b/scripts/functional_benefics.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Functional benefic/malefic classification by ascendant.""" + +from __future__ import annotations + +from typing import Any + + +SIGNS = [ + "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", + "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces", +] +SIGN_TO_INDEX = {name: idx for idx, name in enumerate(SIGNS)} +SIGN_LORDS = { + "Aries": "Mars", + "Taurus": "Venus", + "Gemini": "Mercury", + "Cancer": "Moon", + "Leo": "Sun", + "Virgo": "Mercury", + "Libra": "Venus", + "Scorpio": "Mars", + "Sagittarius": "Jupiter", + "Capricorn": "Saturn", + "Aquarius": "Saturn", + "Pisces": "Jupiter", +} +PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn") +SOURCE = "strict_functional_benefic_malefic_v1" + + +def normalize_sign(value: Any) -> str | None: + if not isinstance(value, str): + return None + lowered = value.strip().lower() + for sign in SIGNS: + if sign.lower() == lowered: + return sign + return None + + +def derive_functional_benefic_malefic(ascendant: Any) -> dict[str, Any]: + """Return functional benefics, malefics and ownership roles for a Lagna.""" + asc_sign = normalize_sign(ascendant) + if asc_sign is None: + return { + "status": "blocked", + "ascendant": ascendant if isinstance(ascendant, str) else None, + "functional_benefics": [], + "functional_malefics": [], + "functional_neutrals": [], + "yogakarakas": [], + "owned_houses": {}, + "effect_on_confidence": "Functional layer blocked: unknown ascendant sign.", + "source": SOURCE, + } + + asc_idx = SIGN_TO_INDEX[asc_sign] + owned_houses: dict[str, list[int]] = {} + for house_num in range(1, 13): + sign = SIGNS[(asc_idx + house_num - 1) % 12] + lord = SIGN_LORDS.get(sign) + if lord: + owned_houses.setdefault(lord, []).append(house_num) + + trines = {1, 5, 9} + kendras = {1, 4, 7, 10} + challenging = {3, 6, 8, 11, 12} + benefics: set[str] = set() + malefics: set[str] = set() + yogakarakas: set[str] = set() + neutrals: set[str] = set() + + for planet in PLANETS: + houses = owned_houses.get(planet, []) + if not houses: + continue + owns_trine = any(house in trines for house in houses) + owns_kendra = any(house in kendras for house in houses) + owns_challenge = any(house in challenging for house in houses) + + if owns_trine and owns_kendra and planet not in {"Sun", "Moon"}: + yogakarakas.add(planet) + benefics.add(planet) + elif owns_trine: + benefics.add(planet) + elif owns_challenge and 1 not in houses: + malefics.add(planet) + elif owns_kendra and planet in {"Jupiter", "Venus", "Mercury", "Moon"}: + neutrals.add(planet) + else: + neutrals.add(planet) + + eighth_lord = next((planet for planet, houses in owned_houses.items() if 8 in houses), None) + if eighth_lord in {"Sun", "Moon"} and eighth_lord in malefics: + malefics.remove(eighth_lord) + neutrals.add(eighth_lord) + + return { + "status": "used", + "ascendant": asc_sign, + "functional_benefics": sorted(benefics), + "functional_malefics": sorted(malefics), + "functional_neutrals": sorted(neutrals - benefics - malefics), + "yogakarakas": sorted(yogakarakas), + "owned_houses": {planet: houses for planet, houses in sorted(owned_houses.items())}, + "effect_on_confidence": ( + "高严谨模式下必须叠加功能性宫主吉凶与自然吉凶；" + "若功能属性与自然属性冲突，应降低置信度或显式标记冲突。" + ), + "source": SOURCE, + } diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index b01700f6..c42ed5a1 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -9,6 +9,7 @@ import argparse import base64 +import html as html_lib import io import json, sys, os, math import importlib.util @@ -441,6 +442,42 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): raise BadRequest('report html cannot include active content') return html + def _inject_functional_benefic_malefic_summary(self, html, snapshot): + if not isinstance(snapshot, dict): + return html + if snapshot.get('status') in {None, 'blocked', 'not_used'}: + return html + benefics = snapshot.get('functional_benefics') + malefics = snapshot.get('functional_malefics') + if not isinstance(benefics, list) or not isinstance(malefics, list): + return html + + def _escape(value): + return html_lib.escape(str(value or '')) + + ascendant = _escape(snapshot.get('ascendant') or snapshot.get('asc_sign') or 'Unknown') + benefic_text = _escape(', '.join(str(item) for item in benefics) or 'None') + malefic_text = _escape(', '.join(str(item) for item in malefics) or 'None') + confidence_text = _escape(snapshot.get('effect_on_confidence') or 'Functional role layer was used in the final judgement.') + source_text = _escape(snapshot.get('source') or 'strict_functional_benefic_malefic_v1') + + summary = ( + '
' + '

Functional Benefic/Malefic

' + f'

Ascendant: {ascendant}

' + f'

Functional Benefics: {benefic_text}

' + f'

Functional Malefics: {malefic_text}

' + f'

Confidence Impact: {confidence_text}

' + f'

Source: {source_text}

' + '
' + ) + body_close = re.search(r'', html, re.IGNORECASE) + if body_close: + return html[:body_close.start()] + summary + html[body_close.start():] + return html + summary + def _artifact_base64(self, path): size = os.path.getsize(path) if size > MAX_REPORT_BASE64_BYTES: @@ -505,6 +542,10 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): def _compute_report_artifact(self, body): html = self._validate_report_html(body.get('html')) + html = self._inject_functional_benefic_malefic_summary( + html, + body.get('functional_benefic_malefic'), + ) fmt = body.get('format', 'html') if fmt not in {'html', 'pdf'}: raise BadRequest('format must be html or pdf') @@ -1921,30 +1962,20 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): def _functional_benefic_malefic_snapshot(self, planets, ascendant): try: - from yoga_engine import YogaContext + from functional_benefics import derive_functional_benefic_malefic asc_sign = ascendant.get('sign') - if not asc_sign or not isinstance(planets, dict): - raise ValueError('missing ascendant sign or planets') - context = YogaContext(planets, asc_sign) - benefics = context.functional_benefics() - malefics = context.functional_malefics() - return { - 'status': 'used', - 'ascendant': asc_sign, - 'functional_benefics': benefics, - 'functional_malefics': malefics, - 'effect_on_confidence': ( - '高严谨模式下必须叠加功能性吉凶星；若与自然吉凶属性冲突，' - '应降低置信度并在 Technique Audit Table 中显式说明。' - ), - } + return derive_functional_benefic_malefic(asc_sign) except Exception as exc: return { 'status': 'blocked', 'ascendant': ascendant.get('sign'), 'functional_benefics': [], 'functional_malefics': [], + 'functional_neutrals': [], + 'yogakarakas': [], + 'owned_houses': {}, 'effect_on_confidence': f'未完成功能性吉凶星判定，需降低高严谨结论置信度: {exc}', + 'source': 'strict_functional_benefic_malefic_v1', } def _detect_yogas(self, planets, asc_idx): diff --git a/scripts/jyotish_engine.py b/scripts/jyotish_engine.py index f11488ac..6e553942 100644 --- a/scripts/jyotish_engine.py +++ b/scripts/jyotish_engine.py @@ -859,30 +859,20 @@ def _oracle_progress_snapshot(): def _functional_benefic_malefic_snapshot(planets, ascendant): try: - from yoga_engine import YogaContext + from functional_benefics import derive_functional_benefic_malefic asc_sign = ascendant.get('sign') if isinstance(ascendant, dict) else None - if not asc_sign or not isinstance(planets, dict): - raise ValueError('missing ascendant sign or planets') - context = YogaContext(planets, asc_sign) - benefics = context.functional_benefics() - malefics = context.functional_malefics() - return { - 'status': 'used', - 'ascendant': asc_sign, - 'functional_benefics': benefics, - 'functional_malefics': malefics, - 'effect_on_confidence': ( - '高严谨模式下必须叠加功能性吉凶星；若与自然吉凶属性冲突，' - '应降低置信度并在 Technique Audit Table 中显式说明。' - ), - } + return derive_functional_benefic_malefic(asc_sign) except Exception as exc: return { 'status': 'blocked', 'ascendant': ascendant.get('sign') if isinstance(ascendant, dict) else None, 'functional_benefics': [], 'functional_malefics': [], + 'functional_neutrals': [], + 'yogakarakas': [], + 'owned_houses': {}, 'effect_on_confidence': f'未完成功能性吉凶星判定，需降低高严谨结论置信度: {exc}', + 'source': 'strict_functional_benefic_malefic_v1', } diff --git a/scripts/oracle_functional_benefics.py b/scripts/oracle_functional_benefics.py new file mode 100644 index 00000000..7e80ca03 --- /dev/null +++ b/scripts/oracle_functional_benefics.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""CLI wrapper for functional benefic/malefic classification.""" + +from __future__ import annotations + +import argparse +import json +from typing import Any + +from functional_benefics import PLANETS, derive_functional_benefic_malefic + + +def _render_text(report: dict[str, Any]) -> str: + lines = [ + "=== Oracle stdout ===", + f"Ascendant: {report.get('ascendant')}", + "--------------------------------------------------", + ] + owned_houses = report.get("owned_houses", {}) + benefics = set(report.get("functional_benefics", [])) + malefics = set(report.get("functional_malefics", [])) + yogakarakas = set(report.get("yogakarakas", [])) + for planet in PLANETS: + owned = owned_houses.get(planet, []) + houses = " & ".join(map(str, owned)) if owned else "-" + if planet in yogakarakas: + lines.append(f"{planet}: Yogakaraka (Lord of {houses}) -> Highly Auspicious") + elif planet in benefics: + lines.append(f"{planet}: Functional Benefic (Lord of {houses}) -> Auspicious") + elif planet in malefics: + lines.append(f"{planet}: Functional Malefic (Lord of {houses}) -> Destructive / Obstacle") + else: + lines.append(f"{planet}: Neutral / Mixed (Lord of {houses}) -> Depends on placement") + return "\n".join(lines) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Oracle CLI: Functional Benefic/Malefic Determiner") + parser.add_argument("--ascendant", required=True, type=str, help="The ascendant sign, e.g. Leo") + parser.add_argument("--format", choices=["text", "json"], default="text") + args = parser.parse_args() + + report = derive_functional_benefic_malefic(args.ascendant) + if args.format == "json": + print(json.dumps(report, ensure_ascii=False, indent=2)) + else: + print(_render_text(report)) + return 0 if report.get("status") == "used" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_api_server_security.py b/tests/test_api_server_security.py index 12a87331..376a1396 100644 --- a/tests/test_api_server_security.py +++ b/tests/test_api_server_security.py @@ -8,6 +8,7 @@ import json import os import sys from io import BytesIO +from pathlib import Path import pytest @@ -433,6 +434,30 @@ def test_report_artifact_generates_html_fallback_artifact() -> None: assert result['delivery']['next_action'] +def test_report_artifact_can_render_functional_benefic_malefic_summary() -> None: + handler = _handler() + result = handler._compute_report_artifact({ + 'format': 'html', + 'name': 'functional-role-report', + 'html': '

Jyotish

', + 'functional_benefic_malefic': { + 'status': 'used', + 'ascendant': 'Leo', + 'functional_benefics': ['Sun', 'Mars', 'Jupiter'], + 'functional_malefics': ['Venus', 'Saturn'], + 'effect_on_confidence': '高严谨模式下必须叠加功能性吉凶星。', + }, + }) + + assert result['success'] is True + html = Path(result['html_path']).read_text(encoding='utf-8') + assert 'Functional Benefic/Malefic' in html + assert 'Leo' in html + assert 'Sun, Mars, Jupiter' in html + assert 'Venus, Saturn' in html + assert '高严谨模式下必须叠加功能性吉凶星。' in html + + def test_report_artifact_pdf_fallback_exposes_user_visible_delivery(monkeypatch) -> None: class BrokenReportBuilder: @staticmethod diff --git a/tests/test_mcp_strict_workflow_functional_layer.py b/tests/test_mcp_strict_workflow_functional_layer.py index e08472e6..04c52398 100644 --- a/tests/test_mcp_strict_workflow_functional_layer.py +++ b/tests/test_mcp_strict_workflow_functional_layer.py @@ -4,6 +4,7 @@ from __future__ import annotations from mcp_server import _collect_strict_evidence +from scripts.functional_benefics import derive_functional_benefic_malefic def _base_relationship_result() -> dict: @@ -48,3 +49,28 @@ def test_relationship_strict_workflow_exposes_functional_benefic_malefic_layer() assert "Venus" in functional["functional_malefics"] assert "functional_benefic_malefic_used" in strict["event_judgement"]["secondary_context"] + +def test_functional_benefic_module_calculates_lagna_roles_without_planets() -> None: + functional = derive_functional_benefic_malefic("Leo") + + assert functional["status"] == "used" + assert functional["ascendant"] == "Leo" + assert functional["owned_houses"]["Sun"] == [1] + assert functional["owned_houses"]["Venus"] == [3, 10] + assert "Sun" in functional["functional_benefics"] + assert "Venus" in functional["functional_malefics"] + assert functional["source"] == "strict_functional_benefic_malefic_v1" + + +def test_relationship_functional_layer_uses_ascendant_even_when_planets_missing() -> None: + result = _base_relationship_result() + result["modules"]["chart"] = {"ascendant": {"sign": "Leo"}} + + strict = _collect_strict_evidence("relationship", result) + + functional = strict["present_evidence"]["functional_benefic_malefic"] + assert functional["status"] == "used" + assert functional["ascendant"] == "Leo" + assert "Sun" in functional["functional_benefics"] + assert "Venus" in functional["functional_malefics"] + assert "functional_benefic_malefic_used" in strict["event_judgement"]["secondary_context"]