diff --git a/report_analyst/consolidated_results_view.py b/report_analyst/consolidated_results_view.py new file mode 100644 index 00000000..68c2fd64 --- /dev/null +++ b/report_analyst/consolidated_results_view.py @@ -0,0 +1,246 @@ +"""All Results chunk search and consolidated report rendering.""" + +from __future__ import annotations + +import logging +from pathlib import Path + +import numpy as np +import pandas as pd +import streamlit as st + +logger = logging.getLogger(__name__) + + +def _chunk_text(chunk: dict) -> str: + return chunk.get("text", chunk.get("chunk_text", "")) + + +def _chunks_to_rows(chunks: list[dict], *, ranked: bool = False) -> list[dict]: + rows = [] + for i, chunk in enumerate(chunks): + row = { + "Text": _chunk_text(chunk), + "Has Embedding": chunk.get("embedding") is not None, + "Chunk Size": chunk.get("chunk_size", "N/A"), + "Chunk Overlap": chunk.get("chunk_overlap", "N/A"), + } + if ranked and "similarity" in chunk: + row["Rank"] = i + 1 + row["Similarity"] = chunk["similarity"] + else: + row["Chunk #"] = i + 1 + rows.append(row) + return rows + + +def _display_chunk_table(chunks_rows: list[dict], *, similarity_view: bool) -> None: + chunks_df = pd.DataFrame(chunks_rows) + if similarity_view and "Rank" in chunks_df.columns: + column_config = { + "Rank": st.column_config.NumberColumn("Rank", width="small"), + "Similarity": st.column_config.NumberColumn("Similarity", format="%.4f", width="small"), + "Text": st.column_config.TextColumn("Text", width="large"), + "Has Embedding": st.column_config.CheckboxColumn("Has Embedding"), + "Chunk Size": st.column_config.NumberColumn("Chunk Size", width="small"), + "Chunk Overlap": st.column_config.NumberColumn("Chunk Overlap", width="small"), + } + else: + column_config = { + "Chunk #": st.column_config.NumberColumn("Chunk #", width="small"), + "Text": st.column_config.TextColumn("Text", width="large"), + "Has Embedding": st.column_config.CheckboxColumn("Has Embedding"), + "Chunk Size": st.column_config.NumberColumn("Chunk Size", width="small"), + "Chunk Overlap": st.column_config.NumberColumn("Chunk Overlap", width="small"), + } + + st.dataframe( + data=chunks_df, + use_container_width=True, + hide_index=True, + column_config=column_config, + ) + + +def _render_text_only_chunks(raw_chunks: list[dict]) -> None: + """Chunk-only cache: list chunks without similarity controls.""" + st.subheader("Document Chunks") + st.caption("Text-only chunks. Run the **Embed** step on Report Analyst to enable similarity search.") + rows = _chunks_to_rows(raw_chunks) + _display_chunk_table(rows, similarity_view=False) + st.info(f"Found {len(rows)} document chunks (not embedded yet).") + + +def _render_similarity_chunk_search(analyzer, question_set: str, file_path: str, raw_chunks: list[dict]) -> None: + """Embedded chunks: question/custom query similarity ranking.""" + st.subheader("Similarity Search") + + if analyzer.analyzer.question_set != question_set: + analyzer.analyzer.update_question_set(question_set) + questions = analyzer.analyzer.questions + + col1, col2 = st.columns([1, 1]) + with col1: + question_options = ["None"] + [f"{q_id}" for q_id in questions.keys()] + selected_question_id = st.selectbox( + "Select a question to sort by similarity:", + options=question_options, + key=f"chunk_similarity_question_{Path(file_path).name}", + help="Choose a question from the current question set", + ) + if selected_question_id != "None" and selected_question_id in questions: + st.caption(f"**{selected_question_id}:** {questions[selected_question_id]['text'][:100]}...") + + with col2: + custom_question = st.text_input( + "Or enter custom question:", + placeholder="Enter your own question to compare chunks against...", + key=f"chunk_similarity_custom_{Path(file_path).name}", + ) + + query_text = None + if custom_question.strip(): + query_text = custom_question.strip() + st.info(f"Using custom question: {query_text[:100]}...") + elif selected_question_id != "None" and selected_question_id in questions: + query_text = questions[selected_question_id]["text"] + st.info(f"Using question {selected_question_id}: {query_text[:100]}...") + + embedded_chunks = [c for c in raw_chunks if c.get("embedding") is not None] + display_chunks = raw_chunks + similarity_view = False + + if query_text: + try: + if analyzer.analyzer.use_backend_llm: + st.warning("Similarity search is unavailable in backend LLM mode.") + else: + analyzer.analyzer._ensure_embeddings_client() + query_embedding = np.array( + analyzer.analyzer.embeddings.get_text_embedding(query_text), + dtype=np.float32, + ) + ranked = [] + for chunk in embedded_chunks: + chunk_embedding = np.frombuffer(chunk["embedding"], dtype=np.float32) + similarity = float( + np.dot(query_embedding, chunk_embedding) + / (np.linalg.norm(query_embedding) * np.linalg.norm(chunk_embedding)) + ) + ranked.append({**chunk, "similarity": similarity}) + ranked.sort(key=lambda c: c["similarity"], reverse=True) + display_chunks = ranked + similarity_view = True + st.success(f"Sorted {len(ranked)} embedded chunks by similarity to query") + except RuntimeError as exc: + st.warning(str(exc)) + except Exception as e: + st.error(f"Error computing similarity: {e!s}") + logger.error(f"Error computing similarity: {e!s}", exc_info=True) + + rows = _chunks_to_rows(display_chunks, ranked=similarity_view) + _display_chunk_table(rows, similarity_view=similarity_view) + st.info(f"Found {len(raw_chunks)} document chunks ({len(embedded_chunks)} with embeddings).") + + +def render_consolidated_chunk_search(analyzer, question_set: str, file_path: str, config: dict) -> bool: + """Render cached document chunks. Similarity search only when embeddings exist.""" + raw_chunks = analyzer.analyzer.cache_manager.resolve_document_chunks( + file_path=file_path, + chunk_size=config["chunk_size"], + chunk_overlap=config["chunk_overlap"], + ) + + if not raw_chunks: + st.subheader("Document Chunks") + st.warning( + "No document chunks in cache for this report and configuration. " + "On **Report Analyst**, set the processing step to **Chunk** (or **Embed**) " + "and click **Analyze Selected Questions** for this file first." + ) + return False + + embedded_count = sum(1 for c in raw_chunks if c.get("embedding") is not None) + if embedded_count == 0: + _render_text_only_chunks(raw_chunks) + else: + _render_similarity_chunk_search(analyzer, question_set, file_path, raw_chunks) + + return True + + +def render_consolidated_report_view( + analyzer, + question_set: str, + file_path: str, + config: dict, + *, + display_analysis_results, + display_pdf_viewer=None, +) -> None: + """Show chunk search and optional cached answer results for one report/config.""" + logger.info(f"Getting results for {Path(file_path).name} with config: {config}") + + try: + had_chunks = render_consolidated_chunk_search(analyzer, question_set, file_path, config) + except Exception as e: # noqa: BLE001 + logger.warning(f"Error displaying document chunks: {e!s}") + had_chunks = False + + cached_results = analyzer.analyzer.cache_manager.get_analysis( + file_path=file_path, + config=config, + ) + + if not cached_results: + if config.get("chunks_only") or had_chunks: + st.info("No answer results yet for this configuration.") + else: + st.warning("No stored results found for this configuration") + return + + if analyzer.analyzer.question_set != question_set: + analyzer.analyzer.update_question_set(question_set) + questions = analyzer.analyzer.questions + + analysis_rows = [] + question_chunks_rows = [] + + for question_id, data in cached_results.items(): + try: + result = data.get("result", {}) + analysis_rows.append( + { + "Question ID": question_id, + "Question Text": (questions[question_id]["text"] if question_id in questions else question_id), + "Analysis": result.get("ANSWER", ""), + "Score": float(result.get("SCORE", 0)), + "Key Evidence": "\n".join([e.get("text", "") for e in result.get("EVIDENCE", [])]), + "Gaps": "\n".join(result.get("GAPS", [])), + "Sources": ", ".join(map(str, result.get("SOURCES", []))), + } + ) + if "chunks" in data: + for chunk in data["chunks"]: + question_chunks_rows.append( + { + "Question ID": question_id, + "Text": chunk.get("text", ""), + "Vector Similarity": chunk.get("similarity_score", 0.0), + "LLM Score": chunk.get("llm_score", 0.0), + "Is Evidence": chunk.get("is_evidence", False), + "Position": chunk.get("chunk_order", 0), + } + ) + except Exception as e: + logger.error(f"Error processing result for question {question_id}: {e!s}", exc_info=True) + + if analysis_rows: + analysis_df = pd.DataFrame(analysis_rows) + chunks_df = pd.DataFrame(question_chunks_rows) if question_chunks_rows else pd.DataFrame() + file_key = f"{Path(file_path).stem}_cs{config['chunk_size']}" + display_analysis_results(analysis_df, chunks_df, file_key) + if display_pdf_viewer is not None: + display_pdf_viewer(file_path, cached_results, questions) + else: + st.warning("No results found in stored for this configuration") diff --git a/report_analyst/core/analyzer.py b/report_analyst/core/analyzer.py index 7a7c0423..5f39251b 100644 --- a/report_analyst/core/analyzer.py +++ b/report_analyst/core/analyzer.py @@ -14,18 +14,41 @@ from llama_index.core.ingestion import IngestionCache from llama_index.core.llms import ChatMessage from llama_index.core.node_parser import SentenceSplitter -from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.readers.file import PyMuPDFReader from .api_key_manager import APIKeyManager from .cache_manager import CacheManager -from .llm_providers import get_llm +from .llm_providers import build_openai_embedding, get_llm, is_permanent_openai_quota_error from .prompt_manager import PromptManager from .storage import LlamaVectorStore # Setup logging at the top of the file logger = logging.getLogger(__name__) +PROCESSING_STEPS = ("chunk", "embed", "map", "answer") +PROCESSING_STEP_LABELS = { + "chunk": "Chunk", + "embed": "Embed", + "map": "Map", + "answer": "Answer", +} + + +def normalize_processing_step(step: str | None) -> str: + """Map UI labels (Chunk/Embed/...) or keys to a processing step key.""" + if not step: + return "answer" + key = step.strip().lower() + if key in PROCESSING_STEPS: + return key + label_map = {v.lower(): k for k, v in PROCESSING_STEP_LABELS.items()} + return label_map.get(key, "answer") + + +def processing_step_rank(step: str) -> int: + return PROCESSING_STEPS.index(normalize_processing_step(step)) + + # Load environment variables load_dotenv() @@ -207,12 +230,7 @@ def _initialize_llm_clients(self) -> None: current_openai_key = os.getenv("OPENAI_API_KEY") if APIKeyManager.is_configured_key(current_openai_key): - self.embeddings = OpenAIEmbedding( - api_key=current_openai_key, - api_base=os.getenv("OPENAI_API_BASE"), - model_name=os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-ada-002"), - embed_batch_size=100, - ) + self.embeddings = build_openai_embedding(api_key=current_openai_key) Settings.embed_model = self.embeddings else: logger.warning("No OpenAI API key - embedding functionality will be limited") @@ -552,6 +570,7 @@ async def process_document( single_call: bool = True, force_recompute: bool = False, pre_retrieved_chunks: Optional[List[Dict[str, Any]]] = None, + max_processing_step: str = "answer", ) -> AsyncGenerator[Dict, None]: """Process a document for selected questions @@ -562,16 +581,20 @@ async def process_document( single_call: Whether to use single LLM call per question force_recompute: Whether to force recomputation pre_retrieved_chunks: Optional pre-retrieved chunks (e.g., from backend) + max_processing_step: Stop after this pipeline stage (chunk/embed/map/answer) """ try: + step_key = normalize_processing_step(max_processing_step) + step_level = processing_step_rank(step_key) # Add more detailed logging logger.info(f"[ANALYSIS] Starting document processing for {file_path}") logger.info(f"[ANALYSIS] Selected questions: {selected_questions}") logger.info( - "[ANALYSIS] LLM scoring: %s, Single call: %s, Force recompute: %s", + "[ANALYSIS] LLM scoring: %s, Single call: %s, Force recompute: %s, Max step: %s", use_llm_scoring, single_call, force_recompute, + step_key, ) logger.info( "[ANALYSIS] Current chunk parameters: size=%s, overlap=%s, top_k=%s", @@ -580,7 +603,7 @@ async def process_document( self.chunk_params["top_k"], ) - if not self.use_backend_llm and self.llm is None: + if step_level >= processing_step_rank("map") and not self.use_backend_llm and self.llm is None: yield { "error": ( "No API key configured. Add an OpenAI or Google/Gemini key in Settings → API Keys " @@ -632,7 +655,6 @@ async def process_document( if not chunks: logger.info("[ANALYSIS] No chunks found in cache with current parameters, creating new chunks") - # If no chunks in cache with current parameters, create them # Check if file_path is a URN (backend resource) if file_path.startswith("urn:report-analyst:backend:"): logger.warning( @@ -646,19 +668,52 @@ async def process_document( ) } return + if step_level == processing_step_rank("chunk"): + chunks = self._create_text_chunks(file_path) + logger.info(f"[ANALYSIS] Created {len(chunks)} text-only chunks") + self.cache_manager.save_text_only_chunks( + file_path=file_path, + chunks=chunks, + chunk_size=self.chunk_params["chunk_size"], + chunk_overlap=self.chunk_params["chunk_overlap"], + ) + logger.info(f"[ANALYSIS] Saved {len(chunks)} text-only chunks to cache") + else: + chunks = self._create_chunks(file_path) + logger.info(f"[ANALYSIS] Created {len(chunks)} new chunks with embeddings") + self.cache_manager.save_document_chunks( + file_path=file_path, + chunks=chunks, + chunk_size=self.chunk_params["chunk_size"], + chunk_overlap=self.chunk_params["chunk_overlap"], + ) + logger.info(f"[ANALYSIS] Saved {len(chunks)} chunks to cache") + + yield {"status": f"Document loaded with {len(chunks)} chunks"} + + if step_level >= processing_step_rank("embed") and not any(c.get("embedding") is not None for c in chunks): + if not chunks: + if file_path.startswith("urn:report-analyst:backend:"): + yield {"error": "Backend resource requires pre-retrieved chunks with embeddings."} + return chunks = self._create_chunks(file_path) - logger.info(f"[ANALYSIS] Created {len(chunks)} new chunks") + else: + chunks = self._add_embeddings_to_chunks(chunks) + self.cache_manager.save_document_chunks( + file_path=file_path, + chunks=chunks, + chunk_size=self.chunk_params["chunk_size"], + chunk_overlap=self.chunk_params["chunk_overlap"], + ) + logger.info(f"[ANALYSIS] Stored {len(chunks)} chunks with embeddings") - # Save chunks to cache with current parameters - self.cache_manager.save_document_chunks( - file_path=file_path, - chunks=chunks, - chunk_size=self.chunk_params["chunk_size"], - chunk_overlap=self.chunk_params["chunk_overlap"], - ) - logger.info(f"[ANALYSIS] Saved {len(chunks)} chunks to cache") + if step_level <= processing_step_rank("embed"): + yield {"status": f"Completed {PROCESSING_STEP_LABELS[step_key]} step ({len(chunks)} chunks)"} + return - yield {"status": f"Document loaded with {len(chunks)} chunks"} + if not selected_questions: + yield {"error": "Select at least one question for Map or Answer steps."} + return # 2. Process each question for question_number in selected_questions: @@ -734,6 +789,14 @@ async def process_document( for chunk in similar_chunks: chunk["llm_score"] = None + if step_level <= processing_step_rank("map"): + yield { + "status": ( + f"Completed Map step for question {question_number} " f"({len(similar_chunks)} chunks ranked)" + ) + } + continue + # 4. Run LLM analysis (evidence determination happens here) logger.info(f"[ANALYSIS] Running LLM analysis for question {question_id}") result = await self._analyze_chunks(question_data, similar_chunks, use_llm_scoring) @@ -826,6 +889,75 @@ async def process_document( logger.error(f"[ANALYSIS] Error processing document: {e!s}", exc_info=True) yield {"error": f"Error processing document: {e!s}"} + def _split_pdf_to_text_chunks(self, file_path: str) -> List[Dict[str, Any]]: + """Split a PDF into text chunks without computing embeddings.""" + reader = PyMuPDFReader() + docs = reader.load(file_path=file_path) + logger.info(f"Loaded {len(docs)} pages from document") + + text_chunks: List[Dict[str, Any]] = [] + for doc in docs: + nodes = self.text_splitter.split_text(doc.text) + for chunk in nodes: + text = chunk.strip() + if not text: + continue + text_chunks.append( + { + "text": text, + "metadata": { + **doc.metadata, + "chunk_size": self.chunk_params["chunk_size"], + "chunk_overlap": self.chunk_params["chunk_overlap"], + }, + "embedding": None, + "similarity": 0.0, + "computed_score": 0.0, + } + ) + logger.info(f"Split document into {len(text_chunks)} text chunks") + return text_chunks + + def _create_text_chunks(self, file_path: str) -> List[Dict[str, Any]]: + """Create document chunks without embeddings (Chunk step only).""" + try: + return self._split_pdf_to_text_chunks(file_path) + except Exception as e: + logger.error(f"Error creating text chunks: {e!s}", exc_info=True) + raise + + def _ensure_embeddings_client(self) -> None: + """Initialize or refresh OpenAI embeddings client from current OPENAI_API_KEY.""" + openai_key = os.getenv("OPENAI_API_KEY") + if not openai_key: + raise RuntimeError("OpenAI embeddings unavailable — set OPENAI_API_KEY for the Embed step.") + if self.embeddings is None: + self.embeddings = build_openai_embedding(api_key=openai_key) + Settings.embed_model = self.embeddings + + def _add_embeddings_to_chunks(self, chunks: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Compute embeddings for existing text chunks.""" + self._ensure_embeddings_client() + + embedded: List[Dict[str, Any]] = [] + batch_size = 100 + texts = [c["text"] for c in chunks if c.get("text")] + for i in range(0, len(texts), batch_size): + batch_texts = texts[i : i + batch_size] + batch_embeddings = self.embeddings.get_text_embedding_batch(batch_texts) + for text, embedding in zip(batch_texts, batch_embeddings, strict=False): + if embedding is None: + continue + source = next(c for c in chunks if c["text"] == text) + embedded.append( + { + **source, + "embedding": np.array(embedding, dtype=np.float32), + } + ) + logger.info(f"Added embeddings to {len(embedded)}/{len(chunks)} chunks") + return embedded + def _create_chunks(self, file_path: str) -> List[Dict[str, Any]]: """Create document chunks with embeddings""" try: @@ -900,6 +1032,8 @@ def _create_chunks(self, file_path: str) -> List[Dict[str, Any]]: logger.warning("Skipping chunk - embedding is None") except Exception as e: + if is_permanent_openai_quota_error(e): + raise logger.error(f"Error computing embeddings for batch: {e!s}", exc_info=True) # Continue with next batch, storing chunks without embeddings for chunk in batch: @@ -1372,6 +1506,8 @@ async def _get_similar_chunks(self, query_text: str, chunks: List[Dict], top_k: return similar_chunks except Exception as e: + if is_permanent_openai_quota_error(e): + raise logger.error(f"Error getting similar chunks: {e!s}", exc_info=True) return [] diff --git a/report_analyst/core/cache_manager.py b/report_analyst/core/cache_manager.py index cd585742..d49ed2b8 100644 --- a/report_analyst/core/cache_manager.py +++ b/report_analyst/core/cache_manager.py @@ -903,6 +903,92 @@ def get_all_answers_by_question_set(self, question_set: str) -> Dict[str, Any]: logger.error(f"Error retrieving answers for question set {question_set}: {e}") raise + def list_document_chunk_configs(self) -> List[tuple]: + """Distinct (file_path, chunk_size, chunk_overlap) rows from document_chunks.""" + try: + with self.db_manager.get_connection() as conn: + result_obj = conn.execute(text(""" + SELECT DISTINCT file_path, chunk_size, chunk_overlap + FROM document_chunks + ORDER BY file_path, chunk_size, chunk_overlap + """)) + return list(result_obj.fetchall()) + except Exception as e: + logger.error(f"Error listing document chunk configs: {e}", exc_info=True) + return [] + + def resolve_document_chunks( + self, + file_path: str, + chunk_size: int | None = None, + chunk_overlap: int | None = None, + ) -> List[Dict]: + """Load document chunks, falling back to the same filename on a different path.""" + chunks = self.get_document_chunks(file_path, chunk_size, chunk_overlap) + if chunks: + return chunks + + target_name = Path(file_path).name + for alt_path, cs, co in self.list_document_chunk_configs(): + if Path(alt_path).name != target_name: + continue + if chunk_size is not None and cs != chunk_size: + continue + if chunk_overlap is not None and co != chunk_overlap: + continue + chunks = self.get_document_chunks(alt_path, chunk_size, chunk_overlap) + if chunks: + logger.info( + "Resolved document chunks for %s via alternate path %s", + file_path, + alt_path, + ) + return chunks + return [] + + def save_text_only_chunks(self, file_path: str, chunks: List[Dict], chunk_size: int, chunk_overlap: int) -> None: + """Save document chunks without embeddings (Chunk step).""" + try: + logger.info(f"Saving {len(chunks)} text-only chunks for {file_path}") + timestamp = datetime.now().isoformat() + + with self.db_manager.get_connection() as conn: + for chunk in chunks: + metadata_json = json.dumps(chunk.get("metadata", {})) + params = { + "file_path": str(file_path), + "chunk_text": chunk["text"], + "chunk_size": chunk_size, + "chunk_overlap": chunk_overlap, + "metadata": metadata_json, + "created_at": timestamp, + } + if self.db_manager.is_postgres(): + conn.execute( + text(""" + INSERT INTO document_chunks + (file_path, chunk_text, chunk_size, chunk_overlap, embedding, metadata, created_at) + VALUES (:file_path, :chunk_text, :chunk_size, :chunk_overlap, NULL, :metadata, :created_at) + ON CONFLICT (file_path, chunk_text, chunk_size, chunk_overlap) DO UPDATE + SET metadata = EXCLUDED.metadata, + created_at = EXCLUDED.created_at + """), + params, + ) + else: + conn.execute( + text(""" + INSERT OR REPLACE INTO document_chunks + (file_path, chunk_text, chunk_size, chunk_overlap, embedding, metadata, created_at) + VALUES (:file_path, :chunk_text, :chunk_size, :chunk_overlap, NULL, :metadata, :created_at) + """), + params, + ) + logger.info(f"Saved {len(chunks)} text-only chunks") + except Exception as e: + logger.error(f"Error saving text-only chunks: {e}") + raise + def save_document_chunks(self, file_path: str, chunks: List[Dict], chunk_size: int, chunk_overlap: int) -> None: """Save document chunks to cache with their embeddings.""" try: diff --git a/report_analyst/core/llm_providers.py b/report_analyst/core/llm_providers.py index bc5a96e8..c939712a 100644 --- a/report_analyst/core/llm_providers.py +++ b/report_analyst/core/llm_providers.py @@ -4,6 +4,7 @@ import os from typing import Any, Optional +from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.llms.google_genai import GoogleGenAI # LlamaIndex LLM imports @@ -12,6 +13,45 @@ # Setup logging logger = logging.getLogger(__name__) +# One HTTP timeout is enough for embed/LLM. LlamaIndex defaults also retry +# RateLimitError 10 times, including permanent insufficient_quota (429). +OPENAI_REQUEST_TIMEOUT_SECONDS = 60.0 +OPENAI_MAX_RETRIES = 0 + +_PERMANENT_QUOTA_CODES = frozenset({"insufficient_quota", "credit_balance_exhausted"}) + + +def is_permanent_openai_quota_error(exc: BaseException) -> bool: + """True for billing/quota 429s that will never succeed on retry.""" + code = getattr(exc, "code", None) + body = getattr(exc, "body", None) + err = body.get("error") if isinstance(body, dict) else None + if isinstance(err, dict): + code = code or err.get("code") + if err.get("type") == "insufficient_quota": + return True + return code in _PERMANENT_QUOTA_CODES + + +def build_openai_embedding( + *, + api_key: str, + api_base: Optional[str] = None, + model_name: Optional[str] = None, + **kwargs: Any, +) -> OpenAIEmbedding: + """OpenAI embeddings client that fails fast on permanent API errors.""" + embed_kwargs = { + "api_key": api_key, + "api_base": api_base if api_base is not None else os.getenv("OPENAI_API_BASE"), + "model_name": model_name or os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-ada-002"), + "embed_batch_size": 100, + "timeout": OPENAI_REQUEST_TIMEOUT_SECONDS, + "max_retries": OPENAI_MAX_RETRIES, + **kwargs, + } + return OpenAIEmbedding(**embed_kwargs) + def get_llm(model_name: str, cache_dir: Optional[str] = None, **kwargs) -> Any: """ @@ -36,13 +76,16 @@ def get_llm(model_name: str, cache_dir: Optional[str] = None, **kwargs) -> Any: logger.error(f"Cannot initialize OpenAI model '{model_name}' - OPENAI_API_KEY environment variable is not set") raise ValueError("OPENAI_API_KEY environment variable is required for OpenAI models") - return OpenAI( - model=model_name, - api_key=api_key, - api_base=os.getenv("OPENAI_API_BASE"), - cache_dir=cache_dir, + llm_kwargs = { + "model": model_name, + "api_key": api_key, + "api_base": os.getenv("OPENAI_API_BASE"), + "cache_dir": cache_dir, + "timeout": OPENAI_REQUEST_TIMEOUT_SECONDS, + "max_retries": OPENAI_MAX_RETRIES, **kwargs, - ) + } + return OpenAI(**llm_kwargs) # Gemini models elif model_name.startswith("gemini-") or model_name.startswith("models/gemini-"): diff --git a/report_analyst/streamlit_app.py b/report_analyst/streamlit_app.py index 80aabfbc..9b7758fd 100644 --- a/report_analyst/streamlit_app.py +++ b/report_analyst/streamlit_app.py @@ -2,7 +2,6 @@ import asyncio import base64 import html -import json import logging import os import sys @@ -11,7 +10,6 @@ from pathlib import Path from typing import Any, AsyncGenerator, Dict, List, Optional -import numpy as np import pandas as pd import streamlit as st from dotenv import load_dotenv @@ -85,6 +83,7 @@ def is_api_key_missing_message(message: str) -> bool: logger.info(f"Added {current_dir} to Python path") # Keep relative imports +from report_analyst.consolidated_results_view import render_consolidated_report_view from report_analyst.core.analyzer import DocumentAnalyzer from report_analyst.core.api_key_manager import APIKeyManager from report_analyst.core.dataframe_manager import ( @@ -225,6 +224,7 @@ async def analyze_document( single_call: bool = True, force_recompute: bool = False, pre_retrieved_chunks: Optional[List[Dict[str, Any]]] = None, + max_processing_step: str = "answer", ) -> AsyncGenerator[Dict, None]: """Analyze a document using the provided questions @@ -260,6 +260,7 @@ async def analyze_document( single_call, force_recompute, pre_retrieved_chunks=pre_retrieved_chunks, + max_processing_step=max_processing_step, ): # Pass through status and error messages if "status" in result or "error" in result: @@ -304,10 +305,17 @@ def process_document( single_call: bool = True, force_recompute: bool = False, pre_retrieved_chunks: Optional[List[Dict[str, Any]]] = None, + max_processing_step: str = "answer", ): """Delegate to the analyzer's process_document method""" return self.analyzer.process_document( - file_path, selected_questions, use_llm_scoring, single_call, force_recompute, pre_retrieved_chunks + file_path, + selected_questions, + use_llm_scoring, + single_call, + force_recompute, + pre_retrieved_chunks=pre_retrieved_chunks, + max_processing_step=max_processing_step, ) @@ -488,6 +496,7 @@ async def analyze_document_and_display( use_llm_scoring: bool = False, single_call: bool = True, force_recompute: bool = False, + max_processing_step: str = "answer", ): """Analyze document and display results as they come in""" try: @@ -574,6 +583,7 @@ async def analyze_document_and_display( single_call, force_recompute, pre_retrieved_chunks=pre_retrieved_chunks, + max_processing_step=max_processing_step, ): # Add debug logging to see what results we're getting log_analysis_step(f"Received result: {str(result)[:200]}...") @@ -822,6 +832,56 @@ def get_uploaded_files_history(backend_config=None) -> List[Dict]: return result +def display_cached_document_chunks( + report_analyzer, + file_path: str, + *, + chunk_size: int | None = None, + chunk_overlap: int | None = None, +) -> tuple[int, int]: + """Render document chunks from cache (Chunk/Embed steps — no question analysis required). + + Returns: + (total_chunks, chunks_with_embeddings) + """ + raw_chunks = report_analyzer.cache_manager.get_document_chunks( + file_path=file_path, + chunk_size=chunk_size, + chunk_overlap=chunk_overlap, + ) + if not raw_chunks: + st.warning("No chunks found in cache for this file and configuration.") + return 0, 0 + + rows = [] + embedded_count = 0 + for i, chunk in enumerate(raw_chunks): + has_embedding = chunk.get("embedding") is not None + if has_embedding: + embedded_count += 1 + rows.append( + { + "Chunk #": i + 1, + "Text": chunk.get("text", chunk.get("chunk_text", "")), + "Has Embedding": has_embedding, + } + ) + + st.subheader("Document Chunks") + st.dataframe( + pd.DataFrame(rows), + use_container_width=True, + hide_index=True, + column_config={ + "Chunk #": st.column_config.NumberColumn("Chunk #", width="small"), + "Text": st.column_config.TextColumn("Text", width="large"), + "Has Embedding": st.column_config.CheckboxColumn("Has Embedding"), + }, + ) + st.info(f"Found {len(rows)} document chunks ({embedded_count} with embeddings).") + return len(rows), embedded_count + + def display_analysis_results(analysis_df: pd.DataFrame, chunks_df: pd.DataFrame, file_key: str | None = None) -> None: """Display analysis results in a consistent format for both individual and consolidated views""" try: @@ -923,378 +983,120 @@ def display_analysis_results(analysis_df: pd.DataFrame, chunks_df: pd.DataFrame, st.error(f"Error displaying results: {e!s}") +def build_all_results_file_configs( + cache_manager, + db_question_set: str, + *, + default_top_k: int, + default_model: str, +) -> dict[str, list[dict]]: + """Merge analysis_cache configs with document_chunks-only configs for All Results.""" + file_configs: dict[str, list[dict]] = {} + + def add_config( + file_path: str, + chunk_size: int, + chunk_overlap: int, + top_k: int, + model: str, + question_set: str, + *, + chunks_only: bool = False, + ) -> None: + configs = file_configs.setdefault(str(file_path), []) + if any(c["chunk_size"] == chunk_size and c["chunk_overlap"] == chunk_overlap for c in configs): + return + entry = { + "chunk_size": chunk_size, + "chunk_overlap": chunk_overlap, + "top_k": top_k, + "model": model, + "question_set": question_set, + } + if chunks_only: + entry["chunks_only"] = True + configs.append(entry) + + for config in cache_manager.check_cache_status(): + if len(config) != 6: + continue + fp, chunk_size, chunk_overlap, top_k, model, qs = config + if qs == db_question_set: + add_config(fp, chunk_size, chunk_overlap, top_k, model, qs) + + for fp, chunk_size, chunk_overlap in cache_manager.list_document_chunk_configs(): + add_config( + fp, + chunk_size, + chunk_overlap, + default_top_k, + default_model, + db_question_set, + chunks_only=True, + ) + + return file_configs + + def display_pdf_viewer( file_path: str, results: Dict[str, Dict], questions: Dict[str, Dict], raw_chunks: Optional[List[Dict[str, Any]]] = None, ) -> None: - chunks_by_question = {question_id: data.get("chunks", []) for question_id, data in results.items()} - questions_data = {question_id: question.get("text", question_id) for question_id, question in questions.items()} + chunks_by_question = {question_id: data.get("chunks", []) for question_id, data in (results or {}).items()} + questions_data = {question_id: question.get("text", question_id) for question_id, question in (questions or {}).items()} viewer_key = Path(str(file_path)).stem or "analysis" with st.expander("PDF Viewer with Chunks", expanded=True): - pdf_viewer( - pdf_path=str(file_path), - chunks_data=chunks_by_question, - questions_data=questions_data, - unmapped_chunks=raw_chunks, - key=f"pdf_viewer_{viewer_key}", - height=800, - ) + try: + pdf_viewer( + pdf_path=str(file_path), + chunks_data=chunks_by_question, + questions_data=questions_data, + unmapped_chunks=raw_chunks, + key=f"pdf_viewer_{viewer_key}", + height=800, + ) + except Exception as e: + logger.error(f"Error rendering PDF viewer: {e!s}", exc_info=True) + st.error(f"Error rendering PDF viewer: {e!s}") def display_consolidated_results(analyzer, question_set, file_path=None, selected_config=None): - """Display consolidated results for all analyzed documents - - Args: - analyzer: ReportAnalyzer instance - question_set: Selected question set identifier - file_path: Optional file path. If provided, skip file selection and use this file. - If None, will attempt to get file from cache (backward compatibility). - selected_config: Optional configuration dict. If provided, skip config selection and use this config. - If None, will attempt to get config from cache (backward compatibility). - """ + """Display consolidated results for all analyzed documents.""" try: - # Create mapping from question set names to database identifiers question_set_mapping = { "tcfd": "tcfd", "s4m": "s4m", "lucia": "lucia", - "everest": "ev", # Everest questions use 'ev_' prefix, so database stores as 'ev' + "everest": "ev", } - - # Get the database identifier for the selected question set db_question_set = question_set_mapping.get(question_set, question_set) - logger.info(f"Mapping question set '{question_set}' to database identifier '{db_question_set}'") - - # Get all available cache configurations - cache_configs = analyzer.analyzer.cache_manager.check_cache_status() - logger.info(f"Found cache configs: {cache_configs}") - if not cache_configs: - st.warning("No stored analyses found") + if file_path is not None and selected_config is not None: + config = selected_config.get("config", selected_config) + render_consolidated_report_view( + analyzer, + question_set, + file_path, + config, + display_analysis_results=display_analysis_results, + display_pdf_viewer=display_pdf_viewer, + ) return - # Group configurations by file - file_configs = {} - for config in cache_configs: - if len(config) == 6: # Full config row from cache_status - file_path, chunk_size, chunk_overlap, top_k, model, qs = config - if qs == db_question_set: # Only show configs for selected question set using database identifier - if file_path not in file_configs: - file_configs[file_path] = [] - file_configs[file_path].append( - { - "chunk_size": chunk_size, - "chunk_overlap": chunk_overlap, - "top_k": top_k, - "model": model, - "question_set": qs, - } - ) + file_configs = build_all_results_file_configs( + analyzer.analyzer.cache_manager, + db_question_set, + default_top_k=st.session_state.get("top_k", 5), + default_model=st.session_state.get("llm_model", DEFAULT_LLM_MODEL), + ) if not file_configs: - st.warning(f"No stored results found for question set: {question_set}") + st.warning("No stored analyses or cached document chunks found") return - # File selection - st.subheader("Select Report and Configuration") - file_path = st.selectbox( - "Select Report", - options=list(file_configs.keys()), - format_func=lambda x: Path(x).name, - ) - - if file_path: - # Show configurations for selected file - configs = file_configs[file_path] - config_options = [] - for config in configs: - label = ( - f"Chunk: {config['chunk_size']}, Overlap: {config['chunk_overlap']}, " - f"Top-K: {config['top_k']}, Model: {config['model']}" - ) - config_options.append({"label": label, "config": config}) - - selected_config = st.selectbox( - "Select Configuration", - options=config_options, - format_func=lambda x: x["label"], - ) - - if selected_config: - logger.info(f"Getting results for {Path(file_path).name} with config: {selected_config['config']}") - - # Add similarity search section for document chunks - try: - raw_chunks = analyzer.analyzer.cache_manager.get_document_chunks( - file_path=file_path, - chunk_size=selected_config["config"]["chunk_size"], - chunk_overlap=selected_config["config"]["chunk_overlap"], - ) - - if raw_chunks: - # Add similarity search controls - st.subheader("Similarity Search") - - # Get questions for the current question set - # Make sure analyzer is using the correct question set - if analyzer.analyzer.question_set != question_set: - analyzer.analyzer.update_question_set(question_set) - questions = analyzer.analyzer.questions - - col1, col2 = st.columns([1, 1]) - with col1: - # Question dropdown with shorter, cleaner options - question_options = ["None"] + [f"{q_id}" for q_id in questions.keys()] - selected_question_id = st.selectbox( - "Select a question to sort by similarity:", - options=question_options, - key="chunk_similarity_question", - help="Choose a question from the current question set", - ) - - # Show selected question text below dropdown - if selected_question_id != "None" and selected_question_id in questions: - st.caption(f"**{selected_question_id}:** {questions[selected_question_id]['text'][:100]}...") - selected_question = selected_question_id - - with col2: - # Free text input - custom_question = st.text_input( - "Or enter custom question:", - placeholder="Enter your own question to compare chunks against...", - key="chunk_similarity_custom", - ) - - # Determine which question to use - query_text = None - if custom_question.strip(): - query_text = custom_question.strip() - st.info(f"Using custom question: {query_text[:100]}...") - elif selected_question != "None": - if selected_question in questions: - query_text = questions[selected_question]["text"] - st.info(f"Using question {selected_question}: {query_text[:100]}...") - - # Process chunks - chunks_rows = [] - chunks_with_embeddings = [c for c in raw_chunks if c.get("embedding") is not None] - - if query_text and chunks_with_embeddings: - # Compute similarity scores - try: - # Check if embeddings are available - if not analyzer.analyzer.embeddings or analyzer.analyzer.use_backend_llm: - st.warning( - "Embeddings not available for similarity search. " - "Using backend mode or embeddings not initialized." - ) - query_text = None - else: - # Get query embedding - query_embedding = analyzer.analyzer.embeddings.get_text_embedding(query_text) - query_embedding = np.array(query_embedding, dtype=np.float32) - - # Compute similarity for each chunk - similarities = [] - for chunk in raw_chunks: - if chunk.get("embedding") is not None: - chunk_embedding = np.frombuffer(chunk["embedding"], dtype=np.float32) - # Compute cosine similarity - similarity = np.dot(query_embedding, chunk_embedding) / ( - np.linalg.norm(query_embedding) * np.linalg.norm(chunk_embedding) - ) - similarities.append(similarity) - else: - similarities.append(0.0) - - # Sort chunks by similarity - chunk_similarity_pairs = list(zip(raw_chunks, similarities, strict=False)) - chunk_similarity_pairs.sort(key=lambda x: x[1], reverse=True) - - # Create rows with similarity scores - for i, (chunk, similarity) in enumerate(chunk_similarity_pairs): - chunk_row = { - "Rank": i + 1, - "Similarity": similarity, - "Text": chunk.get("text", chunk.get("chunk_text", "")), - "Has Embedding": chunk.get("embedding") is not None, - "Chunk Size": chunk.get("chunk_size", "N/A"), - "Chunk Overlap": chunk.get("chunk_overlap", "N/A"), - } - chunks_rows.append(chunk_row) - - st.success(f"✓ Sorted {len(chunks_rows)} chunks by similarity to query") - - except Exception as e: - st.error(f"Error computing similarity: {e!s}") - logger.error( - f"Error computing similarity: {e!s}", - exc_info=True, - ) - # Fall back to original display - for i, chunk in enumerate(raw_chunks): - chunk_row = { - "Chunk #": i + 1, - "Text": chunk.get("text", chunk.get("chunk_text", "")), - "Has Embedding": chunk.get("embedding") is not None, - "Chunk Size": chunk.get("chunk_size", "N/A"), - "Chunk Overlap": chunk.get("chunk_overlap", "N/A"), - } - chunks_rows.append(chunk_row) - - else: - # No query or no embeddings - show original order - for i, chunk in enumerate(raw_chunks): - chunk_row = { - "Chunk #": i + 1, - "Text": chunk.get("text", chunk.get("chunk_text", "")), - "Has Embedding": chunk.get("embedding") is not None, - "Chunk Size": chunk.get("chunk_size", "N/A"), - "Chunk Overlap": chunk.get("chunk_overlap", "N/A"), - } - chunks_rows.append(chunk_row) - - if query_text and not chunks_with_embeddings: - st.warning( - "No chunks with embeddings found. Run Step 2 to generate embeddings for similarity search." - ) - - # Display chunks - if chunks_rows: - chunks_df = pd.DataFrame(chunks_rows) - - # Configure columns based on whether we have similarity scores - if query_text and chunks_with_embeddings: - column_config = { - "Rank": st.column_config.NumberColumn( - "Rank", - width="small", - ), - "Similarity": st.column_config.NumberColumn( - "Similarity", - format="%.4f", - width="small", - ), - "Text": st.column_config.TextColumn( - "Text", - width="large", - ), - "Has Embedding": st.column_config.CheckboxColumn( - "Has Embedding", - ), - "Chunk Size": st.column_config.NumberColumn( - "Chunk Size", - width="small", - ), - "Chunk Overlap": st.column_config.NumberColumn( - "Chunk Overlap", - width="small", - ), - } - else: - column_config = { - "Chunk #": st.column_config.NumberColumn( - "Chunk #", - width="small", - ), - "Text": st.column_config.TextColumn( - "Text", - width="large", - ), - "Has Embedding": st.column_config.CheckboxColumn( - "Has Embedding", - ), - "Chunk Size": st.column_config.NumberColumn( - "Chunk Size", - width="small", - ), - "Chunk Overlap": st.column_config.NumberColumn( - "Chunk Overlap", - width="small", - ), - } - - st.dataframe( - data=chunks_df, - use_container_width=True, - hide_index=True, - column_config=column_config, - ) - - st.info(f"✓ Found {len(chunks_rows)} total document chunks in this configuration.") - else: - st.warning("No chunks found. Run Step 1 to generate document chunks first.") - - except Exception as e: - logger.warning(f"Error displaying document chunks with similarity search: {e!s}") - # Continue to show analysis results even if chunk display fails - - # Get cached results - cached_results = analyzer.analyzer.cache_manager.get_analysis( - file_path=file_path, config=selected_config["config"] - ) - - if cached_results: - # Get questions data - questions = analyzer.analyzer.questions - - # Process results into analysis rows - analysis_rows = [] - chunks_rows = [] - - for question_id, data in cached_results.items(): - try: - # Create analysis row - result = data.get("result", {}) - analysis_row = { - "Question ID": question_id, - "Question Text": (questions[question_id]["text"] if question_id in questions else question_id), - "Analysis": result.get("ANSWER", ""), - "Score": float(result.get("SCORE", 0)), - "Key Evidence": "\n".join([e.get("text", "") for e in result.get("EVIDENCE", [])]), - "Gaps": "\n".join(result.get("GAPS", [])), - "Sources": ", ".join(map(str, result.get("SOURCES", []))), - } - analysis_rows.append(analysis_row) - logger.debug(f"Added analysis row for {question_id}: {json.dumps(analysis_row, indent=2)}") - - # Process chunks if available - if "chunks" in data: - for chunk in data["chunks"]: - chunk_row = { - "Question ID": question_id, - "Text": chunk.get("text", ""), - "Vector Similarity": chunk.get("similarity_score", 0.0), - "LLM Score": chunk.get("llm_score", 0.0), - "Is Evidence": chunk.get("is_evidence", False), - "Position": chunk.get("chunk_order", 0), - } - chunks_rows.append(chunk_row) - - except Exception as e: - logger.error( - f"Error processing result for question {question_id}: {e!s}", - exc_info=True, - ) - continue - - # Create DataFrames - if analysis_rows: - analysis_df = pd.DataFrame(analysis_rows) - chunks_df = pd.DataFrame(chunks_rows) if chunks_rows else pd.DataFrame() - - # Display results using the existing display function - file_key = f"{Path(file_path).stem}_cs{selected_config['config']['chunk_size']}" - display_analysis_results(analysis_df, chunks_df, file_key) - display_pdf_viewer(file_path, cached_results, questions) - else: - st.warning("No results found in stored for this configuration") - else: - st.warning("No stored results found for this configuration") - except Exception as e: logger.error(f"Error displaying consolidated results: {e!s}", exc_info=True) st.error(f"Error displaying consolidated results: {e!s}") @@ -1370,6 +1172,39 @@ def get_current_settings(st) -> dict: } +def get_max_processing_step() -> str: + """Return processing step key from Streamlit session state (Chunk/Embed/Map/Answer slider).""" + from report_analyst.core.analyzer import normalize_processing_step + + return normalize_processing_step(st.session_state.get("processing_steps_slider", "Answer")) + + +def selected_question_ids_from_editor(edited_df: pd.DataFrame) -> list[str]: + """Return QID values for rows checked in the question data editor.""" + if edited_df.empty or "Select" not in edited_df.columns or "QID" not in edited_df.columns: + return [] + selected = edited_df["Select"].fillna(False).astype(bool) + return edited_df.loc[selected, "QID"].astype(str).tolist() + + +def processing_step_needs_questions_for(step: str) -> bool: + """True when the step runs question scoring/answer (Map or Answer).""" + from report_analyst.core.analyzer import normalize_processing_step + + return normalize_processing_step(step) in ("map", "answer") + + +def processing_step_needs_questions() -> bool: + return processing_step_needs_questions_for(get_max_processing_step()) + + +def is_partial_processing_step(step: str | None = None) -> bool: + """True for Chunk/Embed — pipeline stops before Map/Answer LLM work.""" + if step is None: + step = get_max_processing_step() + return not processing_step_needs_questions_for(step) + + def update_analyzer_parameters(): """Update analyzer parameters based on session state.""" if "analyzer" not in st.session_state: @@ -1423,6 +1258,172 @@ def update_analyzer_parameters(): st.error(f"Error updating parameters: {e!s}") +async def run_analysis(analyzer, file_path, selected_questions, progress_text, max_processing_step: str = "answer"): + """Run analysis and update the UI with progress""" + try: + from report_analyst.core.analyzer import PROCESSING_STEP_LABELS, normalize_processing_step + + step_key = normalize_processing_step(max_processing_step) + partial = is_partial_processing_step(step_key) + + config = { + "chunk_size": st.session_state.chunk_size, + "chunk_overlap": st.session_state.chunk_overlap, + "top_k": st.session_state.top_k, + "model": st.session_state.llm_model, + "question_set": st.session_state.question_set, + } + logger.info(f"[ANALYSIS] User triggered analysis for file: {file_path}") + logger.info(f"[ANALYSIS] Selected questions: {selected_questions}") + if "questions" in st.session_state: + logger.info( + "[ANALYSIS] Selected question texts: %s", + [st.session_state.questions[q]["text"] for q in selected_questions if q in st.session_state.questions], + ) + + force_recompute = st.session_state.get("force_recompute", False) + + if partial: + logger.info(f"[CACHE] Skipping analysis_cache lookup for partial step: {step_key}") + if not force_recompute: + chunk_params = analyzer.analyzer.chunk_params + raw_chunks = analyzer.cache_manager.get_document_chunks( + file_path=file_path, + chunk_size=chunk_params["chunk_size"], + chunk_overlap=chunk_params["chunk_overlap"], + ) + needs_embed = step_key == "embed" and not any(c.get("embedding") is not None for c in raw_chunks) + if raw_chunks and not needs_embed: + chunk_count, embedded_count = display_cached_document_chunks( + analyzer, + file_path, + chunk_size=chunk_params["chunk_size"], + chunk_overlap=chunk_params["chunk_overlap"], + ) + label = PROCESSING_STEP_LABELS.get(step_key, step_key) + if step_key == "embed" and embedded_count == 0: + progress_text.warning( + f"Found {chunk_count} chunks but none have embeddings yet. " + "Run Embed again with a valid OPENAI_API_KEY." + ) + else: + progress_text.success(f"Found stored chunks ({chunk_count}) — {label} step already complete.") + return + else: + logger.info( + f"[CACHE] Looking up cache for file: {file_path} with config: {config} and questions: {selected_questions}" + ) + question_ids = selected_questions + if selected_questions and not all("_" in q for q in selected_questions): + question_ids = [f"{config['question_set']}_{q}" for q in selected_questions] + cached_results = analyzer.cache_manager.get_analysis( + file_path=file_path, + config=config, + question_ids=question_ids, + ) + if cached_results and not force_recompute: + logger.info(f"[CACHE] Cache HIT for config: {config}") + progress_text.success("Found stored results!") + st.session_state.results = cached_results + logger.info(f"[ANALYSIS] Writing results to session state for file: {file_path}") + logger.info(f"[ANALYSIS] Attempting to display results for file: {file_path}") + return + logger.info(f"[CACHE] Cache MISS for config: {config}") + + progress_text.info("Starting analysis...") + + llm_scoring_enabled = st.session_state.get("new_llm_scoring", False) + progress_text.info(f"LLM scoring: {'Enabled' if llm_scoring_enabled else 'Disabled'}") + logger.info(f"Starting analysis with LLM scoring: {llm_scoring_enabled}") + + all_results = {} + question_numbers = [] + for q_id in selected_questions: + parts = q_id.split("_") + if len(parts) > 1: + try: + question_numbers.append(int(parts[1])) + except ValueError: + progress_text.warning(f"Invalid question ID format: {q_id}") + else: + progress_text.warning(f"Invalid question ID format: {q_id}") + + pre_retrieved_chunks = st.session_state.get("backend_chunks") + step_failed = False + async for result in analyzer.process_document( + file_path=file_path, + selected_questions=question_numbers, + use_llm_scoring=st.session_state.get("new_llm_scoring", False), + force_recompute=st.session_state.get("force_recompute", False), + pre_retrieved_chunks=pre_retrieved_chunks, + max_processing_step=max_processing_step, + ): + if "error" in result: + step_failed = True + if is_api_key_missing_message(result["error"]): + render_api_key_missing_alert(progress_text, result["error"]) + else: + progress_text.error(f"Error: {result['error']}") + continue + + if "status" in result: + progress_text.info(result["status"]) + continue + + question_id = result.get("question_id") + if not question_id: + question_number = result.get("question_number") + if question_number: + question_id = f"{st.session_state.question_set}_{question_number}" + else: + continue + + progress_text.info(f"Completed analysis for question {question_id}") + result_data = result.get("result", result) + all_results[question_id] = result_data + + final_results = analyzer.cache_manager.get_analysis( + file_path=file_path, config=config, question_ids=list(all_results.keys()) + ) + + if not final_results: + final_results = all_results + + logger.info(f"[ANALYSIS] Writing results to session state for file: {file_path}") + st.session_state.results = final_results + logger.info(f"[ANALYSIS] Attempting to display results for file: {file_path}") + + if partial: + if step_failed: + return + chunk_params = analyzer.analyzer.chunk_params + chunk_count, embedded_count = display_cached_document_chunks( + analyzer, + file_path, + chunk_size=chunk_params["chunk_size"], + chunk_overlap=chunk_params["chunk_overlap"], + ) + label = PROCESSING_STEP_LABELS.get(step_key, step_key) + if step_key == "embed": + if embedded_count == 0: + progress_text.error( + "Embed step did not produce embeddings. " + "Set a valid OPENAI_API_KEY in Settings (Embed uses OpenAI only, not Gemini)." + ) + else: + progress_text.success(f"Completed Embed step ({embedded_count}/{chunk_count} chunks with embeddings).") + elif chunk_count: + progress_text.success(f"Completed {label} step ({chunk_count} chunks).") + else: + progress_text.warning(f"Completed {label} step, but no chunks were found in cache.") + else: + progress_text.success("Analysis complete!") + + except Exception as e: + progress_text.error(f"Error during analysis: {e!s}") + logger.error(f"Error during analysis: {e!s}", exc_info=True) + + def main(): """Main function for the Streamlit app""" try: @@ -1445,8 +1446,11 @@ def main(): if "use_llm_scoring" not in st.session_state: st.session_state.use_llm_scoring = False # Default LLM scoring setting + if "force_recompute" not in st.session_state: + st.session_state.force_recompute = False # Default recompute setting + if "results" not in st.session_state: - st.session_state.results = {"answers": {}} + st.session_state.results = {} # Initialize empty results if "current_file" not in st.session_state: st.session_state.current_file = None # Initialize current file @@ -3829,22 +3833,62 @@ def main(): fresh_viewer_results = {} if analyze_clicked or reanalyze_clicked: - # NOW sync the selection state from the widget - # Get selected questions from the edited dataframe - selected_questions = edited_df.loc[edited_df["Select"], "QID"].tolist() + selected_questions = selected_question_ids_from_editor(edited_df) - # Update session state for individual question checkboxes (for backward compatibility) for q_id in questions.keys(): is_selected = q_id in selected_questions st.session_state[f"individual_question_{q_id}"] = is_selected - if not selected_questions: + max_step = get_max_processing_step() + needs_questions = processing_step_needs_questions() + + if needs_questions and not selected_questions: st.warning("Please select at least one question to analyze.") + elif is_partial_processing_step(max_step): + try: + update_analyzer_parameters() + st.session_state.force_recompute = reanalyze_clicked + progress_text = st.empty() + is_backend = st.session_state.get("backend_chunks") is not None + backend_uri = st.session_state.get("backend_resource_uri") + if is_backend and backend_uri: + analysis_file_path = backend_uri + else: + analysis_file_path = str(Path(file_path).resolve()) if file_path else file_path + progress_text.info(f"Running {max_step} step...") + asyncio.run( + run_analysis( + analyzer, + file_path=analysis_file_path, + selected_questions=[], + progress_text=progress_text, + max_processing_step=max_step, + ) + ) + except Exception as e: + st.error(f"Error analyzing document: {e!s}") else: try: - # Initialize progress display + st.session_state.force_recompute = reanalyze_clicked + + config = { + "chunk_size": st.session_state.new_chunk_size, + "chunk_overlap": st.session_state.new_overlap, + "top_k": st.session_state.new_top_k, + "model": st.session_state.get("new_llm_model", st.session_state.llm_model), + "question_set": st.session_state.new_question_set, + } + progress_text = st.empty() + is_backend = st.session_state.get("backend_chunks") is not None + backend_uri = st.session_state.get("backend_resource_uri") + + if is_backend and backend_uri: + analysis_file_path = backend_uri + else: + analysis_file_path = str(Path(file_path).resolve()) if file_path else file_path + if reanalyze_clicked: progress_text.info(f"Reanalyzing {len(selected_questions)} questions...") asyncio.run( @@ -3856,10 +3900,10 @@ def main(): use_llm_scoring=st.session_state.new_llm_scoring, single_call=st.session_state.new_batch_scoring, force_recompute=True, + max_processing_step=max_step, ) ) else: - # For normal analysis, check cache first cached_results = analyzer.analyzer.cache_manager.get_analysis( file_path=analysis_file_path, config=config, @@ -3867,13 +3911,11 @@ def main(): ) if cached_results: - # Process cached results - st.session_state.results["answers"].update(cached_results) + for question_id, result in cached_results.items(): + st.session_state.results["answers"][question_id] = result - # Generate file key for display file_key = generate_file_key(analysis_file_path, st) - # Update display analysis_df, chunks_df = create_analysis_dataframes( st.session_state.results["answers"], file_key, @@ -3882,19 +3924,18 @@ def main(): st.session_state.chunks_df = chunks_df st.session_state.analysis_complete = True else: - # Run analysis for uncached questions progress_text.info(f"Processing {len(selected_questions)} questions...") try: - # Run analysis for uncached questions asyncio.run( analyze_document_and_display( analyzer, - file_path=analysis_file_path, # Use URN for backend, file path for local + file_path=analysis_file_path, questions=questions, selected_questions=selected_questions, use_llm_scoring=st.session_state.new_llm_scoring, single_call=st.session_state.new_batch_scoring, + max_processing_step=max_step, ) ) @@ -3904,7 +3945,10 @@ def main(): st.error(f"Error during analysis: {e!s}") st.exception(e) - session_answers = st.session_state.results["answers"] + results_bag = st.session_state.get("results") or {} + session_answers = ( + results_bag.get("answers", results_bag) if isinstance(results_bag, dict) else {} + ) all_results = { question_id: session_answers[question_id] for question_id in selected_questions @@ -3927,28 +3971,33 @@ def main(): ) st.error(f"Error during analysis: {e!s}") - viewer_results = fresh_viewer_results - if not viewer_results: - viewer_results = analyzer.analyzer.cache_manager.get_analysis( - file_path=analysis_file_path, - config=config, - ) - if viewer_results: - raw_chunks = [] - elif is_backend: - raw_chunks = st.session_state.get("backend_chunks") or [] - else: - raw_chunks = analyzer.analyzer.cache_manager.get_document_chunks( - file_path=analysis_file_path, - chunk_size=config["chunk_size"], - chunk_overlap=config["chunk_overlap"], + try: + viewer_results = fresh_viewer_results + if not viewer_results: + viewer_results = analyzer.analyzer.cache_manager.get_analysis( + file_path=analysis_file_path, + config=config, + ) + if viewer_results: + raw_chunks = [] + elif is_backend: + raw_chunks = st.session_state.get("backend_chunks") or [] + else: + raw_chunks = analyzer.analyzer.cache_manager.get_document_chunks( + file_path=analysis_file_path, + chunk_size=config["chunk_size"], + chunk_overlap=config["chunk_overlap"], + ) + display_pdf_viewer( + str(analysis_file_path), + viewer_results, + questions, + raw_chunks, ) - display_pdf_viewer( - str(analysis_file_path), - viewer_results, - questions, - raw_chunks, - ) + except Exception as e: + logger.error(f"Error displaying PDF viewer: {e!s}", exc_info=True) + st.error(f"Error displaying PDF viewer: {e!s}") + else: # Show helpful error message if file_path is None: @@ -4303,26 +4352,13 @@ def main(): # Get the database identifier for the selected question set db_question_set = question_set_mapping.get(selected_set, selected_set) - # Get all available cache configurations - cache_configs = analyzer.analyzer.cache_manager.check_cache_status() - - # Group configurations by file for the selected question set - if cache_configs: - for config in cache_configs: - if len(config) == 6: - file_path, chunk_size, chunk_overlap, top_k, model, qs = config - if qs == db_question_set: - if file_path not in file_configs: - file_configs[file_path] = [] - file_configs[file_path].append( - { - "chunk_size": chunk_size, - "chunk_overlap": chunk_overlap, - "top_k": top_k, - "model": model, - "question_set": qs, - } - ) + # Get all available cache + document-chunk configurations + file_configs = build_all_results_file_configs( + analyzer.analyzer.cache_manager, + db_question_set, + default_top_k=st.session_state.get("top_k", 10), + default_model=st.session_state.get("llm_model", DEFAULT_LLM_MODEL), + ) with col2: # Report selector in green container diff --git a/tests/conftest.py b/tests/conftest.py index c8cc7989..f87eb268 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -74,6 +74,17 @@ def pytest_configure(config): # Register custom markers config.addinivalue_line("markers", "postgres: mark test as requiring PostgreSQL") config.addinivalue_line("markers", "integration: mark test as integration test") + os.environ.setdefault("MPLBACKEND", "Agg") + + +@pytest.fixture(autouse=True) +def _reset_document_analyzer_singleton(): + """Keep DocumentAnalyzer from leaking cache/LLM state across tests.""" + from report_analyst.core.analyzer import DocumentAnalyzer + + DocumentAnalyzer.reset_instance() + yield + DocumentAnalyzer.reset_instance() # ============================================================================= @@ -278,6 +289,7 @@ async def fake_analyze_document( single_call=True, force_recompute=False, pre_retrieved_chunks=None, + max_processing_step="answer", ): calls.append( { @@ -288,6 +300,7 @@ async def fake_analyze_document( "single_call": single_call, "force_recompute": force_recompute, "pre_retrieved_chunks": pre_retrieved_chunks, + "max_processing_step": max_processing_step, } ) @@ -319,6 +332,7 @@ async def fake_process_document( single_call=True, force_recompute=False, pre_retrieved_chunks=None, + max_processing_step="answer", ): calls.append( { @@ -328,6 +342,7 @@ async def fake_process_document( "single_call": single_call, "force_recompute": force_recompute, "pre_retrieved_chunks": pre_retrieved_chunks, + "max_processing_step": max_processing_step, } ) diff --git a/tests/test_all_results_file_configs.py b/tests/test_all_results_file_configs.py new file mode 100644 index 00000000..d32c4c52 --- /dev/null +++ b/tests/test_all_results_file_configs.py @@ -0,0 +1,260 @@ +"""Tests for All Results file listing and chunk resolution.""" + +import pandas as pd +import pytest + +from report_analyst.core.cache_manager import CacheManager +from report_analyst.streamlit_app import ( + build_all_results_file_configs, + selected_question_ids_from_editor, +) + + +def test_build_all_results_file_configs_includes_chunk_only_reports(tmp_path): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + befesa = str(tmp_path / "Befesa_Annual_Report_2025.pdf") + sustainability = str(tmp_path / "sustainability_report2024.pdf") + + cache.save_text_only_chunks( + file_path=befesa, + chunks=[{"text": "Chunk A", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + cache.save_analysis( + file_path=sustainability, + question_id="tcfd_1", + result={"ANSWER": "Yes", "SCORE": 1.0}, + config={ + "chunk_size": 500, + "chunk_overlap": 20, + "top_k": 5, + "model": "gpt-4o-mini", + "question_set": "tcfd", + }, + ) + + configs = build_all_results_file_configs( + cache, + "tcfd", + default_top_k=5, + default_model="gpt-4o-mini", + ) + + assert befesa in configs + assert configs[befesa][0]["chunks_only"] is True + assert sustainability in configs + assert not configs[sustainability][0].get("chunks_only") + + +def test_resolve_document_chunks_matches_by_filename(tmp_path): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + stored_path = str(tmp_path / "storage" / "uploads" / "Befesa_Annual_Report_2025.pdf") + lookup_path = str(tmp_path / "temp" / "Befesa_Annual_Report_2025.pdf") + + cache.save_text_only_chunks( + file_path=stored_path, + chunks=[{"text": "Same report, different path.", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + + chunks = cache.resolve_document_chunks(lookup_path, chunk_size=500, chunk_overlap=20) + assert len(chunks) == 1 + assert chunks[0]["text"] == "Same report, different path." + + +def test_build_all_results_skips_malformed_and_dedupes(tmp_path): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + path = str(tmp_path / "report.pdf") + cache.save_text_only_chunks( + file_path=path, + chunks=[{"text": "Chunk A", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + cache.save_analysis( + file_path=path, + question_id="tcfd_1", + result={"ANSWER": "Yes", "SCORE": 1.0}, + config={ + "chunk_size": 500, + "chunk_overlap": 20, + "top_k": 5, + "model": "gpt-4o-mini", + "question_set": "tcfd", + }, + ) + + original = cache.check_cache_status + + def fake_status(): + rows = list(original()) + rows.append(("bad",)) + return rows + + cache.check_cache_status = fake_status + configs = build_all_results_file_configs(cache, "tcfd", default_top_k=5, default_model="gpt-4o-mini") + assert path in configs + assert len(configs[path]) == 1 + assert not configs[path][0].get("chunks_only") + + +def test_resolve_document_chunks_skips_mismatched_size_and_name(tmp_path): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + stored = str(tmp_path / "storage" / "report.pdf") + other = str(tmp_path / "storage" / "other.pdf") + lookup = str(tmp_path / "temp" / "report.pdf") + cache.save_text_only_chunks( + file_path=stored, + chunks=[{"text": "ok", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + cache.save_text_only_chunks( + file_path=other, + chunks=[{"text": "nope", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + assert cache.resolve_document_chunks(lookup, chunk_size=999, chunk_overlap=20) == [] + assert cache.resolve_document_chunks(str(tmp_path / "temp" / "missing.pdf"), chunk_size=500, chunk_overlap=20) == [] + assert len(cache.resolve_document_chunks(lookup, chunk_size=500, chunk_overlap=20)) == 1 + + +def test_list_document_chunk_configs_returns_empty_on_error(tmp_path, monkeypatch): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + + class Boom: + def __enter__(self): + raise RuntimeError("db down") + + def __exit__(self, *args): + return False + + monkeypatch.setattr(cache.db_manager, "get_connection", lambda: Boom()) + assert cache.list_document_chunk_configs() == [] + + +def test_save_text_only_chunks_raises_on_error(tmp_path, monkeypatch): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + + class Boom: + def __enter__(self): + raise RuntimeError("db down") + + def __exit__(self, *args): + return False + + monkeypatch.setattr(cache.db_manager, "get_connection", lambda: Boom()) + with pytest.raises(RuntimeError, match="db down"): + cache.save_text_only_chunks( + file_path=str(tmp_path / "report.pdf"), + chunks=[{"text": "x", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + + +def test_selected_question_ids_from_editor_helpers(): + assert selected_question_ids_from_editor(pd.DataFrame()) == [] + assert selected_question_ids_from_editor(pd.DataFrame({"QID": ["a"]})) == [] + df = pd.DataFrame({"Select": [True, False, False], "QID": ["a", "b", "c"]}) + assert selected_question_ids_from_editor(df) == ["a"] + + +def test_resolve_document_chunks_direct_hit(tmp_path): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + path = str(tmp_path / "report.pdf") + cache.save_text_only_chunks( + file_path=path, + chunks=[{"text": "direct", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + chunks = cache.resolve_document_chunks(path, chunk_size=500, chunk_overlap=20) + assert len(chunks) == 1 + assert chunks[0]["text"] == "direct" + + +def test_resolve_document_chunks_skips_overlap_mismatch(tmp_path): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + stored = str(tmp_path / "storage" / "report.pdf") + lookup = str(tmp_path / "temp" / "report.pdf") + cache.save_text_only_chunks( + file_path=stored, + chunks=[{"text": "ok", "metadata": {}}], + chunk_size=500, + chunk_overlap=20, + ) + assert cache.resolve_document_chunks(lookup, chunk_size=500, chunk_overlap=99) == [] + + +def test_save_text_only_chunks_postgres_path(tmp_path, monkeypatch): + cache = CacheManager(db_path=str(tmp_path / "cache.db")) + monkeypatch.setattr(cache.db_manager, "is_postgres", lambda: True) + executed = [] + + class FakeConn: + def execute(self, statement, params=None): + executed.append((str(statement), params)) + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + monkeypatch.setattr(cache.db_manager, "get_connection", lambda: FakeConn()) + cache.save_text_only_chunks( + file_path=str(tmp_path / "report.pdf"), + chunks=[{"text": "pg", "metadata": {"page": 1}}], + chunk_size=500, + chunk_overlap=20, + ) + assert executed + assert "ON CONFLICT" in executed[0][0] + + +def test_display_consolidated_results_single_config(monkeypatch): + from report_analyst import streamlit_app as app + + analyzer = type("A", (), {})() + analyzer.analyzer = type("Inner", (), {})() + analyzer.analyzer.cache_manager = object() + called = {} + + def fake_render(a, qs, fp, config, display_analysis_results=None, display_pdf_viewer=None): + called["args"] = (qs, fp, config) + + monkeypatch.setattr(app, "render_consolidated_report_view", fake_render) + monkeypatch.setattr(app.st, "session_state", {"top_k": 5, "llm_model": "gpt-4o-mini"}) + app.display_consolidated_results( + analyzer, + "tcfd", + file_path="report.pdf", + selected_config={"config": {"chunk_size": 500, "chunk_overlap": 20}}, + ) + assert called["args"][0] == "tcfd" + assert called["args"][1] == "report.pdf" + + +def test_display_consolidated_results_empty_configs(monkeypatch): + from report_analyst import streamlit_app as app + + warnings = [] + analyzer = type("A", (), {})() + analyzer.analyzer = type("Inner", (), {})() + analyzer.analyzer.cache_manager = object() + monkeypatch.setattr(app, "build_all_results_file_configs", lambda *a, **k: {}) + monkeypatch.setattr(app.st, "session_state", {"top_k": 5, "llm_model": "gpt-4o-mini"}) + monkeypatch.setattr(app.st, "warning", lambda m: warnings.append(m)) + app.display_consolidated_results(analyzer, "tcfd") + assert warnings + + +def test_is_partial_processing_step_reads_session(monkeypatch): + from report_analyst import streamlit_app as app + + monkeypatch.setattr(app.st, "session_state", {"processing_steps_slider": "Chunk"}) + assert app.is_partial_processing_step() is True diff --git a/tests/test_consolidated_chunk_search.py b/tests/test_consolidated_chunk_search.py new file mode 100644 index 00000000..dd693874 --- /dev/null +++ b/tests/test_consolidated_chunk_search.py @@ -0,0 +1,249 @@ +"""Tests for All Results chunk search modes.""" + +from unittest.mock import Mock + +import numpy as np +import pytest + +from report_analyst.consolidated_results_view import ( + render_consolidated_chunk_search, + render_consolidated_report_view, +) + + +@pytest.fixture +def mock_analyzer(): + analyzer = Mock() + analyzer.analyzer = Mock() + analyzer.analyzer.question_set = "tcfd" + analyzer.analyzer.questions = {"tcfd_1": {"text": "What are Scope 1 emissions?"}} + analyzer.analyzer.use_backend_llm = False + analyzer.analyzer.embeddings = Mock() + analyzer.analyzer.embeddings.get_text_embedding.return_value = [1.0, 0.0] + analyzer.analyzer._ensure_embeddings_client = Mock() + analyzer.analyzer.update_question_set = Mock() + analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [] + analyzer.analyzer.cache_manager.get_analysis.return_value = None + return analyzer + + +def _patch_streamlit(monkeypatch, *, selectbox="None", text_input=""): + subheaders: list[str] = [] + captions: list[str] = [] + infos: list[str] = [] + warnings: list[str] = [] + errors: list[str] = [] + successes: list[str] = [] + col = Mock() + col.__enter__ = Mock(return_value=col) + col.__exit__ = Mock(return_value=False) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.subheader", lambda t: subheaders.append(t)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.caption", lambda t: captions.append(t)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.columns", lambda n: (col, col)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.selectbox", Mock(return_value=selectbox)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.text_input", Mock(return_value=text_input)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.dataframe", Mock()) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.info", lambda t: infos.append(t)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.warning", lambda t: warnings.append(t)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.error", lambda t: errors.append(t)) + monkeypatch.setattr("report_analyst.consolidated_results_view.st.success", lambda t: successes.append(t)) + monkeypatch.setattr( + "report_analyst.consolidated_results_view.st.column_config", + Mock(NumberColumn=Mock, TextColumn=Mock, CheckboxColumn=Mock), + ) + return { + "subheaders": subheaders, + "captions": captions, + "infos": infos, + "warnings": warnings, + "errors": errors, + "successes": successes, + } + + +def _embedded_chunk(vec=(1.0, 0.0)): + return { + "text": "Scope 1 disclosure.", + "embedding": np.array(vec, dtype=np.float32).tobytes(), + "chunk_size": 500, + "chunk_overlap": 20, + } + + +def test_text_only_chunks_skip_similarity_controls(mock_analyzer, monkeypatch): + text_only = [{"text": "Scope 1 disclosure.", "embedding": None, "chunk_size": 500, "chunk_overlap": 20}] + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = text_only + ui = _patch_streamlit(monkeypatch) + monkeypatch.setattr( + "report_analyst.consolidated_results_view.st.selectbox", + Mock(side_effect=AssertionError("no similarity UI")), + ) + monkeypatch.setattr( + "report_analyst.consolidated_results_view.st.text_input", + Mock(side_effect=AssertionError("no similarity UI")), + ) + + assert render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + assert ui["subheaders"] == ["Document Chunks"] + assert any("Embed" in c for c in ui["captions"]) + + +def test_embedded_chunks_show_similarity_search(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [_embedded_chunk()] + ui = _patch_streamlit(monkeypatch) + assert render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + assert ui["subheaders"] == ["Similarity Search"] + + +def test_empty_chunks_warns_and_returns_false(mock_analyzer, monkeypatch): + ui = _patch_streamlit(monkeypatch) + assert not render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + assert ui["subheaders"] == ["Document Chunks"] + assert ui["warnings"] + + +def test_similarity_ranks_with_selected_question(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.question_set = "other" + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [ + _embedded_chunk((1.0, 0.0)), + _embedded_chunk((0.0, 1.0)), + ] + ui = _patch_streamlit(monkeypatch, selectbox="tcfd_1", text_input="") + assert render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + mock_analyzer.analyzer.update_question_set.assert_called_once_with("tcfd") + assert ui["successes"] + assert any("Using question" in i for i in ui["infos"]) + assert any("tcfd_1" in c for c in ui["captions"]) + + +def test_similarity_ranks_with_custom_question(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [_embedded_chunk()] + ui = _patch_streamlit(monkeypatch, selectbox="None", text_input=" custom query ") + assert render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + assert any("custom question" in i for i in ui["infos"]) + assert ui["successes"] + + +def test_similarity_backend_llm_warns(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.use_backend_llm = True + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [_embedded_chunk()] + ui = _patch_streamlit(monkeypatch, selectbox="tcfd_1") + assert render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + assert any("backend LLM" in w for w in ui["warnings"]) + + +def test_similarity_runtime_error_warns(mock_analyzer, monkeypatch): + mock_analyzer.analyzer._ensure_embeddings_client.side_effect = RuntimeError("no key") + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [_embedded_chunk()] + ui = _patch_streamlit(monkeypatch, selectbox="tcfd_1") + assert render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + assert "no key" in ui["warnings"][0] + + +def test_similarity_generic_exception_errors(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.embeddings.get_text_embedding.side_effect = ValueError("boom") + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [_embedded_chunk()] + ui = _patch_streamlit(monkeypatch, selectbox="tcfd_1") + assert render_consolidated_chunk_search(mock_analyzer, "tcfd", "report.pdf", {"chunk_size": 500, "chunk_overlap": 20}) + assert any("boom" in e for e in ui["errors"]) + + +def test_report_view_chunks_only_without_answers(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [ + {"text": "x", "embedding": None, "chunk_size": 500, "chunk_overlap": 20} + ] + ui = _patch_streamlit(monkeypatch) + display = Mock() + render_consolidated_report_view( + mock_analyzer, + "tcfd", + "report.pdf", + {"chunk_size": 500, "chunk_overlap": 20, "chunks_only": True}, + display_analysis_results=display, + ) + assert any("No answer results" in i for i in ui["infos"]) + display.assert_not_called() + + +def test_report_view_warns_when_nothing_cached(mock_analyzer, monkeypatch): + ui = _patch_streamlit(monkeypatch) + render_consolidated_report_view( + mock_analyzer, + "tcfd", + "report.pdf", + {"chunk_size": 500, "chunk_overlap": 20}, + display_analysis_results=Mock(), + ) + assert any("No stored results" in w for w in ui["warnings"]) + + +def test_report_view_handles_chunk_search_exception(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.side_effect = RuntimeError("cache down") + ui = _patch_streamlit(monkeypatch) + render_consolidated_report_view( + mock_analyzer, + "tcfd", + "report.pdf", + {"chunk_size": 500, "chunk_overlap": 20, "chunks_only": True}, + display_analysis_results=Mock(), + ) + assert any("No answer results" in i for i in ui["infos"]) + + +def test_report_view_renders_cached_analysis(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.question_set = "other" + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [ + {"text": "x", "embedding": None, "chunk_size": 500, "chunk_overlap": 20} + ] + mock_analyzer.analyzer.cache_manager.get_analysis.return_value = { + "tcfd_1": { + "result": { + "ANSWER": "Yes", + "SCORE": "1.5", + "EVIDENCE": [{"text": "ev"}], + "GAPS": ["gap"], + "SOURCES": [1, 2], + }, + "chunks": [ + { + "text": "chunk", + "similarity_score": 0.9, + "llm_score": 0.8, + "is_evidence": True, + "chunk_order": 1, + } + ], + }, + "bad": {"result": None}, + } + _patch_streamlit(monkeypatch) + display = Mock() + render_consolidated_report_view( + mock_analyzer, + "tcfd", + "report.pdf", + {"chunk_size": 500, "chunk_overlap": 20}, + display_analysis_results=display, + ) + mock_analyzer.analyzer.update_question_set.assert_called_with("tcfd") + display.assert_called_once() + analysis_df, chunks_df, file_key = display.call_args[0] + assert file_key == "report_cs500" + assert list(analysis_df["Question ID"]) == ["tcfd_1"] + assert not chunks_df.empty + + +def test_report_view_warns_when_all_results_fail(mock_analyzer, monkeypatch): + mock_analyzer.analyzer.cache_manager.resolve_document_chunks.return_value = [ + {"text": "x", "embedding": None, "chunk_size": 500, "chunk_overlap": 20} + ] + mock_analyzer.analyzer.cache_manager.get_analysis.return_value = {"bad": {"result": None}} + ui = _patch_streamlit(monkeypatch) + render_consolidated_report_view( + mock_analyzer, + "tcfd", + "report.pdf", + {"chunk_size": 500, "chunk_overlap": 20}, + display_analysis_results=Mock(), + ) + assert any("No results found" in w for w in ui["warnings"]) diff --git a/tests/test_llm_providers.py b/tests/test_llm_providers.py index 0ebaea0e..f1ebf0e2 100644 --- a/tests/test_llm_providers.py +++ b/tests/test_llm_providers.py @@ -31,3 +31,15 @@ def test_get_llm_requires_google_api_key(monkeypatch): with pytest.raises(ValueError, match="GOOGLE_API_KEY"): llm_providers.get_llm("gemini-2.5-flash") + + +def test_get_llm_openai_uses_fail_fast_retries(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test") + openai_cls = Mock() + monkeypatch.setattr(llm_providers, "OpenAI", openai_cls) + + llm_providers.get_llm("gpt-4o-mini") + + kwargs = openai_cls.call_args.kwargs + assert kwargs["timeout"] == llm_providers.OPENAI_REQUEST_TIMEOUT_SECONDS + assert kwargs["max_retries"] == llm_providers.OPENAI_MAX_RETRIES diff --git a/tests/test_openai_quota_fail_fast.py b/tests/test_openai_quota_fail_fast.py new file mode 100644 index 00000000..ad7dbd4e --- /dev/null +++ b/tests/test_openai_quota_fail_fast.py @@ -0,0 +1,146 @@ +"""Prove insufficient_quota is a permanent error, not a hang/retry loop.""" + +from __future__ import annotations + +import time +from unittest.mock import AsyncMock, Mock, patch + +import httpx +import numpy as np +import pytest +from openai import RateLimitError + +from report_analyst.core.analyzer import DocumentAnalyzer +from report_analyst.core.cache_manager import CacheManager +from report_analyst.core.llm_providers import ( + OPENAI_MAX_RETRIES, + OPENAI_REQUEST_TIMEOUT_SECONDS, + build_openai_embedding, + is_permanent_openai_quota_error, +) + +QUOTA_BODY = { + "error": { + "message": "You have no credits remaining.", + "type": "insufficient_quota", + "param": None, + "code": "credit_balance_exhausted", + } +} + + +def _quota_response() -> httpx.Response: + request = httpx.Request("POST", "https://api.openai.com/v1/embeddings") + return httpx.Response(429, json=QUOTA_BODY, request=request) + + +def _quota_error() -> RateLimitError: + response = _quota_response() + return RateLimitError("Error code: 429", response=response, body=QUOTA_BODY) + + +def _count_embed_http_calls(embed_client) -> tuple[int, str]: + calls = {"n": 0} + + def handler(_request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + if calls["n"] > 500: + raise AssertionError("quota 429 retried more than 500 times (unbounded)") + return _quota_response() + + embed_client._http_client = httpx.Client(transport=httpx.MockTransport(handler)) + embed_client._client = None + try: + embed_client.get_text_embedding("hello") + status = "ok" + except AssertionError: + raise + except Exception as exc: # noqa: BLE001 + status = type(exc).__name__ + return calls["n"], status + + +def test_is_permanent_openai_quota_error_detects_credit_balance(): + assert is_permanent_openai_quota_error(_quota_error()) is True + assert is_permanent_openai_quota_error(RuntimeError("network down")) is False + + +def test_default_embedding_quota_retries_are_bounded_not_endless(monkeypatch): + """Hypothesis: stock OpenAIEmbedding retries quota 429 a finite number of times, then raises. + + Sleep is no-op so tenacity's 60s stop-after-delay does not cut the loop. + That yields the full nested budget (10 LlamaIndex attempts x 11 OpenAI tries). + Real sleep still stops (measured ~22 HTTP calls in ~93s), not an infinite loop. + """ + from llama_index.embeddings.openai import OpenAIEmbedding + + slept: list[float] = [] + monkeypatch.setattr(time, "sleep", lambda seconds: slept.append(float(seconds))) + embed = OpenAIEmbedding(api_key="sk-test", timeout=5.0, reuse_client=True) + calls, status = _count_embed_http_calls(embed) + assert status == "RateLimitError" + assert calls == 110 + assert len(slept) == 109 + assert sum(slept) > 0 + + +def test_build_openai_embedding_quota_exhausted_makes_one_http_call(monkeypatch): + """Hypothesis: our client fails on the first insufficient_quota response.""" + monkeypatch.setattr(time, "sleep", lambda *_args, **_kwargs: None) + embed = build_openai_embedding(api_key="sk-test") + assert embed.timeout == OPENAI_REQUEST_TIMEOUT_SECONDS + assert embed.max_retries == OPENAI_MAX_RETRIES + calls, status = _count_embed_http_calls(embed) + assert status == "RateLimitError" + assert calls == 1 + + +@pytest.fixture +def analyzer(monkeypatch, tmp_path): + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + monkeypatch.setenv("USE_BACKEND", "false") + monkeypatch.setenv("USE_CENTRALIZED_LLM", "false") + DocumentAnalyzer.reset_instance() + with patch("report_analyst.core.llm_providers.build_openai_embedding"), patch( + "report_analyst.core.llm_providers.get_llm" + ) as mock_get_llm: + mock_get_llm.return_value = Mock(model="gpt-4o-mini", achat=AsyncMock()) + doc_analyzer = DocumentAnalyzer() + doc_analyzer.cache_manager = CacheManager(db_path=str(tmp_path / "quota.db")) + doc_analyzer.llm = Mock(model="gpt-4o-mini", achat=AsyncMock()) + doc_analyzer.use_backend_llm = False + yield doc_analyzer + DocumentAnalyzer.reset_instance() + + +@pytest.mark.asyncio +async def test_process_document_map_quota_exhausted_yields_error(analyzer, tmp_path): + """Hypothesis: Map surfaces quota errors instead of completing with empty ranks.""" + file_path = str(tmp_path / "quota-map.pdf") + embedded = [ + { + "text": "Scope 1 emissions disclosure.", + "metadata": {"page": 1}, + "embedding": np.array([0.1, 0.2, 0.3], dtype=np.float32), + } + ] + analyzer.embeddings = Mock() + analyzer.embeddings.get_text_embedding.side_effect = _quota_error() + analyzer.cache_manager = CacheManager(db_path=str(analyzer.cache_manager.db_path)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded), patch.object( + analyzer, + "get_question_by_number", + return_value={"text": "What are Scope 1 emissions?", "guidelines": ""}, + ): + events = [] + async for event in analyzer.process_document( + file_path, + selected_questions=[1], + max_processing_step="map", + ): + events.append(event) + + assert analyzer.embeddings.get_text_embedding.call_count == 1 + assert any("error" in event for event in events) + assert not any("Completed Map" in event.get("status", "") for event in events) diff --git a/tests/test_pdf_viewer.py b/tests/test_pdf_viewer.py index e0a2d7ba..2e38e908 100644 --- a/tests/test_pdf_viewer.py +++ b/tests/test_pdf_viewer.py @@ -138,3 +138,20 @@ def test_display_pdf_viewer_opens_without_analysis(): component.assert_called_once() assert component.call_args.kwargs["chunks_data"] == {} assert component.call_args.kwargs["unmapped_chunks"] == [] + + +def test_display_pdf_viewer_survives_component_errors(): + with ( + patch("report_analyst.streamlit_app.st.expander", return_value=nullcontext()), + patch("report_analyst.streamlit_app.pdf_viewer", side_effect=RuntimeError("boom")), + patch("report_analyst.streamlit_app.st.error") as error, + ): + display_pdf_viewer( + file_path="report.pdf", + results=None, + questions=None, + raw_chunks=None, + ) + + error.assert_called_once() + assert "Error rendering PDF viewer" in error.call_args.args[0] diff --git a/tests/test_processing_steps.py b/tests/test_processing_steps.py new file mode 100644 index 00000000..d6c20dc1 --- /dev/null +++ b/tests/test_processing_steps.py @@ -0,0 +1,1181 @@ +"""Tests for document processing step pipeline (Chunk / Embed / Map / Answer).""" + +from __future__ import annotations + +import shutil +import tempfile +import uuid +from pathlib import Path +from unittest.mock import AsyncMock, Mock, patch + +import numpy as np +import pytest + +from report_analyst.core.analyzer import ( + DocumentAnalyzer, + normalize_processing_step, + processing_step_rank, +) +from report_analyst.core.cache_manager import CacheManager + + +class FakeSessionState(dict): + def get(self, key, default=None): + return super().get(key, default) + + def __getattr__(self, name): + try: + return self[name] + except KeyError as exc: + raise AttributeError(name) from exc + + def __setattr__(self, name, value): + self[name] = value + + +def make_run_analysis_session(**overrides): + base = { + "chunk_size": 500, + "chunk_overlap": 20, + "top_k": 5, + "llm_model": "gpt-4o-mini", + "question_set": "tcfd", + "force_recompute": False, + "new_llm_scoring": False, + "new_chunk_size": 500, + "new_overlap": 20, + "new_top_k": 5, + "new_llm_model": "gpt-4o-mini", + "new_question_set": "tcfd", + } + base.update(overrides) + return FakeSessionState(base) + + +def patch_streamlit_app(monkeypatch, **session_overrides): + import report_analyst.streamlit_app as app + + fake_st = make_run_analysis_session(**session_overrides) + monkeypatch.setattr(app, "st", Mock(session_state=fake_st)) + return app, fake_st + + +async def _async_status_events(*statuses): + for status in statuses: + yield {"status": status} + + +@pytest.fixture(autouse=True) +def _isolate_streamlit_module(monkeypatch): + """Reset streamlit_app.st between tests so random order does not leak mocks.""" + import streamlit as real_streamlit + + import report_analyst.streamlit_app as app + + monkeypatch.setattr(app, "st", real_streamlit, raising=False) + + +@pytest.fixture(scope="session") +def _processing_steps_db_template(): + temp_dir = tempfile.mkdtemp() + db_path = Path(temp_dir) / "analysis_template.db" + CacheManager(str(db_path)) + yield db_path + shutil.rmtree(temp_dir) + + +@pytest.fixture +def clean_db(_processing_steps_db_template): + temp_dir = tempfile.mkdtemp() + db_path = Path(temp_dir) / f"analysis_{uuid.uuid4().hex}.db" + shutil.copy2(_processing_steps_db_template, db_path) + yield db_path + shutil.rmtree(temp_dir) + + +@pytest.fixture +def analyzer(monkeypatch, clean_db): + """Fresh DocumentAnalyzer per test — singleton otherwise leaks cache/chunks across cases.""" + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + monkeypatch.setenv("OPENAI_ORGANIZATION", "test-org") + monkeypatch.setenv("USE_BACKEND", "false") + monkeypatch.setenv("USE_CENTRALIZED_LLM", "false") + DocumentAnalyzer.reset_instance() + with patch("report_analyst.core.llm_providers.build_openai_embedding"), patch( + "report_analyst.core.llm_providers.get_llm" + ) as mock_get_llm: + mock_get_llm.return_value = Mock(model="gpt-4o-mini", achat=AsyncMock()) + doc_analyzer = DocumentAnalyzer() + doc_analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + doc_analyzer.llm = Mock(model="gpt-4o-mini", achat=AsyncMock()) + doc_analyzer.use_backend_llm = False + yield doc_analyzer + DocumentAnalyzer.reset_instance() + + +async def _collect_process_events(analyzer, **kwargs): + events = [] + async for event in analyzer.process_document(**kwargs): + events.append(event) + return events + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("Chunk", "chunk"), + ("Embed", "embed"), + ("Map", "map"), + ("Answer", "answer"), + ("chunk", "chunk"), + (None, "answer"), + ("", "answer"), + ], +) +def test_normalize_processing_step_maps_ui_labels(raw, expected): + assert normalize_processing_step(raw) == expected + + +def test_processing_step_rank_order(): + assert processing_step_rank("chunk") < processing_step_rank("embed") + assert processing_step_rank("embed") < processing_step_rank("map") + assert processing_step_rank("map") < processing_step_rank("answer") + + +def test_save_text_only_chunks_persists_rows_without_embeddings(clean_db, tmp_path): + cache = CacheManager(db_path=str(clean_db)) + file_path = str(tmp_path / "report.pdf") + chunks = [{"text": "Board oversight section.", "metadata": {"page": 1}}] + + cache.save_text_only_chunks( + file_path=file_path, + chunks=chunks, + chunk_size=500, + chunk_overlap=20, + ) + + stored = cache.get_chunks_without_embeddings(file_path, chunk_size=500, chunk_overlap=20) + assert len(stored) == 1 + assert stored[0]["text"] == "Board oversight section." + assert stored[0]["embedding"] is None + + +@pytest.mark.asyncio +async def test_process_document_chunk_step_stops_before_llm(analyzer, clean_db, tmp_path): + """Hypothesis: Chunk-only must not invoke chat LLM even when questions are selected.""" + file_path = str(tmp_path / "chunk-only.pdf") + text_chunks = [{"text": "Climate risk paragraph.", "metadata": {"page": 1}, "embedding": None}] + + analyzer.llm = Mock(achat=AsyncMock()) + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer, "_create_text_chunks", return_value=text_chunks): + events = [] + async for event in analyzer.process_document( + file_path, + selected_questions=[1], + max_processing_step="chunk", + ): + events.append(event) + + analyzer.llm.achat.assert_not_called() + statuses = [e.get("status", "") for e in events if "status" in e] + assert any("Completed Chunk" in s for s in statuses) + assert not any("error" in e for e in events) + + +@pytest.mark.asyncio +async def test_process_document_chunk_step_reuses_cached_chunks(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "cached-chunk.pdf") + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + analyzer.cache_manager.save_text_only_chunks( + file_path=file_path, + chunks=[{"text": "From cache.", "metadata": {"page": 1}}], + chunk_size=500, + chunk_overlap=20, + ) + + with patch.object(analyzer, "_create_text_chunks", side_effect=AssertionError("should not re-chunk")): + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="chunk", + ) + + assert any("Completed Chunk" in e.get("status", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_chunk_step_uses_text_chunks_not_embedded_create(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "new-chunk.pdf") + text_chunks = [{"text": "Fresh chunk.", "metadata": {}, "embedding": None}] + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer, "_create_text_chunks", return_value=text_chunks) as mock_text, patch.object( + analyzer, "_create_chunks", side_effect=AssertionError("chunk step must not embed") + ): + await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="chunk", + ) + + mock_text.assert_called_once() + + +@pytest.mark.asyncio +async def test_process_document_embed_step_skips_reembed_when_already_embedded(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "already-embedded.pdf") + embedded = [ + { + "text": "Has embedding.", + "metadata": {}, + "embedding": np.array([0.1, 0.2], dtype=np.float32), + } + ] + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded), patch.object( + analyzer, "_add_embeddings_to_chunks", side_effect=AssertionError("should not re-embed") + ): + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="embed", + ) + + assert any("Completed Embed" in e.get("status", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_answer_step_requires_questions(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "answer-no-questions.pdf") + embedded = [ + { + "text": "Embedded chunk.", + "metadata": {}, + "embedding": np.array([0.2, 0.3], dtype=np.float32), + } + ] + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded): + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="answer", + ) + + assert any("Select at least one question" in e.get("error", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_map_step_skips_answer_llm(analyzer, clean_db, tmp_path): + """Hypothesis: Map step ranks chunks but must not call _analyze_chunks / achat.""" + file_path = str(tmp_path / "map-only.pdf") + embedded = [ + { + "text": "Scope 1 emissions disclosure.", + "metadata": {"page": 2}, + "embedding": np.array([0.1, 0.2, 0.3], dtype=np.float32), + } + ] + + analyzer.llm = Mock(achat=AsyncMock()) + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded), patch.object( + analyzer, "_get_similar_chunks", AsyncMock(return_value=embedded) + ), patch.object(analyzer, "_analyze_chunks", AsyncMock()) as mock_analyze, patch.object( + analyzer, "get_question_by_number", return_value={"text": "What are Scope 1 emissions?", "guidelines": ""} + ): + events = [] + async for event in analyzer.process_document( + file_path, + selected_questions=[1], + max_processing_step="map", + ): + events.append(event) + + mock_analyze.assert_not_called() + analyzer.llm.achat.assert_not_called() + assert any("Completed Map" in e.get("status", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_embed_step_upgrades_text_chunks(analyzer, clean_db, tmp_path): + """Hypothesis: Embed step adds embeddings to text-only cache rows and stops before questions.""" + file_path = str(tmp_path / "embed-only.pdf") + text_only = [{"text": "Paragraph awaiting embeddings.", "metadata": {"page": 1}, "embedding": None}] + embedded = [ + { + "text": "Paragraph awaiting embeddings.", + "metadata": {"page": 1}, + "embedding": np.array([0.4, 0.5, 0.6], dtype=np.float32), + } + ] + + analyzer.llm = Mock(achat=AsyncMock()) + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=text_only), patch.object( + analyzer, "_add_embeddings_to_chunks", return_value=embedded + ) as mock_embed, patch.object(analyzer.cache_manager, "save_document_chunks") as mock_save, patch.object( + analyzer, "_analyze_chunks", AsyncMock() + ) as mock_analyze: + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[1], + max_processing_step="embed", + ) + + mock_embed.assert_called_once_with(text_only) + mock_save.assert_called_once() + mock_analyze.assert_not_called() + analyzer.llm.achat.assert_not_called() + assert any("Completed Embed" in e.get("status", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_chunk_step_allows_empty_question_list(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "chunk-no-questions.pdf") + text_chunks = [{"text": "No questions needed.", "metadata": {}, "embedding": None}] + + analyzer.llm = Mock(achat=AsyncMock()) + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer, "_create_text_chunks", return_value=text_chunks): + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="chunk", + ) + + assert not any("error" in e for e in events) + assert any("Completed Chunk" in e.get("status", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_embed_step_allows_empty_question_list(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "embed-no-questions.pdf") + embedded = [ + { + "text": "Already embedded.", + "metadata": {}, + "embedding": np.array([0.1, 0.2], dtype=np.float32), + } + ] + + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded): + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="embed", + ) + + assert not any("error" in e for e in events) + assert any("Completed Embed" in e.get("status", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_map_step_requires_questions(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "map-no-questions.pdf") + embedded = [ + { + "text": "Embedded chunk.", + "metadata": {}, + "embedding": np.array([0.2, 0.3], dtype=np.float32), + } + ] + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded): + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="map", + ) + + assert any("Select at least one question" in e.get("error", "") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_answer_step_calls_analyze_chunks(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "full-answer.pdf") + embedded = [ + { + "text": "Evidence paragraph.", + "metadata": {"page": 3}, + "embedding": np.array([0.7, 0.8], dtype=np.float32), + } + ] + analyze_result = {"ANSWER": "Yes", "SCORE": 0.9, "EVIDENCE": []} + + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded), patch.object( + analyzer, "_get_similar_chunks", AsyncMock(return_value=embedded) + ), patch.object( + analyzer, "get_question_by_number", return_value={"text": "Board oversight?", "guidelines": ""} + ), patch.object( + analyzer, "_analyze_chunks", AsyncMock(return_value=analyze_result) + ) as mock_analyze, patch.object( + analyzer.cache_manager, "save_analysis" + ): + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[1], + max_processing_step="answer", + ) + + mock_analyze.assert_called_once() + assert any(e.get("question_number") == 1 for e in events) + + +@pytest.mark.asyncio +async def test_process_document_map_with_llm_scoring_uses_scoring_not_answer(analyzer, clean_db, tmp_path): + """Map + LLM scoring may call scoring batch, but must not run full answer analysis.""" + file_path = str(tmp_path / "map-scoring.pdf") + embedded = [ + { + "text": "Metric disclosure.", + "metadata": {}, + "embedding": np.array([0.3, 0.4], dtype=np.float32), + } + ] + + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + + with patch.object(analyzer.cache_manager, "get_document_chunks", return_value=embedded), patch.object( + analyzer, "_get_similar_chunks", AsyncMock(return_value=embedded) + ), patch.object(analyzer, "get_question_by_number", return_value={"text": "Emissions?", "guidelines": ""}), patch.object( + analyzer, "score_chunk_relevance_batch", AsyncMock(return_value=[0.85]) + ) as mock_score_batch, patch.object( + analyzer, "_analyze_chunks", AsyncMock() + ) as mock_analyze: + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[1], + use_llm_scoring=True, + max_processing_step="map", + ) + + mock_score_batch.assert_called_once() + mock_analyze.assert_not_called() + assert any("Completed Map" in e.get("status", "") for e in events) + + +@pytest.mark.parametrize( + ("slider", "needs_questions"), + [ + ("Chunk", False), + ("Embed", False), + ("Map", True), + ("Answer", True), + ], +) +def test_processing_step_needs_questions(slider, needs_questions): + import report_analyst.streamlit_app as app + + original = app.st.session_state + app.st.session_state = FakeSessionState({"processing_steps_slider": slider}) + try: + assert app.processing_step_needs_questions() is needs_questions + finally: + app.st.session_state = original + + +@pytest.mark.parametrize( + ("step", "partial"), + [ + ("Chunk", True), + ("Embed", True), + ("Map", False), + ("Answer", False), + ], +) +def test_is_partial_processing_step(step, partial): + import report_analyst.streamlit_app as app + + assert app.is_partial_processing_step(step) is partial + assert app.processing_step_needs_questions_for(step) is not partial + + +def test_chunk_with_selected_questions_is_partial_not_full_answer(): + """Regression: Chunk + selected questions must not take the Map/Answer LLM path.""" + import report_analyst.streamlit_app as app + + assert app.is_partial_processing_step("Chunk") is True + assert app.processing_step_needs_questions_for("Chunk") is False + + +def test_display_cached_document_chunks_renders_rows(monkeypatch): + import report_analyst.streamlit_app as app + + report_analyzer = Mock() + report_analyzer.cache_manager.get_document_chunks.return_value = [ + {"text": "Paragraph one.", "embedding": None}, + {"text": "Paragraph two.", "embedding": object()}, + ] + + dataframe_calls: list = [] + + def capture_dataframe(df, **kwargs): + dataframe_calls.append(df) + return Mock() + + monkeypatch.setattr(app.st, "subheader", Mock()) + monkeypatch.setattr(app.st, "info", Mock()) + monkeypatch.setattr(app.st, "warning", Mock()) + monkeypatch.setattr(app.st, "dataframe", capture_dataframe) + monkeypatch.setattr(app.st, "column_config", Mock(NumberColumn=Mock, TextColumn=Mock, CheckboxColumn=Mock)) + + count, embedded = app.display_cached_document_chunks( + report_analyzer, + "report.pdf", + chunk_size=500, + chunk_overlap=20, + ) + + assert count == 2 + assert embedded == 1 + assert len(dataframe_calls) == 1 + assert list(dataframe_calls[0]["Text"]) == ["Paragraph one.", "Paragraph two."] + + +@pytest.mark.asyncio +async def test_report_analyzer_analyze_document_forwards_max_processing_step(clean_db, tmp_path): + """analyze_document must pass max_processing_step through to process_document.""" + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "forward-step.pdf") + + async def fake_process_document(*args, **kwargs): + yield {"status": f"Completed {kwargs.get('max_processing_step')} step"} + + wrapper = Mock() + wrapper.analyzer = Mock(process_document=fake_process_document, question_set="tcfd") + + events = [] + async for event in app.ReportAnalyzer.analyze_document( + wrapper, + file_path, + {"tcfd_1": {"number": 1, "text": "Q1"}}, + ["tcfd_1"], + max_processing_step="chunk", + ): + events.append(event) + + assert any("chunk" in e.get("status", "") for e in events) + + +@pytest.mark.asyncio +async def test_analyze_document_and_display_chunk_step_does_not_default_to_answer(monkeypatch, clean_db, tmp_path): + """Regression: analyze_document_and_display without max_processing_step ran full Answer (OpenAI 401).""" + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "display-chunk.pdf") + captured: dict = {} + + async def fake_analyze_document(*args, **kwargs): + captured["max_processing_step"] = kwargs.get("max_processing_step") + yield {"status": "Completed Chunk step (2 chunks)"} + + fake_st = make_run_analysis_session( + results={"answers": {}}, + current_question_set="tcfd", + analyzed_files=set(), + force_recompute=True, + ) + + report_analyzer = Mock() + report_analyzer.analyze_document = fake_analyze_document + report_analyzer.cache_manager = Mock(get_analysis=Mock(return_value={})) + + monkeypatch.setattr(app, "st", Mock(session_state=fake_st, empty=Mock(return_value=Mock()))) + monkeypatch.setattr(app, "generate_file_key", Mock(return_value="test_key")) + + await app.analyze_document_and_display( + report_analyzer, + file_path=file_path, + questions={"tcfd_1": {"number": 1, "text": "Q1"}}, + selected_questions=["tcfd_1"], + force_recompute=True, + max_processing_step="chunk", + ) + + assert captured.get("max_processing_step") == "chunk" + + +@pytest.mark.asyncio +async def test_run_analysis_chunk_step_ignores_stale_analysis_cache(monkeypatch, tmp_path): + """Regression: empty question_ids analysis_cache hit hid chunks on Chunk/Embed reruns.""" + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "cached-chunks.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.analyzer.chunk_params = {"chunk_size": 500, "chunk_overlap": 20} + report_analyzer.cache_manager.get_analysis.return_value = {"tcfd_1": {"result": {"ANSWER": "Yes"}}} + report_analyzer.cache_manager.get_document_chunks.return_value = [ + {"text": "Cached chunk text.", "embedding": None}, + ] + + display_calls: list = [] + + def capture_display(*args, **kwargs): + display_calls.append(kwargs) + return 1, 1 + + monkeypatch.setattr(app, "st", Mock(session_state=make_run_analysis_session())) + monkeypatch.setattr(app, "display_cached_document_chunks", capture_display) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=[], + progress_text=progress, + max_processing_step="chunk", + ) + + report_analyzer.cache_manager.get_analysis.assert_not_called() + assert display_calls + progress.success.assert_called_once() + assert "stored chunks" in progress.success.call_args[0][0].lower() + + +@pytest.mark.asyncio +async def test_analyze_document_and_display_default_max_step_is_answer(monkeypatch, tmp_path): + """Document prior default: omitting max_processing_step means full Answer pipeline.""" + import report_analyst.streamlit_app as app + + captured: dict = {} + + async def fake_analyze_document(*args, **kwargs): + captured["max_processing_step"] = kwargs.get("max_processing_step") + yield {"status": "done"} + + fake_st = make_run_analysis_session( + results={"answers": {}}, + current_question_set="tcfd", + analyzed_files=set(), + force_recompute=True, + ) + + report_analyzer = Mock() + report_analyzer.analyze_document = fake_analyze_document + report_analyzer.cache_manager = Mock(get_analysis=Mock(return_value={})) + + monkeypatch.setattr(app, "st", Mock(session_state=fake_st, empty=Mock(return_value=Mock()))) + + await app.analyze_document_and_display( + report_analyzer, + file_path=str(tmp_path / "full.pdf"), + questions={"tcfd_1": {"number": 1, "text": "Q1"}}, + selected_questions=["tcfd_1"], + force_recompute=True, + ) + + assert captured.get("max_processing_step") == "answer" + + +def test_get_max_processing_step_reads_streamlit_slider(): + import report_analyst.streamlit_app as app + + original = app.st.session_state + app.st.session_state = FakeSessionState({"processing_steps_slider": "Chunk"}) + try: + assert app.get_max_processing_step() == "chunk" + assert app.processing_step_needs_questions() is False + finally: + app.st.session_state = original + + +@pytest.mark.parametrize("slider", ["Chunk", "Embed", "Map", "Answer"]) +def test_get_max_processing_step_normalizes_all_slider_values(slider): + import report_analyst.streamlit_app as app + + original = app.st.session_state + app.st.session_state = FakeSessionState({"processing_steps_slider": slider}) + try: + assert app.get_max_processing_step() == slider.lower() + finally: + app.st.session_state = original + + +def test_display_cached_document_chunks_empty_returns_zero(monkeypatch): + import report_analyst.streamlit_app as app + + report_analyzer = Mock() + report_analyzer.cache_manager.get_document_chunks.return_value = [] + monkeypatch.setattr(app.st, "warning", Mock()) + + count, embedded = app.display_cached_document_chunks(report_analyzer, "missing.pdf") + + assert count == 0 + assert embedded == 0 + app.st.warning.assert_called_once() + + +def test_display_cached_document_chunks_uses_chunk_text_fallback(monkeypatch): + import report_analyst.streamlit_app as app + + report_analyzer = Mock() + report_analyzer.cache_manager.get_document_chunks.return_value = [{"chunk_text": "Legacy text.", "embedding": None}] + + captured: list = [] + + def capture_dataframe(df, **kwargs): + captured.append(df) + + monkeypatch.setattr(app.st, "subheader", Mock()) + monkeypatch.setattr(app.st, "info", Mock()) + monkeypatch.setattr(app.st, "dataframe", capture_dataframe) + monkeypatch.setattr(app.st, "column_config", Mock(NumberColumn=Mock, TextColumn=Mock, CheckboxColumn=Mock)) + + count, embedded = app.display_cached_document_chunks(report_analyzer, "report.pdf") + + assert count == 1 + assert embedded == 0 + assert captured[0]["Text"].iloc[0] == "Legacy text." + + +@pytest.mark.asyncio +@pytest.mark.parametrize("step", ["chunk", "embed"]) +async def test_run_analysis_partial_step_shows_cached_document_chunks(monkeypatch, tmp_path, step): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / f"cached-{step}.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.analyzer.chunk_params = {"chunk_size": 500, "chunk_overlap": 20} + report_analyzer.cache_manager.get_document_chunks.return_value = [ + {"text": "Cached chunk.", "embedding": object() if step == "embed" else None}, + ] + + display_calls: list = [] + + monkeypatch.setattr(app, "st", Mock(session_state=make_run_analysis_session())) + monkeypatch.setattr( + app, + "display_cached_document_chunks", + lambda *a, **k: display_calls.append(k) or (1, 1 if step == "embed" else 0), + ) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=[], + progress_text=progress, + max_processing_step=step, + ) + + report_analyzer.cache_manager.get_analysis.assert_not_called() + assert display_calls + progress.success.assert_called_once() + + +@pytest.mark.asyncio +async def test_run_analysis_embed_step_reruns_when_only_text_chunks_cached(monkeypatch, tmp_path): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "needs-embed.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.analyzer.chunk_params = {"chunk_size": 500, "chunk_overlap": 20} + report_analyzer.cache_manager.get_document_chunks.return_value = [ + {"text": "Text only.", "embedding": None}, + ] + + async def fake_process_document(*args, **kwargs): + async for event in _async_status_events("Completed Embed step (1 chunks)"): + yield event + + report_analyzer.process_document = fake_process_document + display_calls: list = [] + + monkeypatch.setattr(app, "st", Mock(session_state=make_run_analysis_session())) + monkeypatch.setattr(app, "display_cached_document_chunks", lambda *a, **k: display_calls.append(1) or (1, 1)) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=[], + progress_text=progress, + max_processing_step="embed", + ) + + progress.info.assert_any_call("Starting analysis...") + assert display_calls + + +@pytest.mark.asyncio +async def test_run_analysis_chunk_force_recompute_runs_process_document(monkeypatch, tmp_path): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "force-chunk.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.analyzer.chunk_params = {"chunk_size": 500, "chunk_overlap": 20} + report_analyzer.cache_manager.get_document_chunks.return_value = [ + {"text": "Old cache.", "embedding": None}, + ] + + async def fake_process_document(*args, **kwargs): + async for event in _async_status_events("Completed Chunk step (2 chunks)"): + yield event + + report_analyzer.process_document = fake_process_document + display_mock = Mock(return_value=(2, 0)) + monkeypatch.setattr(app, "st", Mock(session_state=make_run_analysis_session(force_recompute=True))) + monkeypatch.setattr(app, "display_cached_document_chunks", display_mock) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=[], + progress_text=progress, + max_processing_step="chunk", + ) + + progress.info.assert_any_call("Starting analysis...") + display_mock.assert_called_once() + + +@pytest.mark.asyncio +async def test_run_analysis_answer_cache_hit_returns_without_process(monkeypatch, tmp_path): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "answer-cached.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.cache_manager.get_analysis.return_value = { + "tcfd_1": {"result": {"ANSWER": "Yes", "SCORE": 0.9}}, + } + + fake_st = make_run_analysis_session() + monkeypatch.setattr(app, "st", Mock(session_state=fake_st)) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=["tcfd_1"], + progress_text=progress, + max_processing_step="answer", + ) + + report_analyzer.process_document.assert_not_called() + progress.success.assert_called_once_with("Found stored results!") + assert fake_st["results"]["tcfd_1"]["result"]["ANSWER"] == "Yes" + + +@pytest.mark.asyncio +async def test_run_analysis_answer_uses_full_question_ids_without_double_prefix(monkeypatch, tmp_path): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "answer-ids.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.cache_manager.get_analysis.return_value = {} + + async def fake_process_document(*args, **kwargs): + yield {"question_number": 1, "result": {"ANSWER": "Done", "SCORE": 1.0}} + + report_analyzer.process_document = fake_process_document + + monkeypatch.setattr(app, "st", Mock(session_state=make_run_analysis_session(force_recompute=True))) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=["tcfd_1", "tcfd_2"], + progress_text=progress, + max_processing_step="answer", + ) + + report_analyzer.cache_manager.get_analysis.assert_called() + call_kwargs = report_analyzer.cache_manager.get_analysis.call_args_list[0].kwargs + assert call_kwargs["question_ids"] == ["tcfd_1", "tcfd_2"] + + +@pytest.mark.asyncio +async def test_run_analysis_answer_prefixes_numeric_question_ids(monkeypatch, tmp_path): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "answer-numeric.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.cache_manager.get_analysis.return_value = {} + + async def fake_process_document(*args, **kwargs): + if False: + yield {} + + report_analyzer.process_document = fake_process_document + patch_streamlit_app(monkeypatch, force_recompute=True) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=["1", "2"], + progress_text=progress, + max_processing_step="answer", + ) + + call_kwargs = report_analyzer.cache_manager.get_analysis.call_args_list[0].kwargs + assert call_kwargs["question_ids"] == ["tcfd_1", "tcfd_2"] + + +@pytest.mark.asyncio +async def test_run_analysis_answer_cache_miss_runs_process_and_completes(monkeypatch, tmp_path): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "answer-run.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.cache_manager.get_analysis.side_effect = [{}, {"tcfd_1": {"result": {"ANSWER": "Done"}}}] + + async def fake_process_document(*args, **kwargs): + yield {"question_number": 1, "result": {"ANSWER": "Done", "SCORE": 1.0}} + + report_analyzer.process_document = fake_process_document + monkeypatch.setattr(app, "st", Mock(session_state=make_run_analysis_session(force_recompute=True))) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=["tcfd_1"], + progress_text=progress, + max_processing_step="answer", + ) + + progress.success.assert_called_with("Analysis complete!") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("max_step", "expected"), + [ + ("chunk", "chunk"), + ("embed", "embed"), + ("map", "map"), + ("answer", "answer"), + ], +) +async def test_report_analyzer_analyze_document_forwards_all_processing_steps(max_step, expected, clean_db, tmp_path): + import report_analyst.streamlit_app as app + + captured: dict = {} + + async def fake_process_document(*args, **kwargs): + captured["max_processing_step"] = kwargs.get("max_processing_step") + yield {"status": "ok"} + + wrapper = Mock() + wrapper.analyzer = Mock(process_document=fake_process_document, question_set="tcfd") + + async for _ in app.ReportAnalyzer.analyze_document( + wrapper, + str(tmp_path / "doc.pdf"), + {"tcfd_1": {"number": 1, "text": "Q1"}}, + ["tcfd_1"], + max_processing_step=max_step, + ): + pass + + assert captured["max_processing_step"] == expected + + +@pytest.mark.asyncio +async def test_analyze_document_and_display_respects_cached_answers_without_force_recompute(monkeypatch, tmp_path): + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "display-cache.pdf") + report_analyzer = Mock() + report_analyzer.cache_manager.get_analysis.return_value = { + "tcfd_1": {"result": {"ANSWER": "Cached"}}, + } + + async def fail_analyze_document(*args, **kwargs): + raise AssertionError("should not re-analyze when cache hit") + + report_analyzer.analyze_document = fail_analyze_document + + fake_st = make_run_analysis_session( + results={"answers": {}}, + current_question_set="tcfd", + analyzed_files=set(), + force_recompute=False, + ) + + monkeypatch.setattr(app, "st", Mock(session_state=fake_st, info=Mock(), empty=Mock(return_value=Mock()))) + monkeypatch.setattr(app, "generate_file_key", Mock(return_value="file_key")) + monkeypatch.setattr(app, "create_analysis_dataframes", Mock(return_value=(Mock(), Mock()))) + + await app.analyze_document_and_display( + report_analyzer, + file_path=file_path, + questions={"tcfd_1": {"number": 1, "text": "Q1"}}, + selected_questions=["tcfd_1"], + max_processing_step="answer", + ) + + assert fake_st["results"]["answers"]["tcfd_1"]["result"]["ANSWER"] == "Cached" + + +@pytest.mark.asyncio +async def test_report_analyzer_process_document_forwards_pre_retrieved_chunks_and_max_step(): + import report_analyst.streamlit_app as app + + captured: dict = {} + + async def fake_process_document(*args, **kwargs): + captured["pre_retrieved_chunks"] = kwargs.get("pre_retrieved_chunks") + captured["max_processing_step"] = kwargs.get("max_processing_step") + yield {"status": "done"} + + report_analyzer = app.ReportAnalyzer() + report_analyzer.analyzer.process_document = fake_process_document + pre_chunks = [{"text": "chunk one"}] + + async for _ in report_analyzer.process_document( + file_path="test.pdf", + selected_questions=[1], + pre_retrieved_chunks=pre_chunks, + max_processing_step="chunk", + ): + pass + + assert captured["pre_retrieved_chunks"] == pre_chunks + assert captured["max_processing_step"] == "chunk" + + +@pytest.mark.asyncio +async def test_run_analysis_embed_failure_does_not_show_success(monkeypatch, tmp_path): + """Regression: OpenAI embed errors must not be followed by a success banner.""" + import report_analyst.streamlit_app as app + + file_path = str(tmp_path / "embed-fail.pdf") + progress = Mock() + report_analyzer = Mock() + report_analyzer.analyzer.chunk_params = {"chunk_size": 500, "chunk_overlap": 20} + report_analyzer.cache_manager.get_document_chunks.return_value = [ + {"text": "Chunk without embedding.", "embedding": None}, + ] + + async def fake_process_document(*args, **kwargs): + yield {"error": "Error processing document: Incorrect API key"} + + report_analyzer.process_document = fake_process_document + monkeypatch.setattr(app, "st", Mock(session_state=make_run_analysis_session(force_recompute=True))) + monkeypatch.setattr(app, "display_cached_document_chunks", Mock(return_value=(1, 0))) + + await app.run_analysis( + report_analyzer, + file_path=file_path, + selected_questions=[], + progress_text=progress, + max_processing_step="embed", + ) + + progress.error.assert_called() + progress.success.assert_not_called() + + +def test_normalize_processing_step_via_custom_label(monkeypatch): + import report_analyst.core.analyzer as analyzer_mod + + monkeypatch.setitem(analyzer_mod.PROCESSING_STEP_LABELS, "chunk", "Chunk Only") + assert normalize_processing_step("Chunk Only") == "chunk" + assert normalize_processing_step("unknown-step") == "answer" + + +def test_split_pdf_to_text_chunks_skips_blank(analyzer, tmp_path, monkeypatch): + doc = Mock() + doc.text = "Hello world.\n\n\n" + doc.metadata = {"page": 1} + + class FakeReader: + def load(self, file_path): + return [doc] + + monkeypatch.setattr("report_analyst.core.analyzer.PyMuPDFReader", FakeReader) + monkeypatch.setattr(analyzer, "text_splitter", Mock(split_text=Mock(return_value=["Hello world.", " ", ""]))) + chunks = analyzer._split_pdf_to_text_chunks(str(tmp_path / "report.pdf")) + assert len(chunks) == 1 + assert chunks[0]["text"] == "Hello world." + assert chunks[0]["embedding"] is None + + +def test_create_text_chunks_wraps_errors(analyzer, monkeypatch): + monkeypatch.setattr(analyzer, "_split_pdf_to_text_chunks", Mock(side_effect=ValueError("bad pdf"))) + with pytest.raises(ValueError, match="bad pdf"): + analyzer._create_text_chunks("report.pdf") + + +def test_ensure_embeddings_client_requires_key(analyzer, monkeypatch): + analyzer.embeddings = None + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + with pytest.raises(RuntimeError, match="OPENAI_API_KEY"): + analyzer._ensure_embeddings_client() + + +def test_add_embeddings_to_chunks_skips_none(analyzer, monkeypatch): + analyzer.embeddings = Mock() + analyzer.embeddings.get_text_embedding_batch.return_value = [[0.1, 0.2], None] + monkeypatch.setattr(analyzer, "_ensure_embeddings_client", Mock()) + chunks = [{"text": "a"}, {"text": "b"}] + embedded = analyzer._add_embeddings_to_chunks(chunks) + assert len(embedded) == 1 + assert embedded[0]["text"] == "a" + + +@pytest.mark.asyncio +async def test_process_document_backend_urn_without_chunks_errors(analyzer): + events = await _collect_process_events( + analyzer, + file_path="urn:report-analyst:backend:abc", + selected_questions=[], + max_processing_step="chunk", + ) + assert any(e.get("error") for e in events) + + +@pytest.mark.asyncio +async def test_process_document_embed_creates_and_saves_chunks(analyzer, clean_db, tmp_path): + file_path = str(tmp_path / "embed-new.pdf") + embedded = [{"text": "Embedded text.", "metadata": {}, "embedding": [0.1, 0.2]}] + analyzer.cache_manager = CacheManager(db_path=str(clean_db)) + with patch.object(analyzer, "_create_chunks", return_value=embedded) as mock_create: + events = await _collect_process_events( + analyzer, + file_path=file_path, + selected_questions=[], + max_processing_step="embed", + ) + mock_create.assert_called_once() + assert any("Completed Embed" in e.get("status", "") for e in events) + stored = analyzer.cache_manager.get_document_chunks(file_path, chunk_size=500, chunk_overlap=20) + assert stored + + +def test_ensure_embeddings_client_creates_when_missing(analyzer, monkeypatch): + analyzer.embeddings = None + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + fake = Mock() + monkeypatch.setattr("report_analyst.core.analyzer.build_openai_embedding", Mock(return_value=fake)) + monkeypatch.setattr("report_analyst.core.analyzer.Settings", Mock()) + analyzer._ensure_embeddings_client() + assert analyzer.embeddings is fake