File size: 6,093 Bytes
430b54f a453c29 430b54f a453c29 430b54f |
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 164 165 166 |
"""Service for analysis operations: processing arguments, extracting topics, and predicting stance"""
import logging
from typing import List, Dict, Optional
from datetime import datetime
from services.database_service import database_service
from services.topic_similarity_service import topic_similarity_service
from services.stance_model_manager import stance_model_manager
logger = logging.getLogger(__name__)
class AnalysisService:
"""Service for analyzing arguments with topic extraction and stance prediction"""
def __init__(self):
self.table_name = "analysis_results"
def _get_client(self):
"""Get Supabase client"""
return database_service.get_client()
def analyze_arguments(
self,
user_id: str,
arguments: List[str]
) -> List[Dict]:
"""
Analyze arguments: extract topics, predict stance, and save to database
Args:
user_id: User UUID
arguments: List of argument texts
Returns:
List of analysis results with argument, topic, and stance prediction
"""
try:
if not arguments or len(arguments) == 0:
raise ValueError("Arguments list cannot be empty")
logger.info(f"Starting analysis for {len(arguments)} arguments for user {user_id}")
# Step 1: Find most similar topics for all arguments
logger.info("Step 1: Finding most similar topics...")
topics = topic_similarity_service.batch_find_similar_topics(arguments)
if len(topics) != len(arguments):
raise RuntimeError(f"Topic extraction returned {len(topics)} topics but expected {len(arguments)}")
# Step 2: Predict stance for each argument-topic pair
logger.info("Step 2: Predicting stance for argument-topic pairs...")
stance_results = []
for arg, topic in zip(arguments, topics):
if topic is None:
logger.warning(f"Skipping argument with null topic: {arg[:50]}...")
continue
stance_result = stance_model_manager.predict(topic, arg)
stance_results.append({
"argument": arg,
"topic": topic,
"predicted_stance": stance_result["predicted_stance"],
"confidence": stance_result["confidence"],
"probability_con": stance_result["probability_con"],
"probability_pro": stance_result["probability_pro"],
})
# Step 3: Save all results to database
logger.info(f"Step 3: Saving {len(stance_results)} results to database...")
saved_results = self._save_analysis_results(user_id, stance_results)
logger.info(f"Analysis completed: {len(saved_results)} results saved")
return saved_results
except Exception as e:
logger.error(f"Error in analyze_arguments: {str(e)}")
raise RuntimeError(f"Analysis failed: {str(e)}")
def _save_analysis_results(
self,
user_id: str,
results: List[Dict]
) -> List[Dict]:
"""
Save analysis results to database
Args:
user_id: User UUID
results: List of analysis result dictionaries
Returns:
List of saved results with database IDs
"""
try:
client = self._get_client()
# Prepare data for batch insert
insert_data = []
for result in results:
insert_data.append({
"user_id": user_id,
"argument": result["argument"],
"topic": result["topic"],
"predicted_stance": result["predicted_stance"],
"confidence": result["confidence"],
"probability_con": result["probability_con"],
"probability_pro": result["probability_pro"],
"created_at": datetime.utcnow().isoformat(),
"updated_at": datetime.utcnow().isoformat()
})
# Batch insert
response = client.table(self.table_name).insert(insert_data).execute()
if response.data:
logger.info(f"Successfully saved {len(response.data)} analysis results")
return response.data
else:
raise RuntimeError("Failed to save analysis results: no data returned")
except Exception as e:
logger.error(f"Error saving analysis results: {str(e)}")
raise RuntimeError(f"Failed to save analysis results: {str(e)}")
def get_user_analysis_results(
self,
user_id: str,
limit: Optional[int] = 100,
offset: Optional[int] = 0
) -> List[Dict]:
"""
Get analysis results for a user
Args:
user_id: User UUID
limit: Maximum number of results to return
offset: Number of results to skip
Returns:
List of analysis results
"""
try:
client = self._get_client()
query = client.table(self.table_name)\
.select("*")\
.eq("user_id", user_id)\
.order("created_at", desc=True)\
.limit(limit)\
.offset(offset)
result = query.execute()
return result.data if result.data else []
except Exception as e:
logger.error(f"Error getting user analysis results: {str(e)}")
raise RuntimeError(f"Failed to get analysis results: {str(e)}")
# Initialize singleton instance
analysis_service = AnalysisService()
|