--- type: reference area: knowledge status: reference date: 2026-05-17 created: 2026-05-17 updated: 2026-05-17 tags: - knowledge --- I'll provide a comprehensive migration plan from Streamlit monolithic to FastAPI + Outlines + React/HTML frontend. --- # STALGEN-AI Migration Plan: Streamlit → FastAPI + Outlines + React ## Architecture Comparison ``` CURRENT (Monolithic): TARGET (Microservices): ┌─────────────────┐ ┌──────────────┐ ┌──────────────┐ │ Streamlit UI │ │ Frontend │ │ Browser │ │ (Stateful) │ │ (React) │ or │ (HTML/CSS) │ └────────┬────────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────┐ ┌────────────────────────────────┐ │ Tied Together │ │ FastAPI Backend │ │ - UI Logic │ ====> │ ┌──────────┐ ┌───────────┐ │ │ - LLM Calls │ │ │ API Layer│ │ Services │ │ │ - JSON Parsing │ │ └────┬─────┘ └─────┬─────┘ │ │ - Retry Logic │ │ │ │ │ └────────┬────────┘ │ ▼ ▼ │ │ │ ┌──────────────────────┐ │ ▼ │ │ Outlines + LLM │ │ ┌─────────────────┐ │ │ (Structured JSON) │ │ │ Ollama API │ │ └──────────────────────┘ │ └─────────────────┘ └───────────────┬─────────────┘ │ ▼ ┌──────────────┐ │ Ollama │ └──────────────┘ ``` --- ## Phase 1: FastAPI Backend Setup ### 1.1 Project Structure ``` stalgen-api/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI app entry │ ├── api/ │ │ ├── __init__.py │ │ ├── routes/ │ │ │ ├── __init__.py │ │ │ ├── documents.py # Document upload/process endpoints │ │ │ ├── scenarios.py # Scenario generation endpoints │ │ │ └── health.py # Health check │ │ └── deps.py # Dependency injection │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py # Settings using Pydantic │ │ └── exceptions.py # Custom exceptions │ ├── services/ │ │ ├── __init__.py │ │ ├── document_parser.py │ │ ├── data_extractor.py # Refactored from current │ │ ├── scenario_generator.py │ │ └── excel_exporter.py │ ├── schemas/ # Pydantic models (Outlines targets) │ │ ├── __init__.py │ │ ├── document.py │ │ └── scenario.py │ └── llm/ │ ├── __init__.py │ ├── outlines_client.py # NEW: Outlines integration │ ├── prompts.py # Prompt management │ └── models.py # Model configurations ├── tests/ ├── pyproject.toml └── requirements.txt ``` ### 1.2 Core Dependencies ```toml # requirements-api.txt fastapi>=0.115.0 uvicorn[standard]>=0.30.0 pydantic>=2.9.0 pydantic-settings>=2.6.0 outlines>=0.2.0 ollama>=0.3.0 python-multipart>=0.0.20 aiofiles>=24.0.0 openpyxl>=3.1.0 pypandoc>=0.8.0 python-docx>=1.1.0 ``` ### 1.3 Settings Configuration (`app/core/config.py`) ```python from pydantic_settings import BaseSettings, SettingsConfigDict from functools import lru_cache class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", case_sensitive=False, ) # App app_name: str = "STALGEN API" debug: bool = False api_prefix: str = "/api/v1" # Ollama ollama_base_url: str = "http://localhost:11434" default_model: str = "deepseek-r1:14b" # File limits max_file_size_mb: int = 10 temp_dir_ttl_hours: int = 1 # Outlines outlines_max_tokens: int = 4096 outlines_temperature: float = 0.1 @lru_cache def get_settings() -> Settings: return Settings() ``` ### 1.4 API Main Entry (`app/main.py`) ```python from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager from app.api.routes import documents, scenarios, health from app.core.config import get_settings from app.core.exceptions import register_exceptions @asynccontextmanager async def lifespan(app: FastAPI): # Startup settings = get_settings() print(f"Starting {settings.app_name}") yield # Shutdown print("Shutting down...") def create_app() -> FastAPI: settings = get_settings() app = FastAPI( title=settings.app_name, description="STALGEN AI - SFD to Test Scenarios API", version="1.0.0", lifespan=lifespan, ) # CORS for frontend app.add_middleware( CORSMiddleware, allow_origins=["*"], # Restrict in production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Register exceptions register_exceptions(app) # Include routers app.include_router(health.router, prefix=settings.api_prefix) app.include_router(documents.router, prefix=settings.api_prefix) app.include_router(scenarios.router, prefix=settings.api_prefix) return app app = create_app() ``` --- ## Phase 2: Schema Definitions (Outlines Targets) ### 2.1 Pydantic Schemas (`app/schemas/scenario.py`) ```python from pydantic import BaseModel, Field from typing import List, Optional from enum import Enum class TestCaseType(str, Enum): HAPPY_PATH = "happy_path" ALTERNATIVE = "alternative_flow" BOUNDARY = "boundary_testing" NEGATIVE = "negative_testing" ERROR_HANDLING = "error_handling" class TestCase(BaseModel): """Single test case""" description: str = Field( ..., min_length=10, max_length=500, description="Test case description starting with Verify/Check/Validate" ) type: Optional[TestCaseType] = None # Outlines will generate: {"description": "...", "type": "happy_path"} class MacroScenario(BaseModel): """A macro scenario with multiple test cases""" title: str = Field( ..., min_length=5, max_length=200, description="Clear descriptive scenario title" ) test_cases: List[TestCase] = Field( ..., min_length=2, max_length=15, description="List of specific test cases" ) # Outlines generates array of these class ScenarioBundle(BaseModel): """Bundle for a single user story""" user_story_id: str user_story_title: str scenarios: List[MacroScenario] source_document: str class ScenarioGenerationRequest(BaseModel): """Request to generate scenarios for a user story""" user_story_id: str user_story_title: str user_story_content: str business_need: str product_specs: str source_document: str model: Optional[str] = None class ScenarioGenerationResponse(BaseModel): """Response with generated scenarios""" success: bool bundle: ScenarioBundle model_used: str processing_time_seconds: float class DocumentProcessRequest(BaseModel): """Request to process an entire SFD document""" filename: str skip_pages: Optional[List[int]] = [] model: Optional[str] = None class DocumentProcessResponse(BaseModel): """Response after processing document""" success: bool document_id: str user_stories_found: int bundles: List[ScenarioBundle] excel_path: Optional[str] = None summary: dict ``` --- ## Phase 3: Outlines Integration (Replace JSON Parsing + Retries) ### 3.1 Outlines Client (`app/llm/outlines_client.py`) ```python import logging import time from typing import List, Optional, Dict, Any from pathlib import Path import outlines from outlines import models as outlines_models from pydantic import BaseModel from app.core.config import get_settings from app.llm.prompts import ( SYSTEM_PROMPT, USER_PROMPT_TEMPLATE, FEW_SHOT_EXAMPLES, ) logger = logging.getLogger(__name__) class OutlinesClient: """ Structured LLM client using Outlines for deterministic JSON output. Replaces: ResponseProcessor + JSON parsing + retry logic """ def __init__( self, model_name: Optional[str] = None, temperature: float = 0.1, max_tokens: int = 4096, ): settings = get_settings() self.model_name = model_name or settings.default_model self.temperature = temperature self.max_tokens = max_tokens self._setup_model() def _setup_model(self) -> None: """Initialize Outlines model with Ollama backend""" # Connect to local Ollama self.model = outlines_models.ollama( model=self.model_name, base_url=get_settings().ollama_base_url, ) logger.info(f"Outlines initialized with model: {self.model_name}") def generate( self, prompt: str, response_schema: type[BaseModel], system_prompt: Optional[str] = None, few_shot_examples: Optional[List[Dict[str, str]]] = None, ) -> BaseModel: """ Generate structured response using Outlines. Args: prompt: User prompt response_schema: Pydantic model for structured output system_prompt: Optional system prompt few_shot_examples: Optional few-shot examples Returns: Instance of response_schema with generated data """ # Build complete prompt full_prompt = self._build_prompt(prompt, system_prompt, few_shot_examples) # Create Outlines generator for JSON schema generator = outlines.generate.json( self.model, response_schema, temperature=self.temperature, max_tokens=self.max_tokens, ) logger.info(f"Generating with Outlines, model: {self.model_name}") start_time = time.time() try: # Direct generation - no parsing, no retries needed result = generator(full_prompt) elapsed = time.time() - start_time logger.info(f"Generation completed in {elapsed:.2f}s") return result except Exception as e: logger.error(f"Outlines generation failed: {e}") raise def _build_prompt( self, user_content: str, system_prompt: Optional[str], few_shot_examples: Optional[List[Dict[str, str]]], ) -> str: """Build the complete prompt with optional system and few-shot""" parts = [] if system_prompt: parts.append(f"System: {system_prompt}") if few_shot_examples: for example in few_shot_examples: if example.get("role") == "user": parts.append(f"User: {example['content']}") elif example.get("role") == "assistant": parts.append(f"Assistant: {example['content']}") parts.append(f"User: {user_content}") return "\n\n".join(parts) def generate_scenarios( self, user_story_content: str, business_need: str, product_specs: str, user_story_title: str, source_document: str, ) -> Dict[str, Any]: """ Generate test scenarios for a user story using structured output. """ # Import here to avoid circular imports from app.schemas.scenario import ( TestCase, MacroScenario, ScenarioBundle ) # Build user prompt user_prompt = USER_PROMPT_TEMPLATE.format( user_story_title=user_story_title, user_story_content=user_story_content, business_need=business_need, product_specs=product_specs, ) # Use Outlines to generate structured MacroScenario list generator = outlines.generate.json( self.model, List[MacroScenario], temperature=self.temperature, max_tokens=self.max_tokens, ) # Generate scenarios directly scenarios = generator(user_prompt) bundle = ScenarioBundle( user_story_id=user_story_content[:20], # Should use actual ID user_story_title=user_story_title, scenarios=scenarios, source_document=source_document, ) return bundle ``` ### 3.2 Prompt Management (`app/llm/prompts.py`) ```python SYSTEM_PROMPT = """You are STALGEN-AI, an expert test automation assistant specialized in automotive software testing for Stellantis-ALTEN projects. Generate comprehensive Macro Scenarios and detailed Test Cases from Software Functional Documents (SFD). ... """ USER_PROMPT_TEMPLATE = """ ## Task Generate test scenarios for the following user story. ## User Story Title: {user_story_title} Content: {user_story_content} ## Business Need {business_need} ## Product Specifications {product_specs} ## CRITICAL INSTRUCTIONS 1. Generate test scenarios ONLY from the content provided above 2. Do NOT invent features, rules, or requirements not mentioned 3. Each test case MUST directly trace to a specific requirement 4. Start each test case with: Verify, Check, Validate, Ensure, or Confirm 5. Generate 3-8 test cases per macro scenario ## Output Format Generate a JSON array of macro scenarios, each containing test cases. """ ``` --- ## Phase 4: Refactored Services ### 4.1 Document Parser Service (`app/services/document_parser.py`) ```python import logging import tempfile import shutil from pathlib import Path from typing import List, Optional from dataclasses import dataclass import pypandoc from docx import Document as DocxDocument from docx.oxml.ns import qn logger = logging.getLogger(__name__) @dataclass class ParsedDocument: markdown_path: Path skipped_pages: List[int] total_pages: int class DocumentParser: """ Handles Word document to Markdown conversion. Refactored from: word_parser_pandoc.py """ def __init__(self, output_format: str = "gfm"): self.output_format = output_format self._check_pandoc() def _check_pandoc(self) -> None: """Verify pandoc is installed""" try: pypandoc.get_pandoc_version() except RuntimeError: raise RuntimeError( "Pandoc is not installed. " "Install from: https://pandoc.org/installing.html" ) def parse( self, input_path: Path, output_path: Optional[Path] = None, skip_pages: Optional[List[int]] = None, ) -> ParsedDocument: """ Parse Word document to Markdown. Args: input_path: Path to .docx file output_path: Optional output path for .md file skip_pages: Pages to skip (1-indexed) Returns: ParsedDocument with paths and metadata """ if not input_path.exists(): raise FileNotFoundError(f"Input file not found: {input_path}") if input_path.suffix.lower() not in ['.docx', '.doc']: raise ValueError(f"Expected .docx file, got: {input_path.suffix}") # Create temp output if not specified if output_path is None: temp_md = tempfile.NamedTemporaryFile(suffix=".md", delete=False) output_path = Path(temp_md.name) # Build filtered document (removing skipped pages) skip_set = set(skip_pages or []) filtered_doc = self._filter_document(input_path, skip_set) # Save filtered doc temporarily temp_docx = tempfile.NamedTemporaryFile(suffix=".docx", delete=False) filtered_doc.save(temp_docx.name) try: # Convert with pypandoc pypandoc.convert_file( temp_docx.name, to=self.output_format, outputfile=str(output_path), extra_args=['--wrap=none'], ) finally: Path(temp_docx.name).unlink(missing_ok=True) logger.info(f"Converted {input_path} -> {output_path}") return ParsedDocument( markdown_path=output_path, skipped_pages=sorted(skip_set), total_pages=self._estimate_pages(input_path), ) def _filter_document( self, doc_path: Path, skip_pages: set[int], ) -> DocxDocument: """Filter out specified pages from document""" doc = DocxDocument(doc_path) page_boundaries = self._locate_page_breaks(doc) new_doc = DocxDocument() self._clear_document(new_doc) for page_num, blocks in enumerate(page_boundaries, 1): if page_num not in skip_pages: for block in blocks: self._copy_block(new_doc, block) return new_doc def _locate_page_breaks(self, doc: DocxDocument) -> List[List]: """Locate page break boundaries in document""" # Implementation from original code ... ``` ### 4.2 Data Extractor Service (`app/services/data_extractor.py`) **Bug Fix Applied:** Line breaks preserved in descriptions ```python from dataclasses import dataclass from typing import List, Optional import re @dataclass class UserStoryData: id: str title: str description: str start_index: int end_index: int @dataclass class SFDStructuredData: business_need: str product_specs: str user_stories: List[UserStoryData] source_file: str class DataExtractor: """ Extracts structured data from SFD markdown. Refactored: Single responsibility - extraction only """ BUSINESS_NEED_PATTERN = r"## Business Need Statement.*?\n(.*?)\n## Scope Statement" PRODUCT_SPECS_PATTERN = r"# PRODUCT SPECIFICATIONS\n(.*?)\n## Parameter management" USER_STORY_PATTERNS = [ r"### User Story:", r"### User Story:", ] def __init__( self, business_need_pattern: Optional[str] = None, product_specs_pattern: Optional[str] = None, user_story_patterns: Optional[List[str]] = None, ): self.business_need_pattern = business_need_pattern or self.BUSINESS_NEED_PATTERN self.product_specs_pattern = product_specs_pattern or self.PRODUCT_SPECS_PATTERN self.user_story_patterns = user_story_patterns or self.USER_STORY_PATTERNS self._normalized_patterns = [ self._normalize_pattern(p) for p in self.user_story_patterns ] def extract(self, markdown_content: str, source_filename: str) -> SFDStructuredData: """Extract structured data from markdown""" business_need = self._extract_business_need(markdown_content) product_specs = self._extract_product_specs(markdown_content) text_lines = markdown_content.splitlines(keepends=True) user_stories = self._extract_user_stories(text_lines) return SFDStructuredData( business_need=business_need, product_specs=product_specs, user_stories=user_stories, source_file=source_filename, ) def _extract_user_stories(self, text_lines: List[str]) -> List[UserStoryData]: """Extract user stories from text lines - FIXED: preserves line breaks""" user_story_positions = [] for i, line in enumerate(text_lines): normalized = self._normalize_pattern(line) for pattern in self._normalized_patterns: if normalized.startswith(pattern): title = self._extract_title(line) user_story_positions.append((i, title)) break user_stories = [] for idx, (start_idx, title) in enumerate(user_story_positions): # FIX: Changed from "".join() to "\n".join() - preserves line breaks end_idx = ( user_story_positions[idx + 1][0] if idx + 1 < len(user_story_positions) else len(text_lines) ) description_lines = text_lines[start_idx + 1 : end_idx] description = "\n".join(description_lines).strip() # BUG FIX user_stories.append(UserStoryData( id=f"US-{idx + 1:03d}", title=title, description=description, start_index=start_idx, end_index=end_idx, )) return user_stories @staticmethod def _normalize_pattern(text: str) -> str: """Normalize heading pattern for comparison""" cleaned = text.replace("\\<", "<").replace("\\>", ">").strip() cleaned = cleaned.lstrip("#").strip() cleaned = re.sub(r"\s+", " ", cleaned) return cleaned.replace(":", "").lower() @staticmethod def _extract_title(text: str) -> str: """Extract title from heading""" cleaned = text.strip().lstrip("#").strip() if not cleaned: return "" parts = cleaned.split(":", 1) remainder = parts[1].strip() if len(parts) == 2 else cleaned remainder = re.sub( r"^(\s+)?user story\s*", "", remainder, flags=re.IGNORECASE, ) return remainder.strip() ``` ### 4.3 Scenario Generator Service (`app/services/scenario_generator.py`) ```python from typing import Dict, Any import time from app.llm.outlines_client import OutlinesClient from app.schemas.scenario import ( ScenarioBundle, MacroScenario, TestCase, ScenarioGenerationRequest, ScenarioGenerationResponse, ) from app.core.config import get_settings class ScenarioGenerator: """ Generates test scenarios using Outlines. Replaces: llm_system_refactored.py + ResponseProcessor """ def __init__(self, model_name: str = None): settings = get_settings() self.model_name = model_name or settings.default_model self._client: OutlinesClient = None @property def client(self) -> OutlinesClient: """Lazy initialization of Outlines client""" if self._client is None: self._client = OutlinesClient(model_name=self.model_name) return self._client def generate( self, request: ScenarioGenerationRequest, ) -> ScenarioGenerationResponse: """ Generate scenarios for a user story. This replaces the old JSON parsing + retry logic with Outlines' structured generation. """ start_time = time.time() # Build the prompt user_prompt = self._build_prompt( user_story_content=request.user_story_content, user_story_title=request.user_story_title, business_need=request.business_need, product_specs=request.product_specs, ) # Generate directly with Outlines - NO parsing, NO retries scenarios = self.client.generate( prompt=user_prompt, response_schema=List[MacroScenario], system_prompt=self._get_system_prompt(), ) bundle = ScenarioBundle( user_story_id=request.user_story_id, user_story_title=request.user_story_title, scenarios=scenarios, source_document=request.source_document, ) elapsed = time.time() - start_time return ScenarioGenerationResponse( success=True, bundle=bundle, model_used=self.model_name, processing_time_seconds=elapsed, ) def _build_prompt( self, user_story_content: str, user_story_title: str, business_need: str, product_specs: str, ) -> str: """Build the generation prompt""" return f"""## Task: Generate test scenarios for this user story ### User Story Title: {user_story_title} Content: {user_story_content} ### Business Need {business_need} ### Product Specifications {product_specs} ### CRITICAL REQUIREMENTS 1. Generate scenarios ONLY from the provided content 2. Do NOT invent features or requirements not in the input 3. Each test case must trace to a specific input requirement 4. Start test cases with: Verify, Check, Validate, Ensure, or Confirm 5. Generate 3-8 test cases per scenario (quality over quantity) ### Output Format Return a JSON array of macro scenarios, each with: - "title": Clear descriptive title from input - "test_cases": Array of test case objects with "description" field Generate the scenarios now:""" def _get_system_prompt(self) -> str: return """You are STALGEN-AI, an expert test automation assistant for automotive software testing at Stellantis-ALTEN projects. Generate precise, testable scenarios from the provided SFD content. Output ONLY valid JSON matching the specified schema.""" ``` --- ## Phase 5: API Routes ### 5.1 Document Routes (`app/api/routes/documents.py`) ```python from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks from fastapi.responses import FileResponse from typing import Optional, List import uuid import shutil from pathlib import Path from app.schemas.scenario import DocumentProcessRequest, DocumentProcessResponse from app.services.document_parser import DocumentParser from app.services.data_extractor import DataExtractor from app.services.scenario_generator import ScenarioGenerator from app.services.excel_exporter import ExcelExporter from app.core.config import get_settings from app.core.exceptions import ProcessingError router = APIRouter(prefix="/documents", tags=["documents"]) @router.post("/process", response_model=DocumentProcessResponse) async def process_document( background_tasks: BackgroundTasks, file: UploadFile = File(...), skip_pages: Optional[str] = None, # comma-separated "1,2,3" model: Optional[str] = None, ): """ Process a Word document and generate test scenarios. 1. Upload and parse document 2. Extract user stories 3. Generate scenarios for each 4. Export to Excel """ settings = get_settings() # Validate file if not file.filename.endswith(".docx"): raise HTTPException(400, "Only .docx files are supported") # Parse skip pages skip_list = [] if skip_pages: try: skip_list = [int(x.strip()) for x in skip_pages.split(",")] except ValueError: raise HTTPException(400, "Invalid skip_pages format") # Save uploaded file document_id = str(uuid.uuid4()) temp_dir = Path(settings.temp_dir) / document_id temp_dir.mkdir(parents=True, exist_ok=True) upload_path = temp_dir / file.filename with open(upload_path, "wb") as f: shutil.copyfileobj(file.file, f) try: # Parse document parser = DocumentParser() parsed = parser.parse(upload_path, skip_pages=skip_list) # Extract data extractor = DataExtractor() markdown_content = parsed.markdown_path.read_text(encoding="utf-8") structured = extractor.extract(markdown_content, file.filename) # Generate scenarios generator = ScenarioGenerator(model_name=model) bundles = [] for us in structured.user_stories: # Find the actual content for this user story us_content = extractor.get_user_story_content( markdown_content, us ) request = ScenarioGenerationRequest( user_story_id=us.id, user_story_title=us.title, user_story_content=us_content, business_need=structured.business_need, product_specs=structured.product_specs, source_document=file.filename, ) response = generator.generate(request) bundles.append(response.bundle) # Export to Excel exporter = ExcelExporter() excel_path = temp_dir / f"{Path(file.filename).stem}_scenarios.xlsm" exporter.export(bundles, excel_path) return DocumentProcessResponse( success=True, document_id=document_id, user_stories_found=len(structured.user_stories), bundles=bundles, excel_path=str(excel_path), summary={ "filename": file.filename, "total_scenarios": sum(len(b.scenarios) for b in bundles), "total_test_cases": sum( sum(len(s.test_cases) for s in b.scenarios) for b in bundles ), }, ) except Exception as e: raise ProcessingError(f"Document processing failed: {e}") ``` ### 5.2 Scenario Routes (`app/api/routes/scenarios.py`) ```python from fastapi import APIRouter, HTTPException from app.schemas.scenario import ( ScenarioGenerationRequest, ScenarioGenerationResponse, ) from app.services.scenario_generator import ScenarioGenerator router = APIRouter(prefix="/scenarios", tags=["scenarios"]) @router.post("/generate", response_model=ScenarioGenerationResponse) async def generate_scenarios(request: ScenarioGenerationRequest): """ Generate test scenarios for a single user story. Use this endpoint for: - Generating scenarios for a specific user story - Testing prompt configurations - Batch processing custom inputs """ try: generator = ScenarioGenerator(model_name=request.model) return generator.generate(request) except Exception as e: raise HTTPException(500, f"Generation failed: {str(e)}") @router.post("/generate/batch") async def generate_batch( requests: list[ScenarioGenerationRequest], model: str = None, ): """ Generate scenarios for multiple user stories in batch. Processes sequentially but returns all results. For large batches, use background tasks. """ generator = ScenarioGenerator(model_name=model) results = [] for req in requests: try: response = generator.generate(req) results.append({"success": True, "data": response}) except Exception as e: results.append({"success": False, "error": str(e)}) return {"results": results} ``` --- ## Phase 6: Excel Export Service (`app/services/excel_exporter.py`) **Bug Fix Applied:** Cell value set BEFORE merge ```python from pathlib import Path from typing import List from openpyxl import load_workbook from app.schemas.scenario import ScenarioBundle, MacroScenario class ExcelExporter: """ Export scenario bundles to Excel template. Refactored from: json_to_excel.py Bug fix: Set cell value BEFORE merging """ def __init__(self, template_path: str = None): from app.core.config import get_settings settings = get_settings() self.template_path = Path(template_path or settings.excel_template) def export(self, bundles: List[ScenarioBundle], output_path: Path) -> Path: """Export bundles to Excel file""" if not self.template_path.exists(): raise FileNotFoundError(f"Template not found: {self.template_path}") wb = load_workbook(self.template_path, keep_vba=True) ws = wb["Template"] header_row = 20 headers = {cell.value: cell.column for cell in ws[header_row] if cell.value} macro_col = headers.get("Macro- Scenario") test_col = headers.get("Test cases") if not macro_col or not test_col: raise ValueError( f"Required columns not found. Available: {list(headers.keys())}" ) current_row = header_row + 1 for bundle in bundles: for scenario in bundle.scenarios: macro = scenario.title test_cases = [tc.description for tc in scenario.test_cases] if not test_cases: continue start_row = current_row # BUG FIX: Set value BEFORE merge, not after ws.cell(row=start_row, column=macro_col, value=macro) for tc in test_cases: ws.cell(row=current_row, column=test_col, value=tc) current_row += 1 if len(test_cases) > 1: ws.merge_cells( start_row=start_row, start_column=macro_col, end_row=current_row - 1, end_column=macro_col, ) # Clear existing content below last row for row_idx in range(current_row, ws.max_row + 1): for col_idx in [macro_col, test_col]: ws.cell(row=row_idx, column=col_idx).value = None output_path.parent.mkdir(parents=True, exist_ok=True) wb.save(output_path) return output_path ``` --- ## Phase 7: Frontend Options ### Option A: React Frontend (Recommended for Complex Apps) ``` frontend-react/ ├── src/ │ ├── components/ │ │ ├── DocumentUploader.tsx │ │ ├── ProcessingStatus.tsx │ │ ├── ScenarioList.tsx │ │ ├── MetricsDashboard.tsx │ │ └── ExcelDownload.tsx │ ├── hooks/ │ │ ├── useDocumentProcessing.ts │ │ ├── useScenarios.ts │ │ └── useDownload.ts │ ├── services/ │ │ └── api.ts # API client │ ├── types/ │ │ └── index.ts # TypeScript interfaces │ ├── App.tsx │ └── main.tsx ├── package.json └── vite.config.ts ``` **Key React Component (`DocumentUploader.tsx`):** ```tsx import { useState, useCallback } from 'react'; import { useDocumentProcessing } from '../hooks/useDocumentProcessing'; export function DocumentUploader() { const [file, setFile] = useState(null); const [skipPages, setSkipPages] = useState(''); const { process, loading, progress, result, error } = useDocumentProcessing(); const handleSubmit = useCallback(async () => { if (!file) return; await process(file, skipPages); }, [file, skipPages, process]); return (
setFile(e.target.files?.[0] || null)} /> setSkipPages(e.target.value)} /> {loading && } {result && } {error && }
); } ``` ### Option B: Simple HTML/CSS (For Simpler Requirements) ``` frontend-simple/ ├── index.html ├── css/ │ └── styles.css ├── js/ │ ├── app.js # Main application logic │ ├── api.js # API calls │ └── ui.js # DOM manipulation └── attachments/ ``` **Example HTML Structure (`index.html`):** ```html STALGEN - Document Processing

