Spaces:
Sleeping
Sleeping
| import os | |
| import re | |
| import json | |
| import tempfile | |
| import random | |
| from datetime import datetime | |
| from typing import List, Dict, Any, Tuple | |
| import dspy | |
| from dataclasses import dataclass | |
| import gradio as gr | |
| from dotenv import load_dotenv | |
| import os | |
| load_dotenv() | |
| api_key = os.getenv("API_KEY") | |
| if api_key: | |
| openrouter_lm = dspy.LM( | |
| model="openrouter/deepseek/deepseek-chat-v3-0324:free", | |
| api_base="https://openrouter.ai/api/v1", | |
| api_key=api_key | |
| ) | |
| dspy.configure(lm=openrouter_lm) | |
| class ChatMessage: | |
| timestamp: str | |
| sender: str | |
| message: str | |
| datetime_obj: datetime | |
| # === ENHANCED LLM SIGNATURES === | |
| class DeepContextualLoveAnalysis(dspy.Signature): | |
| """Advanced contextual analysis of love and emotional connection through conversation patterns""" | |
| conversation_context = dspy.InputField(desc="Balanced sample of conversation messages across different time periods") | |
| user_name = dspy.InputField(desc="Name of the user receiving messages") | |
| partner_name = dspy.InputField(desc="Name of the partner sending messages") | |
| total_messages = dspy.InputField(desc="Total number of messages in conversation") | |
| love_percentage = dspy.OutputField(desc="Love intensity score (0-100) based on emotional expression, care, and affection patterns") | |
| emotional_connection = dspy.OutputField(desc="Quality and depth of emotional bond demonstrated") | |
| communication_style = dspy.OutputField(desc="How partner expresses feelings and maintains connection") | |
| relationship_investment = dspy.OutputField(desc="Level of investment and commitment shown in conversations") | |
| affection_patterns = dspy.OutputField(desc="Specific ways love and care are expressed") | |
| intimacy_level = dspy.OutputField(desc="Emotional intimacy and closeness demonstrated") | |
| class RelationshipDynamicsAnalysis(dspy.Signature): | |
| """Analyze overall relationship health and compatibility""" | |
| love_context = dspy.InputField(desc="Results from deep love analysis") | |
| conversation_dynamics = dspy.InputField(desc="Overall conversation patterns and frequency") | |
| relationship_health = dspy.OutputField(desc="Overall relationship wellness score (0-100)") | |
| compatibility_indicators = dspy.OutputField(desc="Signs of long-term compatibility") | |
| communication_quality = dspy.OutputField(desc="Effectiveness and warmth of communication") | |
| growth_potential = dspy.OutputField(desc="Potential for relationship development") | |
| unique_strengths = dspy.OutputField(desc="What makes this relationship special") | |
| future_outlook = dspy.OutputField(desc="Likely trajectory of the relationship") | |
| # === ENHANCED LOVE ANALYZER === | |
| class AdvancedLoveAnalyzer: | |
| def __init__(self): | |
| self.love_analyzer = dspy.Predict(DeepContextualLoveAnalysis) | |
| self.dynamics_analyzer = dspy.Predict(RelationshipDynamicsAnalysis) | |
| random.seed(42) # For reproducible sampling | |
| def parse_messages(self, text: str) -> List[ChatMessage]: | |
| """Enhanced message parsing with better error handling""" | |
| messages = [] | |
| patterns = [ | |
| r'\[(\d{2}/\d{2}/\d{2}, \d{2}\.\d{2}\.\d{2})\] ([^:]+): (.+)', | |
| r'(\d{2}/\d{2}/\d{2}, \d{2}\.\d{2}\.\d{2}) - ([^:]+): (.+)', | |
| r'(\d{1,2}/\d{1,2}/\d{2,4}, \d{1,2}:\d{2}) - ([^:]+): (.+)', | |
| r'(\d{1,2}/\d{1,2}/\d{2,4}, \d{1,2}:\d{2}:\d{2}) - ([^:]+): (.+)', | |
| ] | |
| for line in text.strip().split('\n'): | |
| line = line.strip() | |
| if not line or line.startswith('') or len(line) < 10: | |
| continue | |
| for pattern in patterns: | |
| match = re.match(pattern, line) | |
| if match: | |
| timestamp_str, sender, message = match.groups() | |
| try: | |
| # Enhanced timestamp parsing | |
| timestamp_formats = [ | |
| '%d/%m/%y, %H.%M.%S', | |
| '%d/%m/%y, %H:%M', | |
| '%m/%d/%Y, %H:%M', | |
| '%d/%m/%Y, %H:%M:%S', | |
| '%m/%d/%y, %H:%M' | |
| ] | |
| dt = None | |
| for fmt in timestamp_formats: | |
| try: | |
| dt = datetime.strptime(timestamp_str, fmt) | |
| break | |
| except ValueError: | |
| continue | |
| if dt: | |
| messages.append(ChatMessage( | |
| timestamp_str, | |
| sender.strip(), | |
| message.strip(), | |
| dt | |
| )) | |
| break | |
| except Exception: | |
| continue | |
| return sorted(messages, key=lambda x: x.datetime_obj) | |
| def identify_partner(self, messages: List[ChatMessage], user_name: str) -> str: | |
| """Smart partner identification""" | |
| participants = {} | |
| user_variations = [user_name.lower(), user_name.split()[0].lower(), "thariq", "arian"] | |
| for msg in messages: | |
| sender = msg.sender.strip() | |
| if sender in participants: | |
| participants[sender] += 1 | |
| else: | |
| participants[sender] = 1 | |
| # Find partner (not user) | |
| for participant, count in sorted(participants.items(), key=lambda x: x[1], reverse=True): | |
| if not any(var in participant.lower() for var in user_variations): | |
| return participant | |
| return list(participants.keys())[0] if participants else "Unknown" | |
| def balanced_sampling(self, messages: List[ChatMessage], sample_size: int = 200) -> List[ChatMessage]: | |
| """Create balanced random sample across conversation timeline""" | |
| if len(messages) <= sample_size: | |
| return messages | |
| # Sort by timestamp | |
| sorted_messages = sorted(messages, key=lambda x: x.datetime_obj) | |
| total_msgs = len(sorted_messages) | |
| # Divide into time segments for balanced sampling | |
| num_segments = min(10, total_msgs // 20) # At least 20 messages per segment | |
| segment_size = total_msgs // num_segments | |
| sampled_messages = [] | |
| messages_per_segment = sample_size // num_segments | |
| remaining_samples = sample_size % num_segments | |
| for i in range(num_segments): | |
| start_idx = i * segment_size | |
| end_idx = start_idx + segment_size if i < num_segments - 1 else total_msgs | |
| segment_messages = sorted_messages[start_idx:end_idx] | |
| # Sample from this segment | |
| segment_sample_size = messages_per_segment | |
| if i < remaining_samples: | |
| segment_sample_size += 1 | |
| if len(segment_messages) <= segment_sample_size: | |
| sampled_messages.extend(segment_messages) | |
| else: | |
| sampled_messages.extend(random.sample(segment_messages, segment_sample_size)) | |
| return sorted(sampled_messages, key=lambda x: x.datetime_obj) | |
| def analyze_love_from_file(self, file_path: str, user_name: str = "Thariq Arian") -> Dict[str, Any]: | |
| """Enhanced love analysis with balanced sampling and context focus""" | |
| # Parse messages | |
| with open(file_path, 'r', encoding='utf-8') as f: | |
| content = f.read() | |
| messages = self.parse_messages(content) | |
| if not messages: | |
| return {"error": "No messages found"} | |
| partner_name = self.identify_partner(messages, user_name) | |
| partner_messages = [msg for msg in messages if msg.sender == partner_name] | |
| if not partner_messages: | |
| return {"error": f"No messages found from partner: {partner_name}"} | |
| # Balanced sampling | |
| sampled_messages = self.balanced_sampling(partner_messages, 200) | |
| # Prepare context for LLM | |
| conversation_context = "\n".join([ | |
| f"[{msg.timestamp}] {msg.message}" | |
| for msg in sampled_messages | |
| ]) | |
| try: | |
| # Deep love analysis | |
| love_analysis = self.love_analyzer( | |
| conversation_context=conversation_context, | |
| user_name=user_name, | |
| partner_name=partner_name, | |
| total_messages=str(len(partner_messages)) | |
| ) | |
| # Relationship dynamics analysis | |
| dynamics = self.dynamics_analyzer( | |
| love_context=json.dumps({ | |
| 'love_percentage': love_analysis.love_percentage, | |
| 'emotional_connection': love_analysis.emotional_connection, | |
| 'communication_style': love_analysis.communication_style, | |
| 'relationship_investment': love_analysis.relationship_investment | |
| }), | |
| conversation_dynamics=f"Total: {len(partner_messages)}, Sampled: {len(sampled_messages)}, Time span: {partner_messages[0].timestamp} to {partner_messages[-1].timestamp}" | |
| ) | |
| return { | |
| 'user_name': user_name, | |
| 'partner_name': partner_name, | |
| 'total_messages': len(partner_messages), | |
| 'analyzed_sample': len(sampled_messages), | |
| 'love_analysis': { | |
| 'love_percentage': love_analysis.love_percentage, | |
| 'emotional_connection': love_analysis.emotional_connection, | |
| 'communication_style': love_analysis.communication_style, | |
| 'relationship_investment': love_analysis.relationship_investment, | |
| 'affection_patterns': love_analysis.affection_patterns, | |
| 'intimacy_level': love_analysis.intimacy_level | |
| }, | |
| 'relationship_dynamics': { | |
| 'relationship_health': dynamics.relationship_health, | |
| 'compatibility_indicators': dynamics.compatibility_indicators, | |
| 'communication_quality': dynamics.communication_quality, | |
| 'growth_potential': dynamics.growth_potential, | |
| 'unique_strengths': dynamics.unique_strengths, | |
| 'future_outlook': dynamics.future_outlook | |
| }, | |
| 'analysis_timestamp': datetime.now().isoformat() | |
| } | |
| except Exception as e: | |
| return {"error": str(e)} | |
| def extract_score(self, score_text: str) -> int: | |
| """Extract numeric score from LLM response""" | |
| try: | |
| # Look for percentage or number | |
| numbers = re.findall(r'\b(\d{1,3})\b', str(score_text)) | |
| if numbers: | |
| score = int(numbers[0]) | |
| return min(100, max(0, score)) # Clamp to 0-100 | |
| return 0 | |
| except: | |
| return 0 | |
| def generate_clean_report(self, analysis_result: Dict[str, Any]) -> str: | |
| """Generate clean, readable report without excessive characters""" | |
| if 'error' in analysis_result: | |
| return f"Analysis failed: {analysis_result['error']}" | |
| love_data = analysis_result['love_analysis'] | |
| dynamics = analysis_result['relationship_dynamics'] | |
| # Extract clean love score | |
| love_score = self.extract_score(love_data['love_percentage']) | |
| health_score = self.extract_score(dynamics['relationship_health']) | |
| # Simple love meter | |
| hearts = "❤️" * (love_score // 10) | |
| empty_hearts = "🤍" * (10 - love_score // 10) | |
| meter = hearts + empty_hearts | |
| # Love verdict | |
| if love_score >= 85: | |
| verdict = "DEEPLY IN LOVE" | |
| elif love_score >= 70: | |
| verdict = "STRONG LOVE" | |
| elif love_score >= 55: | |
| verdict = "GENUINE AFFECTION" | |
| elif love_score >= 40: | |
| verdict = "CARING FEELINGS" | |
| else: | |
| verdict = "FRIENDLY FEELINGS" | |
| report = f""" | |
| LOVE ANALYSIS REPORT | |
| {analysis_result['partner_name']} → {analysis_result['user_name']} | |
| LOVE SCORE: {love_score}% | |
| {meter} | |
| VERDICT: {verdict} | |
| EMOTIONAL ANALYSIS: | |
| Connection Level: {love_data['emotional_connection']} | |
| Communication Style: {love_data['communication_style']} | |
| Relationship Investment: {love_data['relationship_investment']} | |
| Affection Patterns: {love_data['affection_patterns']} | |
| Intimacy Level: {love_data['intimacy_level']} | |
| RELATIONSHIP DYNAMICS: | |
| Health Score: {health_score}% | |
| Compatibility: {dynamics['compatibility_indicators']} | |
| Communication Quality: {dynamics['communication_quality']} | |
| Growth Potential: {dynamics['growth_potential']} | |
| Unique Strengths: {dynamics['unique_strengths']} | |
| Future Outlook: {dynamics['future_outlook']} | |
| ANALYSIS SUMMARY: | |
| Total Messages: {analysis_result['total_messages']} | |
| Sample Analyzed: {analysis_result['analyzed_sample']} | |
| Analysis Date: {datetime.now().strftime('%Y-%m-%d %H:%M')} | |
| Advanced Context-Based Love Analyzer v4.0 | |
| """ | |
| return report | |
| # === GRADIO FUNCTIONS === | |
| def process_love_analysis(uploaded_file) -> Tuple[str, str]: | |
| """Process uploaded file and return analysis results""" | |
| if uploaded_file is None: | |
| return None, "❌ Please upload a .txt file first!" | |
| try: | |
| analyzer = AdvancedLoveAnalyzer() | |
| # Read uploaded file | |
| file_path = uploaded_file.name | |
| debug_info = f"📁 Processing file: {os.path.basename(file_path)}\n" | |
| # Analyze love | |
| debug_info += "🔮 Starting advanced love analysis...\n" | |
| result = analyzer.analyze_love_from_file(file_path, "Thariq Arian") | |
| if 'error' in result: | |
| debug_info += f"❌ Error: {result['error']}\n" | |
| return None, debug_info | |
| # Generate report | |
| debug_info += f"💕 Found partner: {result['partner_name']}\n" | |
| debug_info += f"📊 Analyzed {result['analyzed_sample']} out of {result['total_messages']} messages\n" | |
| debug_info += f"✨ Love score: {analyzer.extract_score(result['love_analysis']['love_percentage'])}%\n" | |
| debug_info += "🎉 Analysis completed successfully!" | |
| report = analyzer.generate_clean_report(result) | |
| # Save report to temporary file | |
| with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f: | |
| f.write(report) | |
| temp_file_path = f.name | |
| return temp_file_path, debug_info | |
| except Exception as e: | |
| error_msg = f"❌ Analysis failed: {str(e)}" | |
| return None, error_msg | |
| # === GRADIO APP === | |
| with gr.Blocks(title="Ultimate Crush Detector", theme=gr.themes.Monochrome()) as app: | |
| gr.Markdown( | |
| """ | |
| # 💘 <span style='color:#ff99cc;'>Ultimate Crush Detector</span> | |
| --- | |
| <p style='font-size:1.1em;'>Upload your <code>.txt</code> file and click <strong>Submit</strong> to reveal your ultimate crush ✨</p> | |
| """, | |
| elem_id="title-markdown" | |
| ) | |
| with gr.Column(scale=1): | |
| input_file = gr.File( | |
| label="💌 Upload your Love Letter (.txt)", | |
| file_types=[".txt"], | |
| elem_id="input-file" | |
| ) | |
| submit_btn = gr.Button("💖 Reveal My Crush 💖", variant="primary", elem_id="submit-btn") | |
| output_file = gr.File( | |
| label="📥 Download your result (.txt)" | |
| ) | |
| debug_box = gr.Textbox( | |
| label="🔮 Debug & Love Notes", | |
| placeholder="Status, secrets, and logs will appear here...", | |
| interactive=False, | |
| lines=5 | |
| ) | |
| submit_btn.click( | |
| process_love_analysis, | |
| inputs=input_file, | |
| outputs=[output_file, debug_box] | |
| ) | |
| # Advanced Love Theme Custom Styling — Full width | |
| app.css = """ | |
| body { | |
| background: radial-gradient(circle at top, #1a0023, #3a0b52, #8900a1); | |
| color: #ff99cc; | |
| font-family: 'Orbitron', sans-serif; | |
| } | |
| #title-markdown h1 { | |
| font-family: 'Pacifico', cursive; | |
| font-weight: 800; | |
| text-shadow: 0 0 15px #ff99cc; | |
| } | |
| .gr-button { | |
| background: linear-gradient(135deg, #ff66b2 0%, #ff0080 100%); | |
| border: none; | |
| color: #fff; | |
| font-weight: bold; | |
| font-size: 1.2em; | |
| width: 100% !important; | |
| padding: 1em; | |
| border-radius: 12px; | |
| } | |
| .gr-button:hover { | |
| box-shadow: 0 0 20px #ff66b2; | |
| transform: scale(1.02); | |
| } | |
| #input-file, .gr-file, .gr-file .upload-box { | |
| width: 100% !important; | |
| } | |
| """ | |
| if __name__ == "__main__": | |
| app.launch(share=False) |