from flask import Flask, request, jsonify from flask_cors import CORS import os from transformers import pipeline import numpy as np import torch import re from werkzeug.utils import secure_filename import uuid import platform # Set Transformers Cache Directory if platform.system() == "Windows": print("Windows detected. Assigning cache directory to Transformers in AppData\\Local.") transformers_cache_directory = os.path.join(os.getenv('LOCALAPPDATA'), 'transformers_cache') else: print("Non-Windows system detected. Assigning cache directory to /tmp/transformers_cache.") transformers_cache_directory = '/tmp/transformers_cache' # Ensure the directory exists if not os.path.exists(transformers_cache_directory): try: os.makedirs(transformers_cache_directory, exist_ok=True) print(f"Directory '{transformers_cache_directory}' created successfully.") except OSError as e: print(f"Error creating directory '{transformers_cache_directory}': {e}") else: print(f"Directory '{transformers_cache_directory}' already exists.") # Set the TRANSFORMERS_CACHE environment variable os.environ['TRANSFORMERS_CACHE'] = transformers_cache_directory print(f"Environment variable TRANSFORMERS_CACHE set to '{transformers_cache_directory}'.") class Config: UPLOAD_FOLDER = os.path.join(os.path.dirname(__file__), '/tmp/uploads') # Correct path MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB max file size CORS_HEADERS = 'Content-Type' class DialogueSentimentAnalyzer: def __init__(self, model_name: str = "microsoft/DialogRPT-updown"): self.device = 0 if torch.cuda.is_available() else -1 self.dialogue_model = pipeline( 'text-classification', model="microsoft/DialogRPT-updown", device=self.device ) self.sentiment_model = pipeline( 'sentiment-analysis', model="distilbert-base-uncased-finetuned-sst-2-english", device=self.device ) self.max_length = 512 def parse_dialogue(self, text: str): lines = text.strip().split('\n') dialogue = [] current_speaker = None current_text = [] for line in lines: line = line.strip() if not line: continue speaker_match = re.match(r'^([^:]+):', line) if speaker_match: if current_speaker and current_text: dialogue.append({'speaker': current_speaker, 'text': ' '.join(current_text)}) current_speaker = speaker_match.group(1) current_text = [line[len(current_speaker) + 1:].strip()] else: if current_speaker: current_text.append(line.strip()) if current_speaker and current_text: dialogue.append({'speaker': current_speaker, 'text': ' '.join(current_text)}) return dialogue def analyze_utterance(self, utterance): text = utterance['text'] dialogue_score = self.dialogue_model(text)[0] sentiment = self.sentiment_model(text)[0] positive_phrases = ['thank you', 'thanks', 'appreciate', 'great', 'perfect', 'looking forward', 'flexible', 'competitive'] negative_phrases = ['concerned', 'worry', 'issue', 'problem', 'difficult', 'unfortunately', 'sorry'] text_lower = text.lower() positive_count = sum(1 for phrase in positive_phrases if phrase in text_lower) negative_count = sum(1 for phrase in negative_phrases if phrase in text_lower) sentiment_score = float(sentiment['score']) if sentiment['label'] == 'NEGATIVE': sentiment_score = 1 - sentiment_score final_score = sentiment_score if positive_count > negative_count: final_score = min(1.0, final_score + 0.1 * (positive_count - negative_count)) elif negative_count > positive_count: final_score = max(0.0, final_score - 0.1 * (negative_count - positive_count)) return { 'speaker': utterance['speaker'], 'text': text, 'sentiment_score': final_score, 'engagement_score': float(dialogue_score['score']), 'positive_phrases': positive_count, 'negative_phrases': negative_count } def analyze_dialogue(self, text: str): dialogue = self.parse_dialogue(text) utterance_results = [self.analyze_utterance(utterance) for utterance in dialogue] overall_sentiment = np.mean([r['sentiment_score'] for r in utterance_results]) overall_engagement = np.mean([r['engagement_score'] for r in utterance_results]) sentiment_variance = np.std([r['sentiment_score'] for r in utterance_results]) confidence = max(0.0, 1.0 - sentiment_variance) speaker_sentiments = {} for result in utterance_results: if result['speaker'] not in speaker_sentiments: speaker_sentiments[result['speaker']] = [] speaker_sentiments[result['speaker']].append(result['sentiment_score']) speaker_averages = {speaker: np.mean(scores) for speaker, scores in speaker_sentiments.items()} return [{'label': 'Overall Sentiment', 'score': float(overall_sentiment)}, {'label': 'Confidence', 'score': float(confidence)}, {'label': 'Engagement', 'score': float(overall_engagement)}] + [ {'label': f'{speaker} Sentiment', 'score': float(score)} for speaker, score in speaker_averages.items() ] def save_uploaded_file(content, upload_folder): filename = f"{uuid.uuid4().hex}.txt" file_path = os.path.join(upload_folder, secure_filename(filename)) with open(file_path, 'w', encoding='utf-8') as f: f.write(content) return file_path def analyze_sentiment(file_path: str): try: analyzer = DialogueSentimentAnalyzer() with open(file_path, 'r', encoding='utf-8') as f: text = f.read() return analyzer.analyze_dialogue(text) except Exception as e: print(f"Error in sentiment analysis: {str(e)}") return [{'label': 'Error', 'score': 0.5}] def create_app(): app = Flask(__name__) app.config.from_object(Config) # Ensure the uploads directory exists os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) @app.route('/upload', methods=['POST']) def upload_transcript(): try: transcript = request.form.get('transcript') if not transcript: return jsonify({'error': 'No transcript received'}), 400 # Save the transcript in the current folder file_path = os.path.join(os.getcwd(), 'transcript.txt') with open(file_path, 'w') as file: file.write(transcript) # Analyze sentiment sentiment_result = analyze_sentiment(file_path) # Remove the temporary file os.remove(file_path) return jsonify({'sentiment': sentiment_result}), 200 except Exception as e: return jsonify({'error': str(e)}), 500 return app if __name__ == '__main__': app = create_app() app.run(host="0.0.0.0", port=5000)