import json import logging import tempfile from typing import Dict, Optional from fastapi import FastAPI, HTTPException, UploadFile, File from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from starlette.responses import JSONResponse # Transformers from transformers import pipeline import numpy as np from dataclasses import dataclass logging.basicConfig( level=logging.INFO, filename="brave_haven_api.log", # log file ka naam filemode="a", # "a" = append, "w" = overwrite format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("brave_haven_api") # ----------------------------- # Data Models # ----------------------------- class TextIn(BaseModel): text: str # ----------------------------- # AI Model Manager # ----------------------------- class AIModelManager: def __init__(self): self.pipelines = {} self.load_models() def _try_load(self, task: str, model_id: str, alias: str): try: self.pipelines[alias] = pipeline( task, model=model_id, tokenizer=model_id, return_all_scores=True ) logger.info(f"Loaded model: {alias} -> {model_id}") except Exception as e: logger.error(f"Failed to load {alias} ({model_id}): {e}") def load_models(self): logger.info("Loading AI models...") self._try_load("text-classification", "cardiffnlp/twitter-roberta-base-emotion-multilabel-latest", "emotion") self._try_load("text-classification", "cardiffnlp/twitter-roberta-base-sentiment-latest", "sentiment") self._try_load("text-classification", "j-hartmann/emotion-english-distilroberta-base", "tone") tox_candidates = [ "unitary/toxic-bert", "martin-ha/toxic-comment-model", "s-nlp/roberta_toxicity_classifier" ] for mid in tox_candidates: if "toxicity" not in self.pipelines: self._try_load("text-classification", mid, "toxicity") def analyze_text(self, text: str) -> Dict: results = {} if 'emotion' in self.pipelines: E_res = self.pipelines['emotion'](text) results['emotions'] = {item['label'].lower(): float(item['score']) for item in E_res[0]} if 'sentiment' in self.pipelines: S_res = self.pipelines['sentiment'](text) results['sentiment'] = {item['label'].lower(): float(item['score']) for item in S_res[0]} if 'tone' in self.pipelines: T_res = self.pipelines['tone'](text) results['tone'] = {item['label'].lower(): float(item['score']) for item in T_res[0]} if 'toxicity' in self.pipelines: To_res = self.pipelines['toxicity'](text) results['toxicity'] = {item['label'].lower(): float(item['score']) for item in To_res[0]} return results # ----------------------------- # Unified Scoring Function # ----------------------------- import numpy as np def normalize_component(scores: dict, pos_labels: set, neg_labels: set, neutral_labels: set = None): """ Normalize multi-label component. Returns score (float) and label (str). """ if not scores: return 0.0, "Neutral" total = sum(scores.values()) if sum(scores.values()) > 0 else 1.0 pos_sum = sum(v for k, v in scores.items() if k in pos_labels) neg_sum = sum(v for k, v in scores.items() if k in neg_labels) neutral_sum = sum(v for k, v in scores.items() if neutral_labels and k in neutral_labels) score = (pos_sum - neg_sum) / total # Label selection if pos_sum > 0.5 * total and neg_sum < 0.2 * total: label = "Positive" elif neg_sum > 0.5 * total and pos_sum < 0.2 * total: label = "Negative" elif pos_sum > 0.3 * total and neg_sum > 0.3 * total: label = f"Mixed (Positive {round(pos_sum,2)} | Negative {round(neg_sum,2)})" else: label = "Neutral" return round(score, 3), label def compute_overall_analysis(analysis: dict) -> dict: # --- Sentiment --- sentiment_scores = {k.lower(): float(v) for k, v in analysis.get("sentiment", {}).items()} sentiment_score, sentiment_label = normalize_component( sentiment_scores, pos_labels={"positive", "pos"}, neg_labels={"negative", "neg"}, neutral_labels={"neutral"} ) # --- Tone --- tone_scores = {k.lower(): float(v) for k, v in analysis.get("tone", {}).items()} tone_score, tone_label = normalize_component( tone_scores, pos_labels={"joy", "love", "trust"}, neg_labels={"anger", "fear", "disgust", "sadness"} ) # --- Emotions --- emotion_scores = {k.lower(): float(v) for k, v in analysis.get("emotions", {}).items()} emotion_score, emotion_label = normalize_component( emotion_scores, pos_labels={"joy", "optimism", "love"}, neg_labels={"anger", "sadness", "fear", "disgust"} ) # --- Toxicity (using your old detailed logic) --- tox = analysis.get("toxicity", None) toxicity_score, toxicity_label = 0.0, "Neutral" if tox is not None: max_tox = 0.0 if isinstance(tox, dict): max_tox = float(tox.get("toxic", 0.0)) if max_tox == 0.0 and len(tox) > 0: max_tox = float(max(tox.values())) elif isinstance(tox, (int, float)): max_tox = float(tox) toxicity_score = -max_tox toxicity_label = "Negative" if max_tox > 0.4 else "NON-TOCIX" # If toxicity is very low, ignore it if abs(toxicity_score) < 0.04: toxicity_score = 0.0 # --- Final weighted score --- final_score = ( sentiment_score * 0.25+ tone_score * 0.25 + emotion_score * 0.25 + toxicity_score * 0.25 ) # --- Override rules --- if abs(toxicity_score) > 0.7: final_label = "Negative" elif sentiment_score > 0.9 and tone_score > 0.9 and emotion_score > 0.9 and toxicity_score > -0.1: final_label = "Positive" final_score = round((sentiment_score + tone_score + emotion_score) / 3, 2) elif sentiment_score > 0.6 and emotion_score > 0.6 and toxicity_score > -0.4: final_label = "Positive" final_score = round((sentiment_score + emotion_score) / 2, 2) else: final_label = "Positive" if final_score > 0 else ("Negative" if final_score < 0 else "Neutral") return { "raw": analysis, "overall": { "components": { "sentiment_overall": {"label": sentiment_label, "score": round(sentiment_score, 2)}, "tone_overall": {"label": tone_label.title(), "score": round(tone_score, 2)}, "emotion_overall": {"label": emotion_label.title(), "score": round(emotion_score, 2)}, "toxicity_overall": {"label": toxicity_label, "score": round(toxicity_score, 2)}, }, "final_overall": {"label": final_label, "score": round(final_score, 2)} } } # ----------------------------- # FastAPI app and endpoints # ----------------------------- app = FastAPI(title="BraveHaven - Text & Voice Analysis API") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # Replace the direct instantiation with a lazy loader _ai_manager = None def get_ai_manager(): global _ai_manager if _ai_manager is None: _ai_manager = AIModelManager() return _ai_manager @app.get("/") def root(): return {"status": "ok", "message": "BraveHaven analysis API running"} @app.post("/analyze/text") def analyze_text(payload: TextIn): text = payload.text if not text.strip(): raise HTTPException(status_code=400, detail="Empty text provided") try: raw = get_ai_manager().analyze_text(text) overall = compute_overall_analysis(raw) return JSONResponse(content=overall) except Exception as e: logger.exception("Text analysis failed") raise HTTPException(status_code=500, detail=str(e)) @app.post("/analyze/voice") async def analyze_voice(file: UploadFile = File(...)): if not file: raise HTTPException(status_code=400, detail="No file uploaded") try: suffix = "." + file.filename.split(".")[-1] if "." in file.filename else "" with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: contents = await file.read() tmp.write(contents) tmp_path = tmp.name # SpeechRecognition import speech_recognition as sr recognizer = sr.Recognizer() with sr.AudioFile(tmp_path) as source: audio = recognizer.record(source) transcription = recognizer.recognize_google(audio) raw = get_ai_manager().analyze_text(transcription) overall = compute_overall_analysis(raw) return JSONResponse(content=overall) except Exception as e: logger.exception("Voice analysis failed") raise HTTPException(status_code=500, detail=str(e))