Publish analysis results to platform for BYOK contribution

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()

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,19 +258,56 @@ 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"""
         st.subheader("🔍 Analysis Configuration")
 
         # 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))
 

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"