Expose VedAstro gateway professional reading APIs

This commit is contained in:
732642856
2026-07-02 23:46:50 +08:00
parent dd1a792256
commit 329523fa7c
2 changed files with 164 additions and 0 deletions
+81
View File
@@ -801,6 +801,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
self._json(self._technique_catalog())
elif path == '/api/vedastro/status':
self._json(self._vedastro_status())
elif path == '/api/vedastro_gateway/status':
self._json(self._compute_vedastro_gateway_status())
elif path.startswith('/api/chart/jobs/'):
job_id = path.rsplit('/', 1)[-1]
result = self._get_chart_job(job_id)
@@ -864,6 +866,12 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
elif path == '/api/vedastro/range_scan':
result = self._compute_vedastro_range_scan(body)
self._json(result)
elif path == '/api/vedastro_gateway/run':
result = self._compute_vedastro_gateway_run(body)
self._json(result)
elif path == '/api/professional_reading':
result = self._compute_professional_reading(body)
self._json(result)
elif path == '/api/import_chart':
result = self._import_chart_text(body)
self._json(result)
@@ -1621,6 +1629,79 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
body_copy.pop('enqueue', None)
return self._compute_high_rigor_workflow(body_copy)
def _compute_vedastro_gateway_status(self):
from scripts.vedastro_gateway import gateway_status
return gateway_status()
def _compute_vedastro_gateway_run(self, body):
from scripts.vedastro_gateway import run_gateway_packet
payload = dict(body or {})
birth_payload = self._high_rigor_birth_payload(payload)
themes = self._high_rigor_requested_themes(payload)
reference_date = (
payload.get('reference_date')
or payload.get('transit_date')
or payload.get('today')
or payload.get('current_date')
or datetime.now().strftime('%Y-%m-%d')
)
return run_gateway_packet(
birth_payload,
question=str(payload.get('question') or payload.get('query') or ''),
themes=themes,
reference_date=str(reference_date),
)
def _compute_professional_reading(self, body):
payload = dict(body or {})
high_rigor_payload = {
**payload,
'surface': payload.get('surface') or 'professional_reading_web',
'return_high_rigor_shape': True,
}
high_rigor_payload.pop('async', None)
high_rigor_payload.pop('enqueue', None)
high_rigor = self._compute_high_rigor_workflow(high_rigor_payload)
gateway = self._compute_vedastro_gateway_run(payload)
return {
'success': True,
'endpoint': 'professional_reading',
'schema_version': 1,
'professional_reading': {
'input': {
'question': payload.get('question') or payload.get('query') or '',
'themes': self._high_rigor_requested_themes(payload),
'blind_mode': bool(payload.get('blind_mode')),
'reference_date': payload.get('reference_date')
or payload.get('transit_date')
or payload.get('today')
or payload.get('current_date'),
},
'high_rigor_workflow': high_rigor,
'vedastro_gateway': gateway,
'technique_audit_table_required_rows': [
'Functional Benefic/Malefic',
'MEVG / Global Web Evidence',
'Real Case Calibration',
'VedAstro Gateway Boundary',
],
'visibility_contract': {
'requires_technique_audit_table': True,
'requires_source_governance': True,
'requires_confidence_boundary': True,
'requires_user_visible_blocked_reasons': True,
},
'user_led_calibration_controls': {
'blind_mode': bool(payload.get('blind_mode')),
'disable_life_event_feedback': bool(payload.get('disable_life_event_feedback') or payload.get('blind_mode')),
'allow_user_event_selection': bool(payload.get('allow_user_event_selection', True)),
'note': 'User feedback must stay explicit and option-based; do not infer from prior chat memory in blind mode.',
},
},
}
def _high_rigor_workflow_plan_only(self, birth_payload, themes, events):
return {
'success': True,
+83
View File
@@ -261,6 +261,89 @@ def test_vedastro_range_scan_endpoint_uses_user_birth_and_returns_controlled_blo
assert payload['boundary'] == 'VedAstro range scan is optional external timing evidence; local Jyotish gates remain authoritative.'
def test_vedastro_gateway_status_route_is_cn_safe(monkeypatch) -> None:
monkeypatch.setenv('VEDASTRO_GATEWAY_MODE', 'cn_gateway')
handler = _handler()
result = handler._compute_vedastro_gateway_status()
assert result['scope'] == 'vedastro_gateway'
assert result['mode'] == 'cn_gateway'
assert result['direct_browser_access_allowed'] is False
assert result['frontend_secret_safe'] is True
def test_vedastro_gateway_run_route_returns_gateway_packet(monkeypatch) -> None:
monkeypatch.setenv('JYOTISH_SKIP_LOCAL_ENV', '1')
monkeypatch.setenv('VEDASTRO_GATEWAY_MODE', 'cn_gateway')
monkeypatch.setenv('VEDASTRO_CACHE_TTL_SECONDS', '604800')
monkeypatch.setenv('VEDASTRO_FULL_CATALOG_SAMPLE_LIMIT', '0')
handler = _handler()
result = handler._compute_vedastro_gateway_run({
'year': REDACTED_YEAR,
'month': 4,
'day': 17,
'hour': 14,
'minute': 49,
'second': 0,
'lat': 36.4467,
'lon': 114.2,
'tz': 8,
'question': '事业机会什么时候出现',
'themes': ['career', 'health'],
'reference_date': '2026-07-02',
})
assert result['scope'] == 'vedastro_gateway_run'
assert result['gateway_status']['mode'] == 'cn_gateway'
assert result['honesty_boundary']['all_641_methods_executed'] is False
assert result['user_visibility']['mainland_cn_safe'] is True
def test_professional_reading_composes_high_rigor_and_gateway(monkeypatch) -> None:
handler = _handler()
def fake_high_rigor(body):
return {
'success': True,
'endpoint': 'high_rigor_workflow',
'body': dict(body),
'technique_audit': [{'technique': 'MEVG / Global Web Evidence', 'status': 'queued'}],
}
def fake_gateway(body):
return {
'scope': 'vedastro_gateway_run',
'status': 'local_fallback',
'user_visibility': {'boundary': 'VedAstro Gateway Boundary'},
'honesty_boundary': {'all_641_methods_executed': False},
}
monkeypatch.setattr(handler, '_compute_high_rigor_workflow', fake_high_rigor)
monkeypatch.setattr(handler, '_compute_vedastro_gateway_run', fake_gateway)
result = handler._compute_professional_reading({
'year': REDACTED_YEAR,
'month': 4,
'day': 17,
'hour': 14,
'minute': 49,
'lat': 36.4467,
'lon': 114.2,
'tz': 8,
'question': '盲推事业',
'themes': ['career', 'health'],
'blind_mode': True,
})
assert result['endpoint'] == 'professional_reading'
assert result['professional_reading']['high_rigor_workflow']['endpoint'] == 'high_rigor_workflow'
assert result['professional_reading']['vedastro_gateway']['scope'] == 'vedastro_gateway_run'
assert result['professional_reading']['user_led_calibration_controls']['blind_mode'] is True
assert result['professional_reading']['visibility_contract']['requires_technique_audit_table'] is True
@pytest.mark.parametrize(
('key', 'value', 'minimum', 'maximum'),
[