diff --git a/report_analyst_enterprise/contribution.py b/report_analyst_enterprise/contribution.py new file mode 100644 index 00000000..c9a6cccc --- /dev/null +++ b/report_analyst_enterprise/contribution.py @@ -0,0 +1,58 @@ +"""Publish analysis results to platform (BYOK contribution / enterprise). + +Enterprise-only: open-core callers import this optionally and no-op if absent. +""" + +from __future__ import annotations + +import json +import logging +import os +import uuid +from typing import Any, Dict, Optional + +import nats + +logger = logging.getLogger(__name__) + + +async def publish_analysis_result( + resource_id: str, + results: Dict[str, Any], + provenance: Optional[Dict[str, Any]] = None, + analysis_config: Optional[Dict[str, Any]] = None, + owner_user_id: Optional[str] = None, + duration_ms: Optional[int] = None, +) -> str: + """Publish analysis.result.{id} for platform to persist.""" + request_id = str(uuid.uuid4()) + nats_url = os.getenv("NATS_URL", "nats://localhost:4222") + nats_token = os.getenv("NATS_TOKEN") + if nats_token and "@" not in nats_url: + protocol, rest = nats_url.split("://", 1) + nats_url = f"{protocol}://{nats_token}@{rest}" + + payload = { + "request_id": request_id, + "resource_id": resource_id, + "results": results, + "results_summary": results, + "provenance": provenance or {}, + "analysis_config": analysis_config or {}, + "owner_user_id": owner_user_id, + "duration_ms": duration_ms, + "source": os.getenv("NATS_USER", "report-analyst"), + } + + nc = await nats.connect(nats_url, connect_timeout=15) + try: + js = nc.jetstream() + subject = f"analysis.result.{request_id}" + try: + await js.publish(subject, json.dumps(payload).encode()) + except Exception: # noqa: BLE001 — JetStream may be absent; fall back to core NATS + await nc.publish(subject, json.dumps(payload).encode()) + logger.info("Published %s for resource %s", subject, resource_id) + return request_id + finally: + await nc.close() diff --git a/report_analyst_search_backend/flow_orchestrator.py b/report_analyst_search_backend/flow_orchestrator.py index 389301fd..f7acf81a 100644 --- a/report_analyst_search_backend/flow_orchestrator.py +++ b/report_analyst_search_backend/flow_orchestrator.py @@ -88,7 +88,7 @@ def process_document(self, uploaded_file) -> ProcessingResult: return self._process_complete_backend(uploaded_file) else: return ProcessingResult(success=False, error=f"Unknown flow type: {flow_type}") - except Exception as e: + except Exception as e: # noqa: BLE001 logger.error(f"Document processing failed: {e}") return ProcessingResult(success=False, error=str(e)) @@ -114,7 +114,7 @@ def analyze_document(self, chunks: List[Dict[str, Any]], questions: List[str]) - return self._analyze_enhanced(chunks, questions) else: return AnalysisResult(success=False, error=f"Analysis not supported for flow: {flow_type}") - except Exception as e: + except Exception as e: # noqa: BLE001 logger.error(f"Document analysis failed: {e}") return AnalysisResult(success=False, error=str(e)) @@ -258,11 +258,48 @@ def _analyze_local_with_features(self, chunks: List[Dict[str, Any]], questions: return self._analyze_local(chunks, questions) def _analyze_enhanced(self, chunks: List[Dict[str, Any]], questions: List[str]) -> AnalysisResult: - """Enhanced analysis with centralized LLM and data lake""" - # This would use NATS LLM and store in data lake - # For now, fallback to local analysis + """Enhanced analysis with platform contribution publish when backend is enabled.""" + import asyncio + import os + st.info("Enhanced analysis not fully implemented - using local analysis") - return self._analyze_local(chunks, questions) + result = self._analyze_local(chunks, questions) + if not (result.success and self.config.use_backend and result.results): + return result + + try: + from report_analyst_enterprise.contribution import publish_analysis_result + + resource_id = chunks[0].get("resource_id") if chunks else None + if resource_id: + questions_list = result.results.get("questions", questions) + answers_list = result.results.get("answers", []) + asyncio.run( + publish_analysis_result( + resource_id=str(resource_id), + results={ + "answers": answers_list, + "questions": questions_list, + }, + provenance={ + "model": os.getenv("OPENAI_API_MODEL", "local"), + "provider": "report_analyst", + "mode": ( + "byok_contribution" + if os.getenv("USE_BYOK_CONTRIBUTION", "").lower() in ("1", "true", "yes") + else "centralized" + ), + }, + ) + ) + except ImportError: + # Enterprise package not installed — contribution publish is optional. + pass + except Exception as exc: # noqa: BLE001 — contribution publish must not fail analysis + logger.warning("Could not publish analysis result to platform: %s", exc) + + result.stored_in_backend = self.config.use_backend + return result def _configure_question_set(self, default_question_set: str) -> str: """Configure question set for backend analysis""" @@ -270,7 +307,7 @@ def _configure_question_set(self, default_question_set: str) -> str: # Get dynamic question set options if QUESTION_LOADER_AVAILABLE: - question_set_options = question_loader.get_question_set_options() + ["custom"] + question_set_options = [*question_loader.get_question_set_options(), "custom"] # Calculate index for default question set try: index = question_set_options.index(default_question_set) if default_question_set in question_set_options else 0 @@ -468,7 +505,7 @@ async def external_service_analysis( analysis_job_id=request_id, ) - except Exception as e: + except Exception as e: # noqa: BLE001 logger.error(f"External service analysis failed: {e}") return AnalysisResult(success=False, error=str(e)) diff --git a/tests/test_contribution.py b/tests/test_contribution.py new file mode 100644 index 00000000..93c59a6f --- /dev/null +++ b/tests/test_contribution.py @@ -0,0 +1,74 @@ +"""Unit tests for enterprise BYOK contribution publish (analysis.result NATS events).""" + +from __future__ import annotations + +import json +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from report_analyst_enterprise.contribution import publish_analysis_result + + +@pytest.mark.asyncio +async def test_publish_analysis_result_uses_analysis_result_subject_and_payload(monkeypatch): + mock_nc = MagicMock() + mock_js = AsyncMock() + mock_nc.jetstream.return_value = mock_js + mock_nc.close = AsyncMock() + + async def fake_connect(url, **kwargs): + return mock_nc + + monkeypatch.setenv("NATS_URL", "nats://localhost:4222") + monkeypatch.setenv("NATS_USER", "report-analyst-test") + monkeypatch.setattr("report_analyst_enterprise.contribution.nats.connect", fake_connect) + + request_id = await publish_analysis_result( + resource_id="res-123", + results={"answers": ["answer one"], "questions": ["question one"]}, + provenance={"mode": "byok_contribution", "provider": "report_analyst"}, + owner_user_id="user-42", + duration_ms=1500, + ) + + mock_js.publish.assert_awaited_once() + subject, payload_bytes = mock_js.publish.call_args[0] + assert subject == f"analysis.result.{request_id}" + payload = json.loads(payload_bytes.decode()) + assert payload["request_id"] == request_id + assert payload["resource_id"] == "res-123" + assert payload["results"] == {"answers": ["answer one"], "questions": ["question one"]} + assert payload["results_summary"] == payload["results"] + assert payload["provenance"]["mode"] == "byok_contribution" + assert payload["owner_user_id"] == "user-42" + assert payload["duration_ms"] == 1500 + assert payload["source"] == "report-analyst-test" + mock_nc.close.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_publish_analysis_result_falls_back_to_core_publish(monkeypatch): + mock_nc = MagicMock() + mock_js = AsyncMock() + mock_js.publish.side_effect = RuntimeError("jetstream unavailable") + mock_nc.jetstream.return_value = mock_js + mock_nc.close = AsyncMock() + mock_nc.publish = AsyncMock() + + async def fake_connect(url, **kwargs): + return mock_nc + + monkeypatch.setenv("NATS_URL", "nats://localhost:4222") + monkeypatch.setattr("report_analyst_enterprise.contribution.nats.connect", fake_connect) + + request_id = await publish_analysis_result( + resource_id="res-456", + results={"answers": ["a"], "questions": ["q"]}, + ) + + mock_nc.publish.assert_awaited_once() + subject, payload_bytes = mock_nc.publish.call_args[0] + assert subject == f"analysis.result.{request_id}" + payload = json.loads(payload_bytes.decode()) + assert payload["resource_id"] == "res-456"