"""Flask web application for radiology report structuring using Gemini models. This module provides a web API that structures radiology reports into semantic sections using LangExtract and Google's Gemini language models. The application supports caching, multiple model configurations, and provides both a web interface and REST API endpoints. Typical usage example: # Set environment variables export KEY=your_gemini_api_key_here export MODEL_ID=gemini-flash-latest # Run the application python app.py """ import hashlib import json import logging import os from pathlib import Path import shutil import tempfile import time from flask import Flask, jsonify, render_template, request from flask_limiter import Limiter from flask_limiter.util import get_remote_address from cache_manager import CacheManager from sanitize import preprocess_report from social_sharing import SocialSharingConfig from structure_report import ( DEFAULT_MODEL_ID, LEGACY_MODEL_ALIASES, RadiologyReportStructurer, ReportProcessingError, ResponseDict, resolve_model_id, ) # Configuration constants MAX_INPUT_LENGTH = 3000 _PUBLIC_MODEL_IDS = frozenset({DEFAULT_MODEL_ID, *LEGACY_MODEL_ALIASES.values()}) _SAMPLE_REPORTS_PATH = ( Path(__file__).resolve().parent / "static" / "sample_reports.json" ) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s" ) logger = logging.getLogger(__name__) def _load_sample_catalog() -> dict[str, str]: """Loads normalized sample reports keyed by their public sample ID.""" sample_data = json.loads(_SAMPLE_REPORTS_PATH.read_text(encoding="utf-8")) return { sample["id"]: preprocess_report(sample["text"]) for sample in sample_data["samples"] } _SAMPLE_CATALOG = _load_sample_catalog() def _allowed_request_models() -> frozenset[str]: """Returns model IDs that callers may select through the public API.""" configured_model = resolve_model_id(os.environ.get("MODEL_ID", DEFAULT_MODEL_ID)) return _PUBLIC_MODEL_IDS | {configured_model} def _verified_sample_id(sample_id: str | None, report_text: str) -> str | None: """Returns a sample ID only when its normalized catalog text matches.""" if sample_id and _SAMPLE_CATALOG.get(sample_id) == report_text: return sample_id return None class Model: """Manages RadiologyReportStructurer instances for different Gemini model IDs. This class handles initialization, caching, and coordination of structurer instances for various model configurations, ensuring efficient resource usage and consistent API key management. """ def __init__(self): """Initializes the Model manager with default structurer. Sets up the Gemini API key from environment variables and creates a default structurer instance for the configured model. Raises: ValueError: If the KEY environment variable is not set. """ self.gemini_api_key = os.environ.get("KEY") if not self.gemini_api_key: logger.error("KEY environment variable not set.") raise ValueError("KEY environment variable not set.") self._structurers: dict[str, RadiologyReportStructurer] = {} default_model_id = resolve_model_id( os.environ.get("MODEL_ID", DEFAULT_MODEL_ID) ) self._structurers[default_model_id] = RadiologyReportStructurer( api_key=self.gemini_api_key, model_id=default_model_id, ) logger.info( f"RadExtract ready [Worker {os.getpid()}] with model: {default_model_id}" ) def _get_structurer(self, model_id: str) -> RadiologyReportStructurer: """Returns a cached or newly created structurer for the given model ID. Args: model_id: Identifier for the specific model configuration. Returns: RadiologyReportStructurer instance for the specified model. """ resolved_model_id = resolve_model_id(model_id) if resolved_model_id not in self._structurers: logger.info(f"Creating structurer for model: {resolved_model_id}") self._structurers[resolved_model_id] = RadiologyReportStructurer( api_key=self.gemini_api_key, model_id=resolved_model_id, ) return self._structurers[resolved_model_id] def predict(self, data: str, model_id: str) -> ResponseDict: """Processes prediction request using the specified model. Args: data: Input text data to be processed. model_id: Identifier for the model to use for processing. Returns: Dictionary containing the structured prediction results. """ logger.info("Processing prediction with model: %s", model_id) structurer = self._get_structurer(model_id) result = structurer.predict(data) logger.info("Prediction completed with model: %s", structurer.model_id) return result model = Model() # Copy prebuilt cache to writable location if it exists def setup_cache(): """Sets up the cache directory and copies prebuilt cache files. Creates a writable cache directory in /tmp and copies any existing prebuilt cache files to ensure the latest version is available. Returns: Path to the configured cache directory. """ cache_dir = tempfile.gettempdir() + "/cache" os.makedirs(cache_dir, exist_ok=True) source_cache = "cache/sample_cache.json" target_cache = os.path.join(cache_dir, "sample_cache.json") if os.path.exists(source_cache) and not os.path.exists(target_cache): shutil.copy2(source_cache, target_cache) logger.info(f"Initialized cache with {os.path.getsize(target_cache)} bytes") return cache_dir cache_dir = setup_cache() cache_manager = CacheManager(cache_dir=cache_dir) app = Flask( __name__, static_url_path="/static", static_folder="static", template_folder="templates", ) # Initialize rate limiter limiter = Limiter( get_remote_address, app=app, default_limits=[ os.environ.get("RATE_LIMIT_DAY", "200 per day"), os.environ.get("RATE_LIMIT_HOUR", "50 per hour"), ], storage_uri="memory://", ) @app.route("/") def index(): """Renders the main application interface. Returns: Rendered HTML template for the application index page. """ # Get social sharing context social_context = SocialSharingConfig.get_sharing_context(request.url_root) return render_template("index.html", **social_context) @app.route("/cache/stats") def cache_stats(): """Returns cache performance statistics. Returns: JSON response containing cache usage and performance statistics. """ return jsonify(cache_manager.get_cache_stats()) @app.route("/predict", methods=["POST"]) @limiter.limit(os.environ.get("RATE_LIMIT_PREDICT", "100 per hour")) def predict(): """Processes radiology report text and returns structured results. Accepts raw text via POST request body with optional headers for caching, sample identification, and model selection. Supports both cached and real-time processing modes. Returns: JSON response containing structured report segments, annotations, and formatted text. Includes cache status when applicable. Raises: 500: If processing fails due to invalid input or model errors. """ start_time = time.time() try: data = request.get_data(as_text=True) # Validate input to ensure it meets API requirements if not data or not data.strip(): return ( jsonify( { "error": "Empty input", "message": "Input text is required", "max_length": MAX_INPUT_LENGTH, } ), 400, ) if len(data) > MAX_INPUT_LENGTH: return ( jsonify( { "error": "Input too long", "message": ( f"Input length ({len(data)} characters) exceeds " "maximum allowed length of " f"{MAX_INPUT_LENGTH} characters" ), "max_length": MAX_INPUT_LENGTH, } ), 400, ) use_cache = request.headers.get("X-Use-Cache", "true").lower() == "true" requested_sample_id = request.headers.get("X-Sample-ID") model_id = resolve_model_id( request.headers.get( "X-Model-ID", os.environ.get("MODEL_ID", DEFAULT_MODEL_ID) ) ) if model_id not in _allowed_request_models(): return ( jsonify( { "error": "Unsupported model", "message": ( "The requested model is not available in this demo." ), } ), 400, ) processed_data = preprocess_report(data) sample_id = _verified_sample_id(requested_sample_id, processed_data) if use_cache and sample_id: cached_result = cache_manager.get_cached_result(processed_data, sample_id) if cached_result: req_id = hashlib.md5( f"{request.remote_addr}{int(time.time()/3600)}".encode() ).hexdigest()[:8] logger.info( "CACHE HIT [Req %s] [Worker %s] - Returning prebuilt result", req_id, os.getpid(), ) return jsonify({"from_cache": True, **cached_result}) try: req_id = hashlib.md5( f"{request.remote_addr}{int(time.time()/3600)}".encode() ).hexdigest()[:8] logger.info( "API CALL [Req %s] [Worker %s] - Processing with model %s", req_id, os.getpid(), model_id, ) result = model.predict(processed_data, model_id=model_id) result["sanitized_input"] = processed_data logger.info( "Prediction succeeded [Req %s] [Worker %s] with model %s in %.2fs", req_id, os.getpid(), model_id, time.time() - start_time, ) return jsonify(result) except ReportProcessingError as error: cause_type = ( type(error.__cause__).__name__ if error.__cause__ is not None else type(error).__name__ ) logger.error("Prediction provider failure: %s", cause_type) return ( jsonify( { "error": "Processing unavailable", "message": ( "The report could not be processed right now. " "Please try again later." ), } ), 502, ) except Exception as error: logger.error("Prediction failure: %s", type(error).__name__) return ( jsonify( { "error": "Internal processing error", "message": ( "The request could not be completed. Please try again later." ), } ), 500, ) @app.errorhandler(429) def ratelimit_handler(e): """Handle rate limit exceeded errors.""" return ( jsonify( { "error": "Rate limit exceeded. Please try again later.", "message": str(e.description), } ), 429, ) if __name__ == "__main__": logger.info("Starting development server") app.run(host="0.0.0.0", port=7870, debug=True)