STALGEN

Smart Test Acceptance & Logic Generator

Upload Document

``` **API Client (`js/api.js`):** ```javascript const API_BASE = 'http://localhost:8000/api/v1'; const API = { async processDocument(file, skipPages = '') { const formData = new FormData(); formData.append('file', file); if (skipPages) { formData.append('skip_pages', skipPages); } const response = await fetch(`${API_BASE}/documents/process`, { method: 'POST', body: formData, }); if (!response.ok) { const error = await response.json(); throw new Error(error.detail || 'Processing failed'); } return response.json(); }, async generateScenarios(request) { const response = await fetch(`${API_BASE}/scenarios/generate`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(request), }); if (!response.ok) { throw new Error('Generation failed'); } return response.json(); }, getExcelUrl(excelPath) { // Point to download endpoint return `${API_BASE}/documents/download?path=${encodeURIComponent(excelPath)}`; } }; ``` --- ## Phase 8: Migration Checklist ### Immediate (Before Migration) - [ ] Fix Critical bugs in current codebase - [ ] Write integration tests for core flows - [ ] Document current API contracts (if any) ### Phase 1: Backend Core - [ ] Setup FastAPI project structure - [ ] Implement settings/config management - [ ] Create Pydantic schemas for all data models - [ ] Implement Outlines client - [ ] Test LLM generation with Outlines ### Phase 2: Services - [ ] Refactor DocumentParser service - [ ] Refactor DataExtractor with bug fix - [ ] Refactor ScenarioGenerator with Outlines - [ ] Refactor ExcelExporter with bug fix - [ ] Write unit tests for each service ### Phase 3: API Routes - [ ] Implement `/documents/process` endpoint - [ ] Implement `/scenarios/generate` endpoint - [ ] Add error handling and validation - [ ] Add CORS configuration - [ ] Write API integration tests ### Phase 4: Frontend (Choose A or B) **Option A - React:** - [ ] Setup React project with Vite - [ ] Create API client service - [ ] Build DocumentUploader component - [ ] Build ProcessingStatus component - [ ] Build ResultsView component - [ ] Implement download functionality **Option B - HTML/CSS:** - [ ] Create HTML structure - [ ] Write CSS styles - [ ] Implement JS API client - [ ] Build UI interaction logic - [ ] Test in multiple browsers ### Phase 5: Deployment - [ ] Docker containerization - [ ] Environment configuration - [ ] CI/CD pipeline - [ ] API documentation (Swagger/OpenAPI auto-generated) - [ ] Load testing --- ## Key Benefits of This Migration | Aspect | Before | After | |--------|--------|-------| | JSON Parsing | Manual with retries | Outlines direct generation | | Response Reliability | ~70% (needs fallback) | ~95%+ | | API Structure | None (monolithic) | RESTful API | | Frontend Flexibility | Streamlit only | Any client | | Testing | Hard to test UI | Separable components | | Deployment | Single container | Scalable microservices | --- ## Risks & Mitigations | Risk | Impact | Mitigation | |------|--------|------------| | Outlines compatibility | High | Test with target Ollama models early | | Breaking API changes | Medium | Use versioning (/api/v1) | | Frontend complexity | Low | Choose HTML/CSS if timeline tight | | Performance regression | Medium | Benchmark before/after |