Spaces:
Sleeping
Sleeping
| from flask import Blueprint, request, jsonify | |
| from werkzeug.utils import secure_filename | |
| import os | |
| import easyocr | |
| import pytesseract # Ensure this is imported | |
| from PIL import Image | |
| from app.models import audio_model, sentiment_pipeline, emotion_pipeline | |
| from app.services import extract_tasks | |
| from app.utils import generate_tags, error_response | |
| # Initialize Flask Blueprint | |
| bp = Blueprint('main', __name__) | |
| # Initialize the EasyOCR reader for English only (disable GPU if not available) | |
| reader = easyocr.Reader(['en'], gpu=False) | |
| EMOTION_SCORE_THRESHOLD = 0.15 # Adjust based on your testing | |
| MIN_SENTIMENT_CONFIDENCE = 0.4 # Below this becomes "neutral" | |
| # ============================= | |
| # ๐น API Routes | |
| # ============================= | |
| def transcribe(): | |
| if 'file' not in request.files: | |
| return error_response("No file provided", 400) | |
| file = request.files['file'] | |
| file_path = os.path.join("/tmp", secure_filename(file.filename)) | |
| file.save(file_path) | |
| try: | |
| # Transcribe Audio | |
| result = audio_model.transcribe(file_path) | |
| transcription = result.get("text", "") | |
| if not transcription.strip(): | |
| return error_response("Transcription is empty", 400) | |
| # Send transcription to /analyze_text API | |
| analysis_response = analyze_text_internal(transcription) | |
| tags = generate_tags(transcription) # Function to extract tags from text | |
| return jsonify({ | |
| "transcription": transcription, | |
| "sentiment": analysis_response["sentiment"], | |
| "emotion": analysis_response["emotion"], | |
| "confidence": analysis_response["confidence"], | |
| "tags": tags | |
| }) | |
| except Exception as e: | |
| return error_response(str(e), 500) | |
| def analyze_image(): | |
| if 'file' not in request.files: | |
| return error_response("No image file provided", 400) | |
| file = request.files['file'] | |
| filename = secure_filename(file.filename) | |
| file_path = os.path.join("/tmp", filename) | |
| file.save(file_path) | |
| try: | |
| # Use EasyOCR in detail mode to get confidence scores | |
| results = reader.readtext(file_path, detail=1) | |
| # Filter out entries with low confidence (e.g., below 0.5) | |
| filtered_texts = [text for bbox, text, conf in results if conf > 0.5] | |
| extracted_text = "\n".join(filtered_texts) | |
| print("Filtered Extracted text:", extracted_text) | |
| if not extracted_text.strip(): | |
| return error_response("No meaningful text found in image", 400) | |
| # Analyze the extracted text to get sentiment, emotion, etc. | |
| analysis_response = analyze_text_internal(extracted_text) | |
| tags = generate_tags(extracted_text) | |
| return jsonify({ | |
| "extracted_text": extracted_text.strip(), | |
| "sentiment": analysis_response.get("sentiment"), | |
| "emotion": analysis_response.get("emotion"), | |
| "confidence": analysis_response.get("confidence"), | |
| "tags": tags | |
| }) | |
| except Exception as e: | |
| return error_response(str(e), 500) | |
| # Internal function to call analyze_text directly | |
| def analyze_text_internal(text): | |
| try: | |
| # Get sentiment (positive/neutral/negative) | |
| sentiment = sentiment_pipeline(text)[0] | |
| # Get dominant emotion (anger/disgust/fear/joy/neutral/sadness/surprise) | |
| emotion = emotion_pipeline(text)[0][0] | |
| return { | |
| "sentiment": sentiment['label'], | |
| "emotion": emotion['label'], | |
| "confidence": { | |
| "sentiment": round(sentiment['score'], 3), | |
| "emotion": round(emotion['score'], 3) | |
| } | |
| } | |
| except Exception as e: | |
| print(f"Analysis error: {str(e)}") | |
| return error_response(f"Processing error: {str(e)}", 500) | |
| def analyze_text(): | |
| data = request.json | |
| if not data or 'text' not in data: | |
| return error_response("No text provided", 400) | |
| text = data['text'].strip().lower() | |
| try: | |
| # Get sentiment (positive/neutral/negative) | |
| sentiment = sentiment_pipeline(text)[0] | |
| # Get dominant emotion (anger/disgust/fear/joy/neutral/sadness/surprise) | |
| emotion = emotion_pipeline(text)[0][0] | |
| tags = generate_tags(text) | |
| return { | |
| "sentiment": sentiment['label'], | |
| "emotion": emotion['label'], | |
| "confidence": { | |
| "sentiment": round(sentiment['score'], 3), | |
| "emotion": round(emotion['score'], 3) | |
| }, | |
| "tags": tags | |
| } | |
| except Exception as e: | |
| print(f"Analysis error: {str(e)}") | |
| return error_response(f"Processing error: {str(e)}", 500) | |
| # ๐ 3. Extract Actionable Tasks | |
| def extract_actions(): | |
| data = request.json | |
| if not data or 'text' not in data: | |
| return error_response("No text provided", 400) | |
| text = data['text'] | |
| try: | |
| tasks = extract_tasks(text) | |
| return jsonify({"tasks": tasks}) | |
| except Exception as e: | |
| return error_response(str(e), 500) | |
| # ============================= | |
| # ๐น Error Handling | |
| # ============================= | |
| def not_found_error(error): | |
| return error_response("Not Found", 404) | |
| def internal_error(error): | |
| return error_response("Internal Server Error", 500) | |