root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: enforce public release evidence boundaries

Browse Source
This commit is contained in:
732642856
2026-07-08 18:34:13 +08:00
parent 9e46a4814a
commit 07fb5a5f7b
60 changed files with 1475 additions and 573 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/first_oracle_packet_assistant.py
+2 -2
View File
@@ -62,8 +62,8 @@ FRONTS = {
"--format",
"json",
],
"operator_card": "docs/benchmark/shadbala_redacted_place_raman_first_packet_operator_card.md",
"packet_template": "references/oracle/evidence_packet_templates/shadbala_redacted_place_raman_first_packet.json",
"operator_card": "docs/benchmark/shadbala_synthetic_north_china_raman_first_packet_operator_card.md",
"packet_template": "references/oracle/evidence_packet_templates/shadbala_synthetic_north_china_raman_first_packet.json",
"oracle_file": "references/oracle/dasha_shadbala_oracle_cases.json",
"apply_script": "scripts/oracle_collection_queue.py",
"external_sources": [
scripts/public_release_privacy_scan.py
+139
View File
@@ -0,0 +1,139 @@
#!/usr/bin/env python3
"""Scan release files for private birth-data residues."""
from __future__ import annotations
import argparse
import json
import re
import subprocess
from pathlib import Path
from typing import Iterable
ROOT = Path(__file__).resolve().parents[1]
SKIP_DIRS = {
".git",
"__pycache__",
".pytest_cache",
"node_modules",
"scratch",
"references/open_source_sources",
}
SKIP_FILES = {
"scripts/public_release_privacy_scan.py",
"tests/test_public_release_privacy_scan.py",
}
TEXT_SUFFIXES = {
".cfg",
".csv",
".html",
".js",
".json",
".md",
".py",
".txt",
".yaml",
".yml",
}
DENY_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
("private_exact_iso_birth_date", re.compile(r"REDACTED_DATE")),
("private_compact_birth_datetime", re.compile(r"REDACTED_DATE[_-]?REDACTED_TIME")),
("private_slug_birth_date", re.compile(r"REDACTED_YEAR[_-]04[_-]17")),
("private_birth_time_literal", re.compile(r"REDACTED_YEAR.{0,120}\bREDACTED_TIME\b|\bREDACTED_TIME\b.{0,120}REDACTED_YEAR|REDACTED_TIME")),
("private_place_han", re.compile(r"REDACTED_PLACE|REDACTED_PLACE|REDACTED_HOSPITAL")),
("private_case_slug", re.compile(r"user_REDACTED_YEAR|redacted_place", re.IGNORECASE)),
(
"private_birth_dict_tuple",
re.compile(
r"(?s)(?:year|--year|datetime\()\D*REDACTED_YEAR.{0,220}"
r"(?:month|--month|,\s*)\D*4.{0,220}"
r"(?:day|--day|,\s*)\D*17.{0,220}"
r"(?:hour|--hour|,\s*)\D*14.{0,220}"
r"(?:minute|--minute|,\s*)\D*49"
),
),
]
def _is_skipped(path: Path) -> bool:
rel = path.relative_to(ROOT).as_posix()
if rel in SKIP_FILES:
return True
return any(rel == item or rel.startswith(f"{item}/") for item in SKIP_DIRS)
def iter_release_files(root: Path = ROOT) -> Iterable[Path]:
completed = subprocess.run(
["git", "ls-files"],
cwd=root,
text=True,
capture_output=True,
check=True,
)
for line in completed.stdout.splitlines():
path = root / line
if not path.is_file() or _is_skipped(path):
continue
if path.suffix.lower() in TEXT_SUFFIXES:
yield path
def scan_text(path: Path, text: str) -> list[dict[str, object]]:
findings: list[dict[str, object]] = []
try:
display_path = path.relative_to(ROOT).as_posix()
except ValueError:
display_path = path.as_posix()
for rule_id, pattern in DENY_PATTERNS:
for match in pattern.finditer(text):
line = text.count("\n", 0, match.start()) + 1
findings.append(
{
"rule_id": rule_id,
"path": display_path,
"line": line,
}
)
return findings
def build_report(root: Path = ROOT) -> dict[str, object]:
findings: list[dict[str, object]] = []
scanned = 0
for path in iter_release_files(root):
try:
text = path.read_text(encoding="utf-8")
except UnicodeDecodeError:
text = path.read_text(encoding="utf-8", errors="ignore")
scanned += 1
findings.extend(scan_text(path, text))
return {
"scope": "public_release_privacy_scan",
"scanned_files": scanned,
"finding_count": len(findings),
"findings": findings,
"status": "pass" if not findings else "fail",
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--json", action="store_true", help="Emit JSON.")
args = parser.parse_args()
report = build_report()
if args.json:
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
else:
print(f"{report['status']}: {report['finding_count']} findings")
for finding in report["findings"]:
print(f"{finding['path']}:{finding['line']} {finding['rule_id']}")
return 0 if report["status"] == "pass" else 1
if __name__ == "__main__":
raise SystemExit(main())
scripts/shadbala_oracle_closure_status.py
+2 -2
View File
@@ -13,8 +13,8 @@ from typing import Any
ROOT = Path(__file__).resolve().parents[1]
PYTHON = sys.executable
FIRST_PRIORITY_CASE_ID = "template_redacted_place_shadbala_raman"
FIRST_PRIORITY_TEMPLATE_PATH = "references/oracle/evidence_packet_templates/shadbala_redacted_place_raman_first_packet.json"
FIRST_PRIORITY_CASE_ID = "template_synthetic_north_china_shadbala_raman"
FIRST_PRIORITY_TEMPLATE_PATH = "references/oracle/evidence_packet_templates/shadbala_synthetic_north_china_raman_first_packet.json"
SHADBALA_TARGET_FIELD = "target.shadbala_components"
SUPPORTING_TARGET_FIELDS = ["target.moon_sidereal_longitude_deg"]
REQUIRED_PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"]
scripts/sync_final_evidence_packet_status.py
+3 -3
View File
@@ -15,7 +15,7 @@ PACKET_RE = re.compile(r"\.v(\d+)\.json$")
def latest_packet() -> tuple[int, Path]:
packets: list[tuple[int, Path]] = []
for path in WORK_DIR.glob("jhora_master_evidence_packet_REDACTED_DATE_REDACTED_TIME.v*.json"):
for path in WORK_DIR.glob("jhora_master_evidence_packet_public_sample_19550224_1915.v*.json"):
match = PACKET_RE.search(path.name)
if match:
packets.append((int(match.group(1)), path))
@@ -45,11 +45,11 @@ def main() -> int:
if changed:
path.write_text(json.dumps(packet, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
ledger = WORK_DIR / "evidence_packet_status_ledger_REDACTED_DATE_REDACTED_TIME.md"
ledger = WORK_DIR / "evidence_packet_status_ledger_public_sample_19550224_1915.md"
if ledger.exists():
text = ledger.read_text(encoding="utf-8")
line_re = re.compile(
r"\| Master evidence packet \| `jhora_master_evidence_packet_REDACTED_DATE_REDACTED_TIME\.v\d+\.json` \| active \| Current canonical structured packet\. \|"
r"\| Master evidence packet \| `jhora_master_evidence_packet_public_sample_19550224_1915\.v\d+\.json` \| active \| Current canonical structured packet\. \|"
)
wanted_line = f"| Master evidence packet | `{path.name}` | active | Current canonical structured packet. |"
new_text = line_re.sub(wanted_line, text, count=1)
scripts/vedastro_official_contract_audit.py
+217
View File
@@ -0,0 +1,217 @@
#!/usr/bin/env python3
"""Audit the route-level VedAstro official evidence contract.
This is a provenance/contract gate, not a live VedAstro oracle. It protects
the user-facing reading layer from claiming official evidence when the official
raw response is blocked, partial, or only represented by local fallback data.
"""
from __future__ import annotations
import argparse
import json
from copy import deepcopy
from typing import Any
REQUIRED_CONTRACT_FIELDS = [
"source_priority_mode",
"official_primary_evidence",
"local_supplemental_evidence",
"fallback_used",
"blocked_items",
"conflicts",
"confidence_cap",
]
VALID_CONFIDENCE_CAPS = {"high", "medium", "low", "blocked"}
ROUTE_FIXTURES: list[dict[str, Any]] = [
{
"route": "relationship",
"contract": {
"source_priority_mode": "local_fallback_official_blocked",
"official_primary_evidence": {
"status": "blocked",
"required": True,
"raw_response_available": False,
"reason": "official_snapshot_budget_exhausted",
},
"local_supplemental_evidence": {
"status": "available",
"sections": ["D1", "D9", "UL", "Vimshottari", "Narayana"],
},
"fallback_used": ["local_jyotish_core"],
"blocked_items": ["vedastro_official_full_snapshot"],
"conflicts": [],
"confidence_cap": "low",
},
},
{
"route": "career",
"contract": {
"source_priority_mode": "vedastro_official_primary",
"official_primary_evidence": {
"status": "official_verified",
"required": True,
"raw_response_available": True,
"sections": ["chart_core", "dasha", "strength"],
},
"local_supplemental_evidence": {
"status": "available",
"sections": ["D10", "A10", "Shadbala", "Ashtakavarga"],
},
"fallback_used": [],
"blocked_items": [],
"conflicts": [],
"confidence_cap": "high",
},
},
{
"route": "wealth",
"contract": {
"source_priority_mode": "vedastro_official_primary_partial",
"official_primary_evidence": {
"status": "partial",
"required": True,
"raw_response_available": True,
"sections": ["chart_core"],
},
"local_supplemental_evidence": {
"status": "missing_required_sections",
"missing_sections": ["D2", "D11", "AV"],
},
"fallback_used": ["local_chart_core"],
"blocked_items": ["local_wealth_supplemental_bundle"],
"conflicts": [],
"confidence_cap": "low",
},
},
{
"route": "health",
"contract": {
"source_priority_mode": "local_fallback_official_blocked",
"official_primary_evidence": {
"status": "blocked",
"required": True,
"raw_response_available": False,
"reason": "endpoint_unconfigured",
},
"local_supplemental_evidence": {
"status": "missing_required_sections",
"missing_sections": ["D30", "medical_boundary_review"],
},
"fallback_used": [],
"blocked_items": ["vedastro_official_full_snapshot", "local_health_supplemental_bundle"],
"conflicts": [],
"confidence_cap": "blocked",
},
},
]
def _official_status(contract: dict[str, Any]) -> str:
evidence = contract.get("official_primary_evidence")
if not isinstance(evidence, dict):
return "missing"
return str(evidence.get("status") or "missing")
def _local_status(contract: dict[str, Any]) -> str:
evidence = contract.get("local_supplemental_evidence")
if not isinstance(evidence, dict):
return "missing"
return str(evidence.get("status") or "missing")
def expected_confidence_cap(contract: dict[str, Any]) -> str:
"""Return the strictest allowed confidence cap for this evidence state."""
official = _official_status(contract)
local = _local_status(contract)
conflicts = contract.get("conflicts")
fallback_used = contract.get("fallback_used")
unresolved_conflicts = bool(conflicts)
has_fallback = isinstance(fallback_used, list) and bool(fallback_used)
if official in {"blocked", "missing"} and local != "available" and not has_fallback:
return "blocked"
if unresolved_conflicts:
return "low"
if official in {"partial", "blocked", "missing"}:
return "low"
if local != "available":
return "low"
return "high"
def validate_contract(route: str, contract: dict[str, Any]) -> dict[str, Any]:
missing_fields = [field for field in REQUIRED_CONTRACT_FIELDS if field not in contract]
confidence_cap = contract.get("confidence_cap")
expected_cap = expected_confidence_cap(contract)
errors: list[str] = []
if missing_fields:
errors.append(f"missing_required_fields:{','.join(missing_fields)}")
if confidence_cap not in VALID_CONFIDENCE_CAPS:
errors.append(f"invalid_confidence_cap:{confidence_cap}")
if expected_cap == "blocked" and confidence_cap != "blocked":
errors.append("blocked_state_must_use_blocked_confidence_cap")
if expected_cap == "low" and confidence_cap not in {"low", "blocked"}:
errors.append("partial_or_fallback_state_must_not_exceed_low")
if expected_cap == "high" and confidence_cap not in {"high", "medium", "low", "blocked"}:
errors.append("verified_state_has_invalid_confidence_cap")
return {
"route": route,
"valid": not errors,
"missing_fields": missing_fields,
"official_status": _official_status(contract),
"local_status": _local_status(contract),
"fallback_count": len(contract.get("fallback_used") or []),
"blocked_count": len(contract.get("blocked_items") or []),
"conflict_count": len(contract.get("conflicts") or []),
"confidence_cap": confidence_cap,
"expected_confidence_cap": expected_cap,
"errors": errors,
}
def build_audit_report(routes: list[dict[str, Any]] | None = None) -> dict[str, Any]:
fixtures = deepcopy(routes if routes is not None else ROUTE_FIXTURES)
route_reports = [
validate_contract(str(item.get("route") or "unknown"), item.get("contract") or {})
for item in fixtures
]
invalid_routes = [item for item in route_reports if not item["valid"]]
return {
"scope": "vedastro_official_evidence_contract_audit",
"schema_version": 1,
"required_fields": REQUIRED_CONTRACT_FIELDS,
"route_count": len(route_reports),
"routes": route_reports,
"summary": {
"valid_routes": len(route_reports) - len(invalid_routes),
"invalid_routes": len(invalid_routes),
"confidence_cap_policy": "enforced",
},
"boundary": (
"Contract audit only; this does not prove VedAstro official raw "
"response availability, endpoint health, API key validity, or quota."
),
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--json", action="store_true", help="Emit JSON. Kept for CLI symmetry.")
args = parser.parse_args()
_ = args
print(json.dumps(build_audit_report(), ensure_ascii=False, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/vedastro_python_bridge.py
+5 -5
View File
@@ -487,11 +487,11 @@ def schema() -> dict[str, Any]:
},
"time": {
"__vedastro_type__": "Time",
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"year": 1955,
"month": 2,
"day": 24,
"hour": 19,
"minute": 15,
"offset": 8,
"geolocation": {
"__vedastro_type__": "GeoLocation",