#!/usr/bin/env python3 """ Address Detection and Summarization API for Hugging Face Spaces """ from fastapi import FastAPI, HTTPException, Header from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from openai import OpenAI import os from enum import Enum from typing import Dict, Optional, Annotated import logging from dotenv import load_dotenv # Load environment variables load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="Address Detection and Summarization API") # Add CORS middleware for Hugging Face Spaces app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class Actions(Enum): DETECT_ADDRESS = "detect_address" SUMMARIZE = "summarize" # Request model class RequestModel(BaseModel): entity_urn: str = Field(..., description="Unique identifier for the entity") content: str = Field(..., description="Content to be processed") # Response models class ResponseModel(BaseModel): message: str = Field(description="Result of the operation, e.g., 'success' or 'failure'") result: str = Field(description="summary in case of summarization, or || separated addresses in case of address detection") action_type: Actions = Field(..., description="Type of action performed") entity_urn: str = Field(..., description="Unique identifier for the entity") sentiment: Optional[Dict] = Field(default=None, description="Sentiment analysis result") def get_openai_client(api_key: str) -> OpenAI: """Create OpenAI client with provided API key""" if not api_key: raise HTTPException(status_code=400, detail="x-api-key header is required") return OpenAI(api_key=api_key) def get_openai_response(api_key: str, prompt: str, max_tokens: int = 10000) -> str: """Get response from OpenAI API""" try: client = get_openai_client(api_key) response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], max_tokens=max_tokens, temperature=0.3 ) return response.choices[0].message.content.strip() except Exception as e: logger.error(f"OpenAI API error: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to connect to OpenAI: {str(e)}") def detect_addresses(api_key: str, content: str) -> str: """Detect addresses from content using OpenAI""" prompt = f""" Extract all addresses from the following text. Return only the addresses, separated by ' || '. If no addresses are found, return empty string. Text: {content} Addresses: """ try: addresses = get_openai_response(api_key, prompt) return addresses if addresses else "" except Exception as e: logger.error(f"Address detection error: {str(e)}") return "Error detecting addresses" def summarize_and_analyze_sentiment(api_key: str, content: str) -> tuple: """Summarize content and analyze sentiment using OpenAI""" prompt = f""" Please provide: 1. A concise summary of the following text 2. Sentiment analysis with label (POSITIVE/NEGATIVE) and score (-1.0 to 1.0) 3. Ensure the summary is clear and captures the main points Format your response as: SUMMARY: [your summary here] SENTIMENT_LABEL: [POSITIVE or NEGATIVE] SENTIMENT_SCORE: [score between -1.0 and 1.0] Text: {content} """ try: response = get_openai_response(api_key, prompt) # Parse the response summary = "" sentiment_label = "neutral" sentiment_score = 0.0 lines = response.split('\n') for line in lines: if line.startswith('SUMMARY:'): summary = line.replace('SUMMARY:', '').strip() elif line.startswith('SENTIMENT_LABEL:'): sentiment_label = line.replace('SENTIMENT_LABEL:', '').strip() elif line.startswith('SENTIMENT_SCORE:'): try: sentiment_score = float(line.replace('SENTIMENT_SCORE:', '').strip()) # Ensure score is within range sentiment_score = max(-1.0, min(1.0, sentiment_score)) except ValueError: sentiment_score = 0.0 # Ensure valid sentiment label if sentiment_label not in ['POSITIVE', 'NEGATIVE']: sentiment_label = 'POSITIVE' if sentiment_score >= 0 else 'NEGATIVE' return summary, {"label": sentiment_label, "score": sentiment_score} except Exception as e: logger.error(f"Summarization error: {str(e)}") return "Error generating summary" @app.get("/") async def root(): return { "message": "Address Detection and Summarization API", "version": "1.0.0", "status": "healthy", "endpoints": { "address_detection": "/address-detection", "summarization": "/summarize", "health": "/health", "docs": "/docs" } } @app.get("/health") async def health_check(): return { "status": "healthy", "service": "Address Detection and Summarization API", "version": "1.0.0" } @app.post("/address-detection", response_model=ResponseModel) async def address_detection( request: RequestModel, openai_api_key: Annotated[str, Header(alias="x-api-key")] ): """Endpoint for address detection""" try: logger.info(f"Processing address detection for entity_urn: {request.entity_urn}") if not request.content.strip(): return ResponseModel( message="failure", result="Content is empty", action_type=Actions.DETECT_ADDRESS, entity_urn=request.entity_urn, sentiment=None ) addresses = detect_addresses(openai_api_key, request.content) return ResponseModel( message="success", result=addresses, action_type=Actions.DETECT_ADDRESS, entity_urn=request.entity_urn, sentiment=None ) except Exception as e: logger.error(f"Address detection endpoint error: {str(e)}") return ResponseModel( message="failure", result=f"Error: {str(e)}", action_type=Actions.DETECT_ADDRESS, entity_urn=request.entity_urn, sentiment=None ) @app.post("/summarize", response_model=ResponseModel) async def summarize( request: RequestModel, openai_api_key: Annotated[str, Header(alias="x-api-key")] ): """Endpoint for text summarization and sentiment analysis""" try: logger.info(f"Processing summarization for entity_urn: {request.entity_urn}") if not request.content.strip(): return ResponseModel( message="failure", result="Content is empty", action_type=Actions.SUMMARIZE, entity_urn=request.entity_urn, sentiment={} ) summary, sentiment = summarize_and_analyze_sentiment(openai_api_key, request.content) return ResponseModel( message="success", result=summary, sentiment=sentiment, action_type=Actions.SUMMARIZE, entity_urn=request.entity_urn ) except Exception as e: logger.error(f"Summarization endpoint error: {str(e)}") return ResponseModel( message="failure", result=f"Error: {str(e)}", sentiment={}, action_type=Actions.SUMMARIZE, entity_urn=request.entity_urn ) if __name__ == "__main__": import uvicorn # Use port 7860 for Hugging Face Spaces, fallback to 8500 for local port = int(os.getenv("PORT", 7860)) uvicorn.run(app, host="0.0.0.0", port=port)