import os import sys # Ensure 'src' is in sys.path so sibling imports work current_dir = os.path.dirname(os.path.abspath(__file__)) if current_dir not in sys.path: sys.path.append(current_dir) import asyncio import subprocess from pathlib import Path import logging import csv import io import datetime import json import hashlib import re from fastapi import FastAPI, Request, Form, UploadFile, File, Body, HTTPException from fastapi.responses import HTMLResponse, StreamingResponse, PlainTextResponse, Response, FileResponse, JSONResponse from fastapi.templating import Jinja2Templates from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware import yt_dlp import inference_logic import factuality_logic import transcription import user_analysis_logic import agent_logic import common_utils from toon_parser import parse_veracity_toon from labeling_logic import PROMPT_VARIANTS, LABELING_PROMPT_TEMPLATE, LABELING_PROMPT_TEMPLATE_NO_COT, FCOT_MACRO_PROMPT import benchmarking logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) LITE_MODE = os.getenv("LITE_MODE", "true").lower() == "true" app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --- CRITICAL: Mount A2A Agent Application FIRST --- agent_mount_status = "pending" try: logger.info("Attempting to build A2A Agent App...") a2a_agent_app = agent_logic.create_a2a_app() if a2a_agent_app: app.mount("/a2a", a2a_agent_app) agent_mount_status = "success" logger.info("✅ A2A Agent App successfully mounted at /a2a") else: logger.warning("⚠️ Agent factory returned None. Mounting internal fallback.") from fastapi import FastAPI as InnerFastAPI fallback = InnerFastAPI() @fallback.post("/") @fallback.post("/jsonrpc") async def fallback_endpoint(request: Request): return {"jsonrpc": "2.0", "result": {"text": "Fallback Agent (Factory returned None)", "data": {"status": "fallback"}}} app.mount("/a2a", fallback) agent_mount_status = "fallback_none" except Exception as e: logger.critical(f"❌ Failed to mount A2A Agent: {e}", exc_info=True) from fastapi import FastAPI as InnerFastAPI fallback = InnerFastAPI() @fallback.post("/") @fallback.post("/jsonrpc") async def fallback_endpoint(request: Request): return {"jsonrpc": "2.0", "result": {"text": f"Emergency Agent (Mount Error: {str(e)})", "data": {"status": "error"}}} app.mount("/a2a", fallback) agent_mount_status = f"error_{str(e)}" # --- Static Files & Frontend --- STATIC_DIR = "static" if os.path.isdir("/app/static"): STATIC_DIR = "/app/static" elif os.path.isdir("/usr/share/vchat/static"): STATIC_DIR = "/usr/share/vchat/static" elif os.path.isdir("frontend/dist"): STATIC_DIR = "frontend/dist" elif not os.path.isdir(STATIC_DIR): os.makedirs(STATIC_DIR, exist_ok=True) app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") # Explicitly mount assets for Vite support assets_path = os.path.join(STATIC_DIR, "assets") if os.path.exists(assets_path): app.mount("/assets", StaticFiles(directory=assets_path), name="assets") # Safely Ensure all data directories exist data_dirs =[ "data", "data/videos", "data/labels", "data/prompts", "data/responses", "metadata", "data/profiles", "data/comments", "data/mnl_labeled", "data/models/sandbox_autogluon" ] for d in data_dirs: try: os.makedirs(d, exist_ok=True) except PermissionError as e: logger.warning(f"Permission denied creating directory {d}. Skipping. Error: {e}") except Exception as e: logger.warning(f"Failed to create directory {d}: {e}") if os.path.isdir("data/videos"): app.mount("/videos", StaticFiles(directory="data/videos"), name="videos") templates = Jinja2Templates(directory=STATIC_DIR) try: csv.field_size_limit(sys.maxsize) except OverflowError: csv.field_size_limit(2147483647) STOP_QUEUE_SIGNAL = False # --- CONSTANTS --- QUEUE_COLUMNS =["link", "ingest_timestamp", "status", "task_type"] GROUND_TRUTH_FIELDS =[ "id", "link", "timestamp", "caption", "visual_integrity_score", "audio_integrity_score", "source_credibility_score", "logical_consistency_score", "emotional_manipulation_score", "video_audio_score", "video_caption_score", "audio_caption_score", "final_veracity_score", "final_reasoning", "stats_likes", "stats_shares", "stats_comments", "stats_platform", "tags", "classification", "source" ] DATASET_COLUMNS =[ "id", "link", "timestamp", "caption", "final_veracity_score", "visual_score", "audio_score", "source_score", "logic_score", "emotion_score", "align_video_audio", "align_video_caption", "align_audio_caption", "classification", "reasoning", "tags", "raw_toon", "config_type", "config_model", "config_prompt", "config_reasoning", "config_params" ] def ensure_csv_schema(file_path: Path, fieldnames: list): if not file_path.exists(): return try: rows =[] with open(file_path, 'r', encoding='utf-8', errors='replace') as f: start_pos = f.tell() line = f.readline() if not line: return existing_header =[h.strip() for h in line.split(',')] missing =[col for col in fieldnames if col not in existing_header] if not missing: return f.seek(start_pos) dict_reader = csv.DictReader(f) rows = list(dict_reader) with open(file_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore') writer.writeheader() for row in rows: writer.writerow(row) except Exception as e: logger.error(f"Schema migration error: {e}") def get_processed_indices(): processed_ids = set() processed_links = set() for filename in["data/dataset.csv", "data/manual_dataset.csv"]: path = Path(filename) for row in common_utils.robust_read_csv(path): if row.get('id'): processed_ids.add(row.get('id')) if row.get('link'): processed_links.add(common_utils.normalize_link(row.get('link'))) return processed_ids, processed_links def check_if_processed(link: str, processed_ids=None, processed_links=None) -> bool: target_id = common_utils.extract_tweet_id(link) link_clean = common_utils.normalize_link(link) if processed_ids is None or processed_links is None: p_ids, p_links = get_processed_indices() else: p_ids, p_links = processed_ids, processed_links return (target_id and target_id in p_ids) or (link_clean and link_clean in p_links) def update_queue_status(link: str, status: str, task_type: str = None): q_path = Path("data/batch_queue.csv") if not q_path.exists(): return rows =[] updated = False norm_target = common_utils.normalize_link(link) with open(q_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) fieldnames = list(reader.fieldnames) if reader.fieldnames else list(QUEUE_COLUMNS) for f_name in QUEUE_COLUMNS: if f_name not in fieldnames: fieldnames.append(f_name) for row in reader: if "task_type" not in row or not row["task_type"]: row["task_type"] = "Ingest" if common_utils.normalize_link(row.get("link", "")) == norm_target: if task_type is None or row["task_type"] == task_type: row["status"] = status updated = True rows.append(row) if updated: with open(q_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore') writer.writeheader() writer.writerows(rows) def log_queue_error(link: str, error_msg: str, task_type: str = None): p = Path("data/queue_errors.csv") with open(p, 'a', newline='', encoding='utf-8') as f: writer = csv.writer(f) if not p.exists() or p.stat().st_size == 0: writer.writerow(["link", "timestamp", "error"]) writer.writerow([link, datetime.datetime.now().isoformat(), error_msg]) update_queue_status(link, "Error", task_type) @app.on_event("startup") async def startup_event(): ensure_csv_schema(Path("data/dataset.csv"), DATASET_COLUMNS) ensure_csv_schema(Path("data/manual_dataset.csv"), GROUND_TRUTH_FIELDS) ensure_csv_schema(Path("data/batch_queue.csv"), QUEUE_COLUMNS) if not LITE_MODE: try: inference_logic.load_models() except Exception: pass @app.get("/health") async def health_check(): return {"status": "ok", "agent_mount": agent_mount_status} @app.get("/benchmarks/stats") async def get_benchmark_stats(): return benchmarking.calculate_benchmarks() @app.get("/benchmarks/leaderboard") async def get_benchmark_leaderboard(): return benchmarking.generate_leaderboard() @app.get("/config/prompts") async def list_prompts(): return[{"id": k, "name": v['description']} for k, v in PROMPT_VARIANTS.items()] @app.get("/config/tags") async def list_configured_tags(): path = Path("data/tags.json") if path.exists(): with open(path, 'r') as f: return json.load(f) return {} @app.post("/config/tags") async def save_configured_tags(tags: dict = Body(...)): path = Path("data/tags.json") with open(path, 'w', encoding='utf-8') as f: json.dump(tags, f, indent=2) return {"status": "success"} @app.get("/tags/list") async def list_all_tags(): tags_count = {} path = Path("data/dataset.csv") if path.exists(): for row in common_utils.robust_read_csv(path): t_str = row.get("tags", "") if t_str: for t in t_str.split(','): t = t.strip() if t: tags_count[t] = tags_count.get(t, 0) + 1 sorted_tags = sorted(tags_count.items(), key=lambda x: x[1], reverse=True) return[{"name": k, "count": v} for k, v in sorted_tags] @app.post("/extension/ingest") async def extension_ingest_link(request: Request): try: data = await request.json() link = data.get("link") comments = data.get("comments",[]) if not link: raise HTTPException(status_code=400, detail="Link required") q_path = Path("data/batch_queue.csv") existing = set() if q_path.exists(): for r in common_utils.robust_read_csv(q_path): existing.add(common_utils.normalize_link(r.get('link'))) normalized = common_utils.normalize_link(link) if normalized not in existing: with open(q_path, 'a', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=QUEUE_COLUMNS, extrasaction='ignore') if not q_path.exists() or q_path.stat().st_size == 0: writer.writeheader() writer.writerow({"link": link.strip(), "ingest_timestamp": datetime.datetime.now().isoformat(), "status": "Pending", "task_type": "Ingest"}) if comments: tid = common_utils.extract_tweet_id(link) or hashlib.md5(link.encode()).hexdigest()[:10] context_path = Path(f"data/comments/{tid}_ingest.json") with open(context_path, 'w', encoding='utf-8') as f: json.dump({ "link": link, "timestamp": datetime.datetime.now().isoformat(), "comments": comments }, f, indent=2) logger.info(f"Saved {len(comments)} comments for ingestion context: {tid}") return {"status": "success", "link": link, "comments_saved": len(comments)} except Exception as e: logger.error(f"Ingest Error: {e}") return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.post("/manual/promote") async def promote_to_ground_truth(request: Request): try: data = await request.json() target_ids = data.get("ids",[]) if not target_ids and data.get("id"): target_ids =[data.get("id")] if not target_ids: return JSONResponse({"status": "error", "message": "No IDs provided"}, status_code=400) ai_path = Path("data/dataset.csv") ai_rows = {} if ai_path.exists(): for row in common_utils.robust_read_csv(ai_path): if row.get('id'): ai_rows[str(row['id'])] = row manual_path = Path("data/manual_dataset.csv") manual_exists = manual_path.exists() existing_ids = set() if manual_exists: for row in common_utils.robust_read_csv(manual_path): if row.get('id'): existing_ids.add(str(row['id'])) new_rows =[] promoted_count = 0 for tid in target_ids: tid_str = str(tid) if tid_str in existing_ids: continue found_row = ai_rows.get(tid_str) if found_row: mapped_row = { "id": found_row.get("id"), "link": found_row.get("link"), "timestamp": datetime.datetime.now().isoformat(), "caption": found_row.get("caption"), "visual_integrity_score": found_row.get("visual_score", 0), "audio_integrity_score": found_row.get("audio_score", 0), "source_credibility_score": 5, "logical_consistency_score": found_row.get("logic_score", 0), "emotional_manipulation_score": 5, "video_audio_score": 5, "video_caption_score": found_row.get("align_video_caption", 0), "audio_caption_score": 5, "final_veracity_score": found_row.get("final_veracity_score", 0), "final_reasoning": found_row.get("reasoning", ""), "stats_likes": 0, "stats_shares": 0, "stats_comments": 0, "stats_platform": "twitter", "tags": found_row.get("tags", ""), "classification": found_row.get("classification", "None"), "source": "manual_promoted" } new_rows.append(mapped_row) promoted_count += 1 existing_ids.add(tid_str) if not new_rows: return {"status": "success", "promoted_count": 0} mode = 'a' if manual_exists else 'w' with open(manual_path, mode, newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=GROUND_TRUTH_FIELDS, extrasaction='ignore') if not manual_exists or manual_path.stat().st_size == 0: writer.writeheader() for r in new_rows: writer.writerow(r) return {"status": "success", "promoted_count": promoted_count} except Exception as e: return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.post("/manual/delete") async def delete_ground_truth(request: Request): try: data = await request.json() target_ids = data.get("ids",[]) if not target_ids and data.get("id"): target_ids =[data.get("id")] if not target_ids: raise HTTPException(status_code=400) target_ids =[str(t) for t in target_ids] manual_path = Path("data/manual_dataset.csv") if not manual_path.exists(): return {"status": "error", "message": "File not found"} rows =[] deleted_count = 0 with open(manual_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: if str(row.get('id')) in target_ids: deleted_count += 1 else: rows.append(row) with open(manual_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=GROUND_TRUTH_FIELDS) writer.writeheader() writer.writerows(rows) return {"status": "success", "deleted_count": deleted_count} except Exception as e: return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.post("/manual/verify_queue") async def verify_queue_items(request: Request): try: data = await request.json() target_ids = data.get("ids",[]) resample_count = max(1, min(data.get("resample_count", 1), 100)) if not target_ids: return JSONResponse({"status": "error", "message": "No IDs provided"}, status_code=400) manual_path = Path("data/manual_dataset.csv") links_to_queue =[] if manual_path.exists(): for row in common_utils.robust_read_csv(manual_path): if str(row.get('id')) in target_ids: links_to_queue.append(row.get('link')) if not links_to_queue: return {"status": "error", "message": "No matching links found in Ground Truth."} q_path = Path("data/batch_queue.csv") added_count = 0 with open(q_path, 'a', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=QUEUE_COLUMNS, extrasaction='ignore') if not q_path.exists() or q_path.stat().st_size == 0: writer.writeheader() for link in links_to_queue: for _ in range(resample_count): writer.writerow({ "link": link.strip(), "ingest_timestamp": datetime.datetime.now().isoformat(), "status": "Pending", "task_type": "Verify" }) added_count += 1 return {"status": "success", "queued_count": added_count, "message": f"Added {added_count} items to queue for verification pipeline."} except Exception as e: return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.get("/profiles/list") async def list_profiles(): profiles_dir = Path("data/profiles") profiles =[] if not profiles_dir.exists(): return profiles try: for d in profiles_dir.iterdir(): if d.is_dir(): hist = d / "history.csv" count = 0 if hist.exists(): with open(hist, 'r', encoding='utf-8', errors='ignore') as f: count = sum(1 for _ in f) - 1 profiles.append({"username": d.name, "posts_count": max(0, count)}) except Exception: pass return sorted(profiles, key=lambda x: x['username']) @app.get("/profiles/{username}/posts") async def get_profile_posts(username: str): csv_path = Path(f"data/profiles/{username}/history.csv") posts =[] if not csv_path.exists(): return posts p_ids, p_links = get_processed_indices() try: for row in common_utils.robust_read_csv(csv_path): link = row.get('link', '') is_labeled = False t_id = common_utils.extract_tweet_id(link) if t_id and t_id in p_ids: is_labeled = True elif common_utils.normalize_link(link) in p_links: is_labeled = True row['is_labeled'] = is_labeled posts.append(row) except Exception: pass return posts @app.post("/extension/ingest_user_history") async def ingest_user_history(request: Request): try: data = await request.json() username = data.get("username") posts = data.get("posts",[]) if not username or not posts: raise HTTPException(status_code=400) profile_dir = Path(f"data/profiles/{username}") profile_dir.mkdir(parents=True, exist_ok=True) csv_path = profile_dir / "history.csv" file_exists = csv_path.exists() existing = set() if file_exists: for row in common_utils.robust_read_csv(csv_path): existing.add(row.get('link')) with open(csv_path, 'a', newline='', encoding='utf-8') as f: fieldnames =["link", "timestamp", "text", "is_reply", "metric_replies", "metric_reposts", "metric_likes", "metric_views", "ingested_at"] writer = csv.DictWriter(f, fieldnames=fieldnames) if not file_exists: writer.writeheader() ts = datetime.datetime.now().isoformat() count = 0 for p in posts: if p['link'] not in existing: m = p.get('metrics', {}) writer.writerow({ "link": p.get('link'), "timestamp": p.get('timestamp'), "text": p.get('text', '').replace('\n', ' '), "is_reply": p.get('is_reply', False), "metric_replies": m.get('replies', 0), "metric_reposts": m.get('reposts', 0), "metric_likes": m.get('likes', 0), "metric_views": m.get('views', 0), "ingested_at": ts }) count += 1 return {"status": "success", "new_posts": count} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/extension/save_comments") async def extension_save_comments(request: Request): try: data = await request.json() link = data.get("link") comments = data.get("comments",[]) if not link: raise HTTPException(status_code=400) tweet_id = common_utils.extract_tweet_id(link) or hashlib.md5(link.encode()).hexdigest()[:10] csv_path = Path(f"data/comments/{tweet_id}.csv") with open(csv_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=["author", "text", "link", "timestamp"]) writer.writeheader() ts = datetime.datetime.now().isoformat() for c in comments: writer.writerow({ "author": c.get("author", "Unknown"), "text": c.get("text", "").replace("\n", " "), "link": c.get("link", ""), "timestamp": ts }) return {"status": "success", "count": len(comments)} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/extension/save_manual") @app.post("/manual/save") async def save_manual_label(request: Request): try: data = await request.json() link = data.get("link") if not link: return JSONResponse({"status": "error", "message": "Link required"}, status_code=400) tweet_id = common_utils.extract_tweet_id(link) or hashlib.md5(link.encode()).hexdigest()[:10] labels = data.get("labels", data) row = { "id": tweet_id, "link": link, "timestamp": datetime.datetime.now().isoformat(), "caption": data.get("caption", ""), "visual_integrity_score": labels.get("visual_integrity_score", 0), "audio_integrity_score": labels.get("audio_integrity_score", 0), "source_credibility_score": labels.get("source_credibility_score", 0), "logical_consistency_score": labels.get("logical_consistency_score", 0), "emotional_manipulation_score": labels.get("emotional_manipulation_score", 5), "video_audio_score": labels.get("video_audio_score", 0), "video_caption_score": labels.get("video_caption_score", 0), "audio_caption_score": labels.get("audio_caption_score", 0), "final_veracity_score": labels.get("final_veracity_score", 0), "final_reasoning": labels.get("reasoning", labels.get("final_reasoning", "")), "stats_likes": 0, "stats_shares": 0, "stats_comments": 0, "stats_platform": "twitter", "tags": data.get("tags", labels.get("tags", "")), "classification": labels.get("classification", "None"), "source": "Manual" } tag_str = str(row["tags"]) tag_list =[t.strip() for t in tag_str.split(',') if t.strip()] deep_json = { "veracity_vectors": { "visual_integrity_score": str(row["visual_integrity_score"]), "audio_integrity_score": str(row["audio_integrity_score"]), "source_credibility_score": str(row["source_credibility_score"]), "logical_consistency_score": str(row["logical_consistency_score"]), "emotional_manipulation_score": str(row["emotional_manipulation_score"]) }, "modalities": { "video_audio_score": str(row["video_audio_score"]), "video_caption_score": str(row["video_caption_score"]), "audio_caption_score": str(row["audio_caption_score"]) }, "video_context_summary": row["caption"], "tags": tag_list, "factuality_factors": { "claim_accuracy": "Manual", "evidence_gap": "Manual Verification", "grounding_check": "Manual Verification" }, "disinformation_analysis": { "classification": row["classification"], "intent": "Manual Labeling", "threat_vector": "Manual Labeling" }, "final_assessment": { "veracity_score_total": str(row["final_veracity_score"]), "reasoning": row["final_reasoning"] }, "raw_parsed_structure": { "summary": {"text": row["caption"]}, "tags": {"keywords": row["tags"]}, "final": {"score": str(row["final_veracity_score"]), "reasoning": row["final_reasoning"]} }, "meta_info": { "id": tweet_id, "timestamp": row["timestamp"], "link": link, "model_selection": "Manual" } } json_path_direct = Path(f"data/labels/{tweet_id}.json") with open(json_path_direct, 'w', encoding='utf-8') as jf: json.dump(deep_json, jf, indent=2, ensure_ascii=False) with open(Path(f"data/mnl_labeled/{tweet_id}.json"), 'w', encoding='utf-8') as jf: json.dump(row, jf, indent=2, ensure_ascii=False) manual_path = Path("data/manual_dataset.csv") exists = manual_path.exists() ensure_csv_schema(manual_path, GROUND_TRUTH_FIELDS) rows =[] found = False if exists: for r in common_utils.robust_read_csv(manual_path): if str(r.get('id')) == str(tweet_id): clean_row = {k: row.get(k, "") for k in GROUND_TRUTH_FIELDS} rows.append(clean_row) found = True else: rows.append(r) if not found: clean_row = {k: row.get(k, "") for k in GROUND_TRUTH_FIELDS} rows.append(clean_row) with open(manual_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=GROUND_TRUTH_FIELDS, extrasaction='ignore') writer.writeheader() writer.writerows(rows) # Add to User Profiles Catalog author = common_utils.extract_twitter_username(link) if author: prof_dir = Path(f"data/profiles/{author}") prof_dir.mkdir(parents=True, exist_ok=True) hist_path = prof_dir / "history.csv" hist_exists = hist_path.exists() existing_links = set() if hist_exists: for r in common_utils.robust_read_csv(hist_path): existing_links.add(r.get('link')) if link not in existing_links: with open(hist_path, 'a', newline='', encoding='utf-8') as hf: fieldnames =["link", "timestamp", "text", "is_reply", "metric_replies", "metric_reposts", "metric_likes", "metric_views", "ingested_at"] hwriter = csv.DictWriter(hf, fieldnames=fieldnames, extrasaction='ignore') if not hist_exists: hwriter.writeheader() hwriter.writerow({ "link": link, "timestamp": row["timestamp"], "text": row["caption"], "ingested_at": row["timestamp"] }) update_queue_status(link, "Processed") return {"status": "success", "id": tweet_id} except Exception as e: logger.error(f"Save Manual Error: {e}") return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.get("/dataset/list") async def get_dataset_list(): dataset =[] m_path = Path("data/manual_dataset.csv") manual_ids = set() if m_path.exists(): for row in common_utils.robust_read_csv(m_path): row['source'] = 'Manual' if row.get('id'): manual_ids.add(str(row['id'])) dataset.append(row) path = Path("data/dataset.csv") if path.exists(): for row in common_utils.robust_read_csv(path): tid = str(row.get('id', '')) if tid not in manual_ids: row['source'] = 'AI' dataset.append(row) return sorted(dataset, key=lambda x: x.get('timestamp', ''), reverse=True) @app.get("/analytics/account_integrity") async def get_account_integrity(): id_map = {} prof_dir = Path("data/profiles") prof_dir.mkdir(parents=True, exist_ok=True) existing_links_per_user = {} if prof_dir.exists(): for d in prof_dir.iterdir(): if d.is_dir(): hist_file = d / "history.csv" existing_links_per_user[d.name] = set() if hist_file.exists(): for row in common_utils.robust_read_csv(hist_file): link = row.get('link', '') tid = common_utils.extract_tweet_id(link) if tid: id_map[tid] = d.name existing_links_per_user[d.name].add(link) scores_map = {} for fname in["data/dataset.csv", "data/manual_dataset.csv"]: path = Path(fname) if not path.exists(): continue for row in common_utils.robust_read_csv(path): tid = row.get('id') link = row.get('link', '') sc = row.get('final_veracity_score', '0') ts = row.get('timestamp', '') caption = row.get('caption', '') try: val = float(re.sub(r'[^\d.]', '', str(sc))) except: val = -1 # Require scores to be between 0 and 100 if 0 <= val <= 100: auth = common_utils.extract_twitter_username(link) or id_map.get(tid, "Unknown") if auth and auth != "Unknown": if auth not in scores_map: scores_map[auth] = [] scores_map[auth].append({'val': val, 'ts': ts}) # Auto-add missing accounts/links to the Profile catalog if auth not in existing_links_per_user: existing_links_per_user[auth] = set() Path(f"data/profiles/{auth}").mkdir(parents=True, exist_ok=True) if link not in existing_links_per_user[auth]: existing_links_per_user[auth].add(link) hist_path = Path(f"data/profiles/{auth}/history.csv") hist_exists = hist_path.exists() with open(hist_path, 'a', newline='', encoding='utf-8') as hf: fieldnames =["link", "timestamp", "text", "is_reply", "metric_replies", "metric_reposts", "metric_likes", "metric_views", "ingested_at"] hwriter = csv.DictWriter(hf, fieldnames=fieldnames, extrasaction='ignore') if not hist_exists: hwriter.writeheader() hwriter.writerow({ "link": link, "timestamp": ts, "text": caption, "ingested_at": ts }) results =[] for k, v in scores_map.items(): v_sorted = sorted(v, key=lambda x: x['ts'], reverse=True) decay_factor = 0.9 total_weight = 0 weighted_sum = 0 for i, item in enumerate(v_sorted): weight = decay_factor ** i weighted_sum += item['val'] * weight total_weight += weight avg_veracity = round(weighted_sum / total_weight, 1) if total_weight > 0 else 0 results.append({"username": k, "avg_veracity": avg_veracity, "posts_labeled": len(v)}) return sorted(results, key=lambda x: x['avg_veracity'], reverse=True) @app.post("/queue/add") async def add_queue_item(link: str = Body(..., embed=True)): q_path = Path("data/batch_queue.csv") existing = set() if q_path.exists(): for r in common_utils.robust_read_csv(q_path): existing.add(common_utils.normalize_link(r.get('link'))) normalized = common_utils.normalize_link(link) if not normalized: raise HTTPException(status_code=400, detail="Invalid link") if normalized in existing: return {"status": "ignored", "message": "Link already in queue"} with open(q_path, 'a', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=QUEUE_COLUMNS, extrasaction='ignore') if not q_path.exists() or q_path.stat().st_size == 0: writer.writeheader() writer.writerow({"link": link.strip(), "ingest_timestamp": datetime.datetime.now().isoformat(), "status": "Pending", "task_type": "Ingest"}) return {"status": "success", "link": link} @app.post("/queue/upload_csv") async def upload_csv(file: UploadFile = File(...)): contents = await file.read() lines = contents.decode('utf-8').splitlines() q_path = Path("data/batch_queue.csv") existing = set() if q_path.exists(): for r in common_utils.robust_read_csv(q_path): existing.add(common_utils.normalize_link(r.get('link'))) added = 0 with open(q_path, 'a', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=QUEUE_COLUMNS, extrasaction='ignore') if not q_path.exists() or q_path.stat().st_size == 0: writer.writeheader() for line in lines: if 'http' in line: raw = line.split(',')[0].strip() if common_utils.normalize_link(raw) not in existing: writer.writerow({"link": raw, "ingest_timestamp": datetime.datetime.now().isoformat(), "status": "Pending", "task_type": "Ingest"}) added += 1 return {"status": "success", "added_count": added} @app.post("/queue/stop") async def stop_processing(): global STOP_QUEUE_SIGNAL STOP_QUEUE_SIGNAL = True return {"status": "success", "message": "Stopping queue processing..."} @app.post("/queue/clear_processed") async def clear_processed_queue(): q_path = Path("data/batch_queue.csv") if not q_path.exists(): return {"status": "success", "removed_count": 0} p_ids, p_links = get_processed_indices() kept_rows =[] removed_count = 0 for row in common_utils.robust_read_csv(q_path): link = row.get("link") status = row.get("status", "Pending") task_type = row.get("task_type", "Ingest") is_done = False if status == "Processed": is_done = True elif task_type != "Verify" and check_if_processed(link, p_ids, p_links): is_done = True if is_done: removed_count += 1 else: kept_rows.append(row) with open(q_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=QUEUE_COLUMNS, extrasaction='ignore') writer.writeheader() writer.writerows(kept_rows) return {"status": "success", "removed_count": removed_count} @app.post("/queue/delete") async def delete_queue_items(request: Request): try: data = await request.json() target_links = set(common_utils.normalize_link(l) for l in data.get("links",[])) q_path = Path("data/batch_queue.csv") if not q_path.exists(): return {"status": "success", "count": 0} kept_rows =[] deleted_count = 0 for row in common_utils.robust_read_csv(q_path): if common_utils.normalize_link(row.get('link')) in target_links: deleted_count += 1 else: kept_rows.append(row) with open(q_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=QUEUE_COLUMNS, extrasaction='ignore') writer.writeheader() writer.writerows(kept_rows) return {"status": "success", "count": deleted_count} except Exception as e: return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.post("/queue/requeue") async def requeue_items(request: Request): try: data = await request.json() target_links = set(common_utils.normalize_link(l) for l in data.get("links",[])) q_path = Path("data/batch_queue.csv") if not q_path.exists(): return {"status": "success", "count": 0} rows =[] requeued_count = 0 for row in common_utils.robust_read_csv(q_path): if common_utils.normalize_link(row.get('link')) in target_links: row['status'] = 'Pending' requeued_count += 1 rows.append(row) with open(q_path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=QUEUE_COLUMNS, extrasaction='ignore') writer.writeheader() writer.writerows(rows) return {"status": "success", "count": requeued_count} except Exception as e: return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.post("/dataset/delete") async def delete_dataset_items(request: Request): try: data = await request.json() target_ids = data.get("ids",[]) if not target_ids: raise HTTPException(status_code=400) target_ids = set(str(t) for t in target_ids) path = Path("data/dataset.csv") if not path.exists(): return {"status": "success", "count": 0} rows =[] deleted_count = 0 for row in common_utils.robust_read_csv(path): if str(row.get('id')) in target_ids: deleted_count += 1 else: rows.append(row) with open(path, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=DATASET_COLUMNS, extrasaction='ignore') writer.writeheader() writer.writerows(rows) return {"status": "success", "deleted_count": deleted_count} except Exception as e: return JSONResponse({"status": "error", "message": str(e)}, status_code=500) @app.post("/analyze/user_context") async def analyze_user_context(request: Request): try: data = await request.json() rep = await user_analysis_logic.generate_user_profile_report(data.get("username")) return {"status": "success", "report": rep} except Exception as e: return JSONResponse({"error": str(e)}, status_code=500) @app.get("/", response_class=HTMLResponse) async def read_root(request: Request): return templates.TemplateResponse("index.html", {"request": request}) @app.get("/queue/list") async def get_queue_list(): q_path = Path("data/batch_queue.csv") items =[] p_ids, p_links = get_processed_indices() for row in common_utils.robust_read_csv(q_path): if row: l = row.get("link") status = row.get("status", "Pending") task_type = row.get("task_type") or "Ingest" if status == "Pending" and task_type != "Verify" and check_if_processed(l, p_ids, p_links): status = "Processed" # Fetch associated comments to display in the dropdown comments =[] tid = common_utils.extract_tweet_id(l) or hashlib.md5(l.encode()).hexdigest()[:10] c_path = Path(f"data/comments/{tid}_ingest.json") if c_path.exists(): try: with open(c_path, 'r') as f: c_data = json.load(f) comments = c_data.get('comments',[]) except Exception: pass items.append({ "link": l, "timestamp": row.get("ingest_timestamp",""), "status": status, "task_type": task_type, "comments": comments }) return items @app.post("/queue/run") async def run_queue_processing( model_selection: str = Form(...), gemini_api_key: str = Form(""), gemini_model_name: str = Form(""), vertex_project_id: str = Form(""), vertex_location: str = Form(""), vertex_model_name: str = Form(""), vertex_api_key: str = Form(""), nrp_api_key: str = Form(""), nrp_model_name: str = Form(""), nrp_base_url: str = Form("https://ellm.nrp-nautilus.io/v1"), include_comments: bool = Form(False), reasoning_method: str = Form("cot"), prompt_template: str = Form("standard"), custom_query: str = Form(""), max_reprompts: int = Form(1), use_search: bool = Form(False), use_code: bool = Form(False) ): global STOP_QUEUE_SIGNAL STOP_QUEUE_SIGNAL = False gemini_config = {"api_key": gemini_api_key, "model_name": gemini_model_name, "max_retries": max_reprompts, "use_search": use_search, "use_code": use_code} vertex_config = {"project_id": vertex_project_id, "location": vertex_location, "model_name": vertex_model_name, "api_key": vertex_api_key, "max_retries": max_reprompts, "use_search": use_search, "use_code": use_code} nrp_config = {"api_key": nrp_api_key, "model_name": nrp_model_name, "base_url": nrp_base_url, "max_retries": max_reprompts, "use_search": use_search, "use_code": use_code} sel_p = PROMPT_VARIANTS.get(prompt_template, PROMPT_VARIANTS['standard']) system_persona_txt = sel_p['instruction'] if custom_query.strip(): system_persona_txt += f"\n\nSPECIAL INSTRUCTION FOR THIS BATCH: {custom_query}" if model_selection == 'vertex': active_config = vertex_config active_model_name = vertex_model_name elif model_selection == 'nrp': active_config = nrp_config active_model_name = nrp_model_name else: active_config = gemini_config active_model_name = gemini_model_name config_params_dict = { "reprompts": max_reprompts, "include_comments": include_comments, "agent_active": False, "use_search": use_search, "use_code": use_code } config_params_str = json.dumps(config_params_dict) async def queue_stream(): q_path = Path("data/batch_queue.csv") items =[r for r in common_utils.robust_read_csv(q_path) if r.get("link") and r.get("status", "Pending") == "Pending"] p_ids, p_links = get_processed_indices() yield f"data:[SYSTEM] Persona: {sel_p['description']}\n\n" for item in items: link = item.get("link") task_type = item.get("task_type") or "Ingest" if STOP_QUEUE_SIGNAL: yield f"data:[SYSTEM] Stopping by user request.\n\n" break if task_type != "Verify" and check_if_processed(link, p_ids, p_links): update_queue_status(link, "Processed", task_type) continue gt_data = None if task_type == "Verify": manual_path = Path("data/manual_dataset.csv") if manual_path.exists(): for row in common_utils.robust_read_csv(manual_path): if common_utils.normalize_link(row.get('link', '')) == common_utils.normalize_link(link): gt_data = row break yield f"data:[START] {link} (Type: {task_type})\n\n" tid = common_utils.extract_tweet_id(link) or hashlib.md5(link.encode()).hexdigest()[:10] assets = await common_utils.prepare_video_assets(link, tid) if not assets or (not assets.get('video') and not assets.get('caption')): log_queue_error(link, "Download/Fetch Error", task_type) yield f"data: - Download Error.\n\n" continue trans = common_utils.parse_vtt(assets['transcript']) if assets.get('transcript') else "No transcript (Audio/Video missing)." video_file = assets.get('video') if not video_file: yield f"data: - No video found. Text-only analysis.\n\n" video_file = None else: yield f"data: - Video found. Inferencing...\n\n" comments_path = Path(f"data/comments/{tid}_ingest.json") current_system_persona = system_persona_txt if comments_path.exists(): try: with open(comments_path, 'r') as f: c_data = json.load(f) comments = c_data.get('comments',[]) if comments: yield f"data: - Found {len(comments)} comments. Generating Community Context...\n\n" community_summary = await inference_logic.generate_community_summary(comments, model_selection, active_config) current_system_persona += f"\n\n### COMMUNITY NOTES / CONTEXT (from Comments):\n{community_summary}\n\nUse this community context to cross-reference claims but remain objective." yield f"data: - Context Generated.\n\n" except Exception as e: logger.error(f"Error processing comments for context: {e}") res_data = None if model_selection == 'gemini': async for chunk in inference_logic.run_gemini_labeling_pipeline(video_file, assets['caption'], trans, gemini_config, include_comments, reasoning_method, current_system_persona, request_id=tid): if isinstance(chunk, str): yield f"data: - {chunk}\n\n" else: res_data = chunk elif model_selection == 'vertex': async for chunk in inference_logic.run_vertex_labeling_pipeline(video_file, assets['caption'], trans, vertex_config, include_comments, reasoning_method, current_system_persona, request_id=tid): if isinstance(chunk, str): yield f"data: - {chunk}\n\n" else: res_data = chunk elif model_selection == 'nrp': async for chunk in inference_logic.run_nrp_labeling_pipeline(video_file, assets['caption'], trans, nrp_config, include_comments, reasoning_method, current_system_persona, request_id=tid): if isinstance(chunk, str): yield f"data: - {chunk}\n\n" else: res_data = chunk if res_data and "parsed_data" in res_data: parsed = res_data["parsed_data"] d_path = Path("data/dataset.csv") ensure_csv_schema(d_path, DATASET_COLUMNS) exists = d_path.exists() ai_score_val = parsed['final_assessment'].get('veracity_score_total', 0) try: ai_score = float(ai_score_val) except: ai_score = 0 if task_type == "Verify" and gt_data is not None: gt_final = float(gt_data.get('final_veracity_score', 0)) delta = abs(ai_score - gt_final) vec_ai = parsed.get('veracity_vectors', {}) mod_ai = parsed.get('modalities', {}) def s_float(v): try: return float(v) except: return 0.0 yield f"data: -[VERIFICATION PIPELINE] Configuration Analysis:\n" yield f"data: Model: {active_model_name} | Provider: {model_selection}\n" yield f"data: Reasoning: {reasoning_method} | Prompt: {prompt_template} | Reprompts: {max_reprompts}\n" yield f"data: -[VERIFICATION SCORES COMPARISON (AI vs Ground Truth)]\n" yield f"data: Visual Integrity : AI {s_float(vec_ai.get('visual_integrity_score'))} | GT {s_float(gt_data.get('visual_integrity_score'))}\n" yield f"data: Audio Integrity : AI {s_float(vec_ai.get('audio_integrity_score'))} | GT {s_float(gt_data.get('audio_integrity_score'))}\n" yield f"data: Source Credibility : AI {s_float(vec_ai.get('source_credibility_score'))} | GT {s_float(gt_data.get('source_credibility_score'))}\n" yield f"data: Logical Consistency: AI {s_float(vec_ai.get('logical_consistency_score'))} | GT {s_float(gt_data.get('logical_consistency_score'))}\n" yield f"data: Emotional Manipul. : AI {s_float(vec_ai.get('emotional_manipulation_score'))} | GT {s_float(gt_data.get('emotional_manipulation_score'))}\n" yield f"data: Video-Audio Align : AI {s_float(mod_ai.get('video_audio_score'))} | GT {s_float(gt_data.get('video_audio_score'))}\n" yield f"data: Video-Caption Align: AI {s_float(mod_ai.get('video_caption_score'))} | GT {s_float(gt_data.get('video_caption_score'))}\n" yield f"data: Audio-Caption Align: AI {s_float(mod_ai.get('audio_caption_score'))} | GT {s_float(gt_data.get('audio_caption_score'))}\n" yield f"data: FINAL VERACITY : AI {ai_score} | GT {gt_final} | Delta: {delta}\n\n" comp_path = Path("data/comparison.csv") comp_exists = comp_path.exists() with open(comp_path, 'a', newline='', encoding='utf-8') as cf: cw = csv.DictWriter(cf, fieldnames=["id", "link", "timestamp", "gt_score", "ai_score", "delta", "model", "prompt", "reasoning_method"]) if not comp_exists: cw.writeheader() cw.writerow({ "id": tid, "link": link, "timestamp": datetime.datetime.now().isoformat(), "gt_score": gt_final, "ai_score": ai_score, "delta": delta, "model": active_model_name, "prompt": prompt_template, "reasoning_method": reasoning_method }) try: with open(d_path, 'a', newline='', encoding='utf-8') as f: row = { "id": tid, "link": link, "timestamp": datetime.datetime.now().isoformat(), "caption": assets['caption'], "final_veracity_score": ai_score, "visual_score": parsed['veracity_vectors'].get('visual_integrity_score', 0), "audio_score": parsed['veracity_vectors'].get('audio_integrity_score', 0), "source_score": parsed['veracity_vectors'].get('source_credibility_score', 0), "logic_score": parsed['veracity_vectors'].get('logical_consistency_score', 0), "emotion_score": parsed['veracity_vectors'].get('emotional_manipulation_score', 0), "align_video_audio": parsed['modalities'].get('video_audio_score', 0), "align_video_caption": parsed['modalities'].get('video_caption_score', 0), "align_audio_caption": parsed['modalities'].get('audio_caption_score', 0), "classification": parsed['disinformation_analysis'].get('classification', 'None'), "reasoning": parsed['final_assessment'].get('reasoning', ''), "tags": ",".join(parsed.get('tags',[])), "raw_toon": res_data.get("raw_toon", ""), "config_type": "GenAI", "config_model": active_model_name, "config_prompt": prompt_template, "config_reasoning": reasoning_method, "config_params": config_params_str } writer = csv.DictWriter(f, fieldnames=DATASET_COLUMNS, extrasaction='ignore') if not exists: writer.writeheader() writer.writerow(row) except Exception as csv_err: logger.error(f"CSV Write Failed: {csv_err}") try: ts = datetime.datetime.now().isoformat() ts_clean = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") flat_parsed = parsed.copy() flat_parsed["raw_toon"] = res_data.get("raw_toon", "") flat_parsed["meta_info"] = { "id": tid, "timestamp": ts, "link": link, "prompt_used": res_data.get("prompt_used", ""), "model_selection": model_selection, "config_type": "GenAI", "config_model": active_model_name, "config_prompt": prompt_template, "config_reasoning": reasoning_method, "config_params": config_params_dict } with open(Path(f"data/labels/{tid}_{ts_clean}.json"), 'w', encoding='utf-8') as f: json.dump(flat_parsed, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Sidecar Error: {e}") # Add to User Profiles Catalog author = common_utils.extract_twitter_username(link) if author: prof_dir = Path(f"data/profiles/{author}") prof_dir.mkdir(parents=True, exist_ok=True) hist_path = prof_dir / "history.csv" hist_exists = hist_path.exists() existing_links = set() if hist_exists: for r in common_utils.robust_read_csv(hist_path): existing_links.add(r.get('link')) if link not in existing_links: with open(hist_path, 'a', newline='', encoding='utf-8') as hf: fieldnames =["link", "timestamp", "text", "is_reply", "metric_replies", "metric_reposts", "metric_likes", "metric_views", "ingested_at"] hwriter = csv.DictWriter(hf, fieldnames=fieldnames, extrasaction='ignore') if not hist_exists: hwriter.writeheader() hwriter.writerow({ "link": link, "timestamp": datetime.datetime.now().isoformat(), "text": assets['caption'], "ingested_at": datetime.datetime.now().isoformat() }) p_ids.add(tid) p_links.add(common_utils.normalize_link(link)) update_queue_status(link, "Processed", task_type) yield f"data:[SUCCESS] Saved.\n\n" else: err_msg = res_data.get('error') if isinstance(res_data, dict) else "Inference failed" log_queue_error(link, err_msg, task_type) yield f"data: [FAIL] {err_msg}.\n\n" await asyncio.sleep(0.5) yield "event: close\ndata: Done\n\n" return StreamingResponse(queue_stream(), media_type="text/event-stream")