Spaces:
Sleeping
Sleeping
| #!/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" | |
| 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" | |
| } | |
| } | |
| async def health_check(): | |
| return { | |
| "status": "healthy", | |
| "service": "Address Detection and Summarization API", | |
| "version": "1.0.0" | |
| } | |
| 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 | |
| ) | |
| 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) | |