File size: 5,590 Bytes
59e413f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 |
from fastapi import FastAPI, File, Form, UploadFile, HTTPException
from fastapi.responses import FileResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import boto3
import os
from botocore.exceptions import ClientError
import logging
import wikipedia
from gtts import gTTS
from PIL import Image, ImageEnhance
import uuid
import shutil
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
aws_access_key = os.getenv("AWS_ACCESS_KEY_ID", "AKIA6G75DYEK3NWC2AXH")
aws_secret_key = os.getenv("AWS_SECRET_ACCESS_KEY", "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")
region_name = "us-east-1"
try:
rekognition = boto3.client(
"rekognition",
aws_access_key_id=aws_access_key,
aws_secret_access_key=aws_secret_key,
region_name=region_name
)
logger.info("Rekognition client initialized successfully.")
except Exception as e:
logger.error(f"Error initializing Rekognition client: {e}")
rekognition = None
KNOWN_LANDMARKS = {
"Pyramid": "Giza Pyramids",
"Eiffel Tower": "Eiffel Tower",
"Statue of Liberty": "Statue of Liberty",
"Great Wall": "Great Wall of China",
"Colosseum": "Colosseum",
"Taj Mahal": "Taj Mahal",
"Machu Picchu": "Machu Picchu",
"Christ the Redeemer": "Christ the Redeemer",
"Big Ben": "Big Ben",
"Leaning Tower of Pisa": "Leaning Tower of Pisa",
"Sydney Opera House": "Sydney Opera House",
"Mount Rushmore": "Mount Rushmore",
"Burj Khalifa": "Burj Khalifa"
}
def enhance_image(image_path):
with Image.open(image_path) as img:
logger.info("Enhancing image")
img = ImageEnhance.Contrast(img).enhance(1.5)
img = ImageEnhance.Sharpness(img).enhance(2.0)
img.thumbnail((640, 360), Image.Resampling.LANCZOS)
img.save(image_path)
def detect_landmark(image_path):
with open(image_path, "rb") as image_file:
response = rekognition.detect_labels(Image={"Bytes": image_file.read()}, MaxLabels=10)
best_match = None
for label in response.get("Labels", []):
if label["Name"] in KNOWN_LANDMARKS:
return KNOWN_LANDMARKS[label["Name"]]
if best_match is None or label["Confidence"] > best_match[1]:
best_match = (label["Name"], label["Confidence"])
return best_match[0] if best_match else None
def get_wikipedia_info(object_name, lang="en"):
try:
wikipedia.set_lang(lang)
results = wikipedia.search(object_name)
if not results:
return "No results found on Wikipedia."
page = wikipedia.page(results[0])
return page.summary[:600]
except Exception as e:
logger.error(f"Wikipedia error: {e}")
return f"Error fetching Wikipedia info: {e}"
def text_to_speech(text, lang="en"):
try:
tts = gTTS(text=text, lang=lang)
audio_path = f"temp_audio_{uuid.uuid4()}.mp3"
tts.save(audio_path)
return audio_path
except Exception as e:
logger.error(f"TTS error: {e}")
return None
@app.post("/recognize")
async def recognize_landmark(image: UploadFile = File(...), language: str = Form(...)):
if rekognition is None:
raise HTTPException(status_code=500, detail="Rekognition client not initialized")
image_path = f"temp_image_{uuid.uuid4()}.jpg"
try:
with open(image_path, "wb") as buffer:
shutil.copyfileobj(image.file, buffer)
enhance_image(image_path)
landmark_name = detect_landmark(image_path)
if not landmark_name:
os.remove(image_path)
raise HTTPException(status_code=400, detail="No landmark recognized")
wiki_info = get_wikipedia_info(landmark_name, language)
if "error" in wiki_info.lower():
os.remove(image_path)
raise HTTPException(status_code=500, detail=wiki_info)
audio_path = text_to_speech(wiki_info, language)
if not audio_path:
os.remove(image_path)
raise HTTPException(status_code=500, detail="Failed to generate audio")
return {
"landmark": landmark_name,
"information": wiki_info,
"image_url": f"/image/{os.path.basename(image_path)}",
"audio_url": f"/audio/{os.path.basename(audio_path)}"
}
except ClientError as e:
raise HTTPException(status_code=500, detail=f"AWS error: {str(e)}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Server error: {str(e)}")
@app.get("/image/{filename}")
async def serve_image(filename: str):
image_path = os.path.join(os.getcwd(), filename)
if not os.path.exists(image_path):
raise HTTPException(status_code=404, detail="Image not found")
response = FileResponse(image_path, media_type="image/jpeg")
try:
os.remove(image_path)
except Exception as e:
logger.warning(f"Could not delete image: {e}")
return response
@app.get("/audio/{filename}")
async def serve_audio(filename: str):
audio_path = os.path.join(os.getcwd(), filename)
if not os.path.exists(audio_path):
raise HTTPException(status_code=404, detail="Audio not found")
response = FileResponse(audio_path, media_type="audio/mpeg")
try:
os.remove(audio_path)
except Exception as e:
logger.warning(f"Could not delete audio: {e}")
return response
|