from time import time from tokenize import group import pandas as pd from sqlalchemy import create_engine, text import os import random from dotenv import load_dotenv import sys import json from llm_client import ( analyze_social_comments_batch, analyze_poll_comments_batch, analyze_comments_batch_generic, get_social_prompt_version ) load_dotenv() BATCH_SIZE = 15 DB_URI = os.getenv("MYSQL_URI") engine = create_engine(DB_URI, pool_pre_ping=True) debate_context_cache = {} # --------------------------- # FETCH UNPROCESSED COMMENTS # --------------------------- def fetch_unprocessed_comments(limit=100): query = """ SELECT scr.* FROM social_comments_raw scr JOIN article_user_comment auc ON scr.comment_id = auc.external_comment_id WHERE auc.llm_sentiment IS NULL AND scr.article_id IS NOT NULL ORDER BY scr.comment_created_time ASC LIMIT :limit """ with engine.connect() as conn: df = pd.read_sql(text(query), conn, params={"limit": limit}) return df # --------------------------- # FETCH DEBATE CONTEXT # --------------------------- def fetch_debate_context(article_id): with engine.connect() as conn: article = conn.execute(text(""" SELECT has_debate_poll, debate_ques, debate_option_yes, debate_option_no, poll_ques FROM article WHERE id = :article_id """), {"article_id": article_id}).fetchone() if not article: return None kind = article[0] # 'debate', 'poll', or None if kind == "debate" and article[1] and article[2] and article[3]: return { "type": "debate", "question": article[1], "options": [article[2], article[3]], } if kind == "poll" and article[4]: with engine.connect() as conn: rows = conn.execute(text(""" SELECT title FROM article_poll_option WHERE article_id = :article_id ORDER BY id """), {"article_id": article_id}).fetchall() options = [r[0] for r in rows if r[0]] if options: return { "type": "poll", "question": article[4], "options": options, } return None # --------------------------- # MOCK CLASSIFIER (TEMP) # --------------------------- def classify_comment(comment, question, option_a, option_b): # TEMP: random assignment (replace with LLM later) side = random.choice(["A", "B", "Neutral"]) confidence = round(random.uniform(0.6, 0.95), 2) return side, confidence # --------------------------- # EMOTION ANALYSIS (TEMP) # --------------------------- def mock_llm_emotion(comment_text, side, sentiment=None): """ Mock emotion classifier for testing pipeline. Later this will be replaced by real LLM output. """ emotions = [ "anger", "joy", "sadness", "fear", "surprise", "disgust", "neutral", "frustration", "hope", "sarcasm", "confusion" ] # Basic heuristic biasing (optional but useful for realism) text = (comment_text or "").lower() if "?" in text: base = "confusion" elif "!" in text and side == "A": base = "frustration" elif "lol" in text or "haha" in text: base = "sarcasm" elif "love" in text or "great" in text: base = "joy" elif "hate" in text or "terrible" in text: base = "anger" else: base = random.choice(emotions) # Small randomness so it doesn't look too fake if random.random() < 0.2: base = random.choice(emotions) return base # --------------------------- # IMPACT SCORE # --------------------------- def compute_impact(row, confidence): like_count = row.get("comment_like_count") or 0 reply_count = row.get("comment_reply_count") or 0 try: confidence = float(confidence) except: confidence = 0.5 like_score = min(like_count / 10, 1) reply_score = min(reply_count / 5, 1) raw = (0.4 * like_score) + (0.3 * reply_score) + (0.3 * confidence) return round(raw * 2 - 1, 3) # --------------------------- # PROCESS LOOP # --------------------------- def safe_float(val, default=0.5): try: return float(val) except: return default def process_comments(): total_processed = 0 failed_batches = 0 # --------------------------- # FETCH DATA # --------------------------- df = fetch_unprocessed_comments(limit=100) if df.empty: print("No new comments to process") return results = [] # --------------------------- # GROUP BY ARTICLE # --------------------------- # Group by article_id (CRITICAL FIX) df = df.dropna(subset=["article_id"]) grouped = df.groupby("article_id") for article_id, group in grouped: # context = fetch_debate_context(article_id) # old version - moved caching inside loop if article_id not in debate_context_cache: debate_context_cache[article_id] = fetch_debate_context(article_id) context = debate_context_cache[article_id] # if not context: # continue ctx_type = context.get("type") if context else None has_debate = ctx_type == "debate" has_poll = ctx_type == "poll" if has_debate: question = context["question"] option_a = context["options"][0] option_b = context["options"][1] elif has_poll: question = context["question"] poll_options = context["options"] else: question = None option_a = None option_b = None group = group.reset_index(drop=True) # --------------------------- # PROCESS BATCHES # --------------------------- for i in range(0, len(group), BATCH_SIZE): batch_df = group.iloc[i:i+BATCH_SIZE] comments = batch_df["comment_message"].tolist() if has_debate: llm_outputs, used_model = analyze_social_comments_batch( comments, question, option_a, option_b ) elif has_poll: llm_outputs, used_model = analyze_poll_comments_batch( comments, question, poll_options ) else: llm_outputs, used_model = analyze_comments_batch_generic( comments ) if not llm_outputs: print(f"[LLM FAIL] article_id={article_id} batch_size={len(comments)}") failed_batches += 1 continue # map results for item in llm_outputs: try: # idx = item["index"] - 1 # old version - moved index parsing inside try block to handle malformed outputs idx = item.get("index") if idx is None: continue try: idx = int(item.get("index", -1)) - 1 except (ValueError, TypeError): continue if idx < 0 or idx >= len(batch_df): continue row = batch_df.iloc[idx] # --------------------------- # PERSIST RESULTS # --------------------------- # confidence = float(item.get("confidence", 0.5)) has_side = has_debate or has_poll confidence = safe_float(item.get("confidence")) if has_side else None side = item.get("side") if has_side else None sentiment = item.get("sentiment", "Neutral") emotion = item.get("emotion", "neutral") toxicity = safe_float(item.get("toxicity", 0.0)) language_code = item.get("language_code", "en") impact = compute_impact(row, confidence if confidence is not None else 0.5) results.append({ "comment_id": row["comment_id"], "llm_assigned_side": side, "llm_confidence": confidence, "llm_sentiment": sentiment, "llm_emotion": emotion, "llm_impact_score": impact, "llm_toxicity_score": toxicity, "llm_language_code": language_code, "llm_raw_output": json.dumps(item), "model_version": used_model, "llm_prompt_version": get_social_prompt_version(), }) except Exception as e: print("Mapping error:", e) continue total_processed += len(llm_outputs) insert_processed(results) print("\n====================") print("PROCESSING SUMMARY") print("====================") print("Fetched:", len(df)) print("Unique articles:", df["article_id"].nunique()) print(f"Total LLM outputs processed: {total_processed}") print(f"Failed batches: {failed_batches}") print(f"Total DB rows inserted: {len(results)}") return len({r["comment_id"] for r in results}) # --------------------------- # INSERT RESULTS # --------------------------- # def insert_processed(rows): # with engine.begin() as conn: # for row in rows: # query = text(""" # INSERT INTO social_comments_processed ( # raw_comment_id, # article_id, # llm_assigned_side, # llm_confidence, # llm_sentiment, # llm_impact_score, # llm_emotion, # model_version, # llm_raw_output # ) # VALUES ( # :raw_comment_id, # :article_id, # :llm_assigned_side, # :llm_confidence, # :llm_sentiment, # :llm_impact_score, # :llm_emotion, # :model_version, # :llm_raw_output # ) # ON DUPLICATE KEY UPDATE # llm_assigned_side = VALUES(llm_assigned_side), # llm_confidence = VALUES(llm_confidence), # llm_sentiment = VALUES(llm_sentiment), # llm_impact_score = VALUES(llm_impact_score), # llm_emotion = VALUES(llm_emotion), # model_version = VALUES(model_version), # llm_raw_output = VALUES(llm_raw_output) # """) # conn.execute(query, rows) def insert_processed(rows): if not rows: return query = text(""" UPDATE article_user_comment SET llm_assigned_side = :llm_assigned_side, llm_confidence = :llm_confidence, llm_sentiment = :llm_sentiment, llm_emotion = :llm_emotion, llm_impact_score = :llm_impact_score, llm_toxicity_score = :llm_toxicity_score, llm_language_code = :llm_language_code, llm_prompt_version = :llm_prompt_version, llm_raw_output = :llm_raw_output, model_version = :model_version WHERE external_comment_id = :comment_id """) with engine.begin() as conn: conn.execute(query, rows) return len(rows) # --------------------------- # MAIN # --------------------------- if __name__ == "__main__": while True: processed = process_comments() print(f"[DEBUG] processed batch size: {processed}") if not processed: print("\n✅ ALL COMMENTS PROCESSED") break time.sleep(1)