| | import os |
| | import sys |
| | 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 |
| | from factuality_logic import parse_vtt |
| | from toon_parser import parse_veracity_toon |
| |
|
| | |
| | try: |
| | csv.field_size_limit(sys.maxsize) |
| | except OverflowError: |
| | csv.field_size_limit(2147483647) |
| |
|
| | try: |
| | import mlcroissant as mlc |
| | CROISSANT_AVAILABLE = True |
| | except ImportError: |
| | try: |
| | import croissant as mlc |
| | CROISSANT_AVAILABLE = True |
| | except ImportError: |
| | mlc = None |
| | CROISSANT_AVAILABLE = False |
| |
|
| | logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") |
| | LITE_MODE = os.getenv("LITE_MODE", "false").lower() == "true" |
| |
|
| | app = FastAPI() |
| |
|
| | app.add_middleware( |
| | CORSMiddleware, |
| | allow_origins=["*"], |
| | allow_credentials=True, |
| | allow_methods=["*"], |
| | allow_headers=["*"], |
| | ) |
| |
|
| | STATIC_DIR = "static" |
| | if 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) |
| | dummy_index = Path(STATIC_DIR) / "index.html" |
| | if not dummy_index.exists(): |
| | dummy_index.write_text("<html><body>vChat Backend Running. Access via Port 8005 (Go Server).</body></html>") |
| |
|
| | app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") |
| | templates = Jinja2Templates(directory=STATIC_DIR) |
| |
|
| | os.makedirs("data/videos", exist_ok=True) |
| | os.makedirs("data", exist_ok=True) |
| | os.makedirs("data/labels", exist_ok=True) |
| | os.makedirs("data/prompts", exist_ok=True) |
| | os.makedirs("data/responses", exist_ok=True) |
| | os.makedirs("metadata", exist_ok=True) |
| |
|
| | STOP_QUEUE_SIGNAL = False |
| |
|
| | |
| | def robust_read_csv(file_path: Path): |
| | """ |
| | Reads a CSV file tolerantly. Yields dictionaries. |
| | Handles 'line contains NUL', formatting errors, etc. by skipping bad rows. |
| | """ |
| | if not file_path.exists(): |
| | return |
| |
|
| | with open(file_path, 'r', encoding='utf-8', errors='replace') as f: |
| | |
| | try: |
| | reader = csv.DictReader(f) |
| | |
| | while True: |
| | try: |
| | row = next(reader) |
| | yield row |
| | except StopIteration: |
| | break |
| | except csv.Error as e: |
| | logging.warning(f"CSV Parse Error in {file_path}: {e}") |
| | continue |
| | except Exception as e: |
| | logging.error(f"Failed to initialize CSV reader for {file_path}: {e}") |
| | return |
| |
|
| | def ensure_manual_dataset(): |
| | """Ensures the manual dataset file exists with headers.""" |
| | p = Path("data/manual_dataset.csv") |
| | if not p.exists(): |
| | with open(p, 'w', newline='', encoding='utf-8') as f: |
| | |
| | writer = csv.writer(f) |
| | writer.writerow([ |
| | "id", "link", "caption", "collecttime", "source", |
| | "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" |
| | ]) |
| |
|
| | @app.on_event("startup") |
| | async def startup_event(): |
| | logging.info("Application starting up...") |
| | ensure_manual_dataset() |
| | if not LITE_MODE: |
| | try: |
| | inference_logic.load_models() |
| | transcription.load_model() |
| | except Exception as e: |
| | logging.fatal(f"Could not load models. Error: {e}", exc_info=True) |
| | else: |
| | logging.info("Running in LITE mode.") |
| |
|
| | @app.get("/", response_class=HTMLResponse) |
| | async def read_root(request: Request): |
| | custom_model_available = False |
| | if not LITE_MODE: |
| | custom_model_available = inference_logic.peft_model is not None |
| | if not (Path(STATIC_DIR) / "index.html").exists(): |
| | return HTMLResponse(content="Frontend not found. Please build frontend or access via Go server.", status_code=404) |
| | return templates.TemplateResponse("index.html", { |
| | "request": request, |
| | "custom_model_available": custom_model_available, |
| | "lite_mode": LITE_MODE |
| | }) |
| |
|
| | @app.get("/model-architecture", response_class=PlainTextResponse) |
| | async def get_model_architecture(): |
| | if LITE_MODE: return "Running in LITE mode." |
| | if inference_logic.base_model: return str(inference_logic.base_model) |
| | return "Model not loaded." |
| |
|
| | @app.get("/download-dataset") |
| | async def download_dataset(): |
| | file_path = Path("data/dataset.csv") |
| | if file_path.exists(): |
| | return FileResponse(path=file_path, filename="dataset.csv", media_type='text/csv') |
| | return Response("Dataset not found.", status_code=404) |
| |
|
| | progress_message = "" |
| | def progress_hook(d): |
| | global progress_message |
| | if d['status'] == 'downloading': |
| | progress_message = f"Downloading: {d.get('_percent_str', 'N/A')} at {d.get('_speed_str', 'N/A')}\r" |
| | elif d['status'] == 'finished': |
| | progress_message = f"\nDownload finished. Preparing video assets...\n" |
| |
|
| | async def run_subprocess_async(command: list[str]): |
| | process = await asyncio.create_subprocess_exec(*command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| | stdout, stderr = await process.communicate() |
| | if process.returncode != 0: |
| | raise RuntimeError(f"Process failed:\n{stderr.decode()}") |
| | return stdout.decode() |
| |
|
| | def extract_tweet_id(url: str) -> str | None: |
| | match = re.search(r"(?:twitter|x)\.com/[^/]+/status/(\d+)", url) |
| | if match: return match.group(1) |
| | return None |
| |
|
| | def normalize_link(link: str) -> str: |
| | """Standardize links for comparison.""" |
| | if not link: return "" |
| | |
| | s = link.split('?')[0].strip().rstrip('/').replace('http://', '').replace('https://', '').replace('www.', '') |
| | return s |
| |
|
| | def check_if_processed(link: str) -> bool: |
| | target_id = extract_tweet_id(link) |
| | link_clean = normalize_link(link) |
| | |
| | for filename in ["data/dataset.csv", "data/manual_dataset.csv"]: |
| | path = Path(filename) |
| | for row in robust_read_csv(path): |
| | row_link = normalize_link(row.get('link', '')) |
| | if row_link and row_link == link_clean: return True |
| | |
| | row_id = row.get('id', '') |
| | if target_id and row_id == target_id: return True |
| | return False |
| |
|
| | async def prepare_video_assets_async(url: str) -> dict: |
| | global progress_message |
| | loop = asyncio.get_event_loop() |
| | is_local = not (url.startswith("http://") or url.startswith("https://")) |
| | video_id = "unknown" |
| | transcript_path = None |
| | |
| | if is_local: |
| | original_path = Path(url) |
| | if not original_path.exists(): raise FileNotFoundError(f"File not found: {url}") |
| | video_id = hashlib.md5(str(url).encode('utf-8')).hexdigest()[:16] |
| | metadata = {"id": video_id, "link": url, "caption": original_path.stem} |
| | else: |
| | tweet_id = extract_tweet_id(url) |
| | video_id = tweet_id if tweet_id else hashlib.md5(url.encode('utf-8')).hexdigest()[:16] |
| | sanitized_check = Path(f"data/videos/{video_id}_fixed.mp4") |
| | |
| | ydl_opts = { |
| | 'format': 'best[ext=mp4]/best', |
| | 'outtmpl': 'data/videos/%(id)s.%(ext)s', |
| | 'progress_hooks': [progress_hook], 'quiet': True, 'noplaylist': True, 'no_overwrites': True, |
| | 'writesubtitles': True, 'writeautomaticsub': True, 'subtitleslangs': ['en'] |
| | } |
| | |
| | if sanitized_check.exists(): |
| | original_path = sanitized_check |
| | metadata = {"id": video_id, "link": url, "caption": "Cached Video"} |
| | else: |
| | with yt_dlp.YoutubeDL(ydl_opts) as ydl: |
| | info = await loop.run_in_executor(None, lambda: ydl.extract_info(url, download=True)) |
| | original_path = Path(ydl.prepare_filename(info)) |
| | metadata = { |
| | "id": info.get("id", video_id), "link": info.get("webpage_url", url), |
| | "caption": info.get("description", info.get("title", "N/A")).encode('ascii', 'ignore').decode('ascii').strip()[:500], |
| | "postdatetime": info.get("upload_date", "N/A") |
| | } |
| | video_id = info.get("id", video_id) |
| |
|
| | transcript_path = next(Path("data/videos").glob(f"{video_id}*.en.vtt"), None) |
| | if not transcript_path: transcript_path = next(Path("data/videos").glob(f"{video_id}*.vtt"), None) |
| |
|
| | sanitized_path = Path(f"data/videos/{video_id}_fixed.mp4") |
| | if not sanitized_path.exists() and original_path.exists(): |
| | await run_subprocess_async(["ffmpeg", "-i", str(original_path), "-c:v", "libx264", "-preset", "fast", "-crf", "23", "-c:a", "aac", "-y", str(sanitized_path)]) |
| |
|
| | audio_path = sanitized_path.with_suffix('.wav') |
| | if not audio_path.exists() and sanitized_path.exists(): |
| | try: |
| | await run_subprocess_async(["ffmpeg", "-i", str(sanitized_path), "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", "-y", str(audio_path)]) |
| | except: pass |
| | |
| | if not transcript_path and audio_path.exists() and not LITE_MODE: |
| | transcript_path = await loop.run_in_executor(None, transcription.generate_transcript, str(audio_path)) |
| |
|
| | return {"video": str(sanitized_path), "transcript": str(transcript_path) if transcript_path else None, "metadata": metadata} |
| |
|
| | def safe_int(value): |
| | try: |
| | clean = re.sub(r'[^\d]', '', str(value)) |
| | return int(clean) if clean else 0 |
| | except Exception: |
| | return 0 |
| |
|
| | async def generate_and_save_croissant_metadata(row_data: dict) -> str: |
| | try: |
| | sanitized_data = { |
| | "id": str(row_data.get("id", "")), |
| | "link": str(row_data.get("link", "")), |
| | "visual_integrity_score": safe_int(row_data.get("visual_integrity_score")), |
| | "final_veracity_score": safe_int(row_data.get("final_veracity_score")) |
| | } |
| | video_id = sanitized_data["id"] |
| | timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S') |
| | croissant_json = { |
| | "@context": "https://schema.org/", |
| | "@type": "Dataset", |
| | "name": f"vchat-label-{video_id}", |
| | "description": f"Veracity analysis labels for video {video_id}", |
| | "url": sanitized_data["link"], |
| | "variableMeasured": sanitized_data |
| | } |
| | path = Path("metadata") / f"{video_id}_{timestamp}.json" |
| | path.write_text(json.dumps(croissant_json, indent=2)) |
| | return str(path) |
| | except Exception: |
| | return "N/A (Error)" |
| |
|
| | async def get_labels_for_link(video_url: str, gemini_config: dict, vertex_config: dict, model_selection: str, include_comments: bool, reasoning_method: str = "cot"): |
| | try: |
| | yield f"Downloading assets for {video_url}..." |
| | paths = await prepare_video_assets_async(video_url) |
| | video_path = paths["video"] |
| | transcript_text = parse_vtt(paths["transcript"]) if paths["transcript"] else "No transcript." |
| | caption = paths["metadata"].get("caption", "") |
| | |
| | yield f"Assets ready. Running inference ({model_selection}, {reasoning_method.upper()})..." |
| | final_labels = None |
| | raw_toon = "" |
| | prompt_used = "" |
| | |
| | pipeline = inference_logic.run_gemini_labeling_pipeline if model_selection == 'gemini' else inference_logic.run_vertex_labeling_pipeline |
| | config = gemini_config if model_selection == 'gemini' else vertex_config |
| | |
| | async for msg in pipeline(video_path, caption, transcript_text, config, include_comments, reasoning_method): |
| | if isinstance(msg, dict) and "parsed_data" in msg: |
| | final_labels = msg["parsed_data"] |
| | raw_toon = msg.get("raw_toon", "") |
| | prompt_used = msg.get("prompt_used", "") |
| | elif isinstance(msg, str): yield msg |
| |
|
| | if not final_labels: raise RuntimeError("No labels generated.") |
| | |
| | final_labels["meta_info"] = { |
| | "prompt_used": prompt_used, |
| | "model_selection": model_selection, |
| | "reasoning_method": reasoning_method |
| | } |
| | |
| | vec = final_labels.get("veracity_vectors", {}) |
| | mod = final_labels.get("modalities", {}) |
| | fin = final_labels.get("final_assessment", {}) |
| | tags = final_labels.get("tags", []) |
| | |
| | row = { |
| | "id": paths["metadata"]["id"], |
| | "link": paths["metadata"]["link"], |
| | "caption": caption, |
| | "postdatetime": paths["metadata"].get("postdatetime", ""), |
| | "collecttime": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
| | "videotranscriptionpath": paths["transcript"] or "", |
| | "visual_integrity_score": vec.get("visual_integrity_score", "0"), |
| | "audio_integrity_score": vec.get("audio_integrity_score", "0"), |
| | "source_credibility_score": vec.get("source_credibility_score", "0"), |
| | "logical_consistency_score": vec.get("logical_consistency_score", "0"), |
| | "emotional_manipulation_score": vec.get("emotional_manipulation_score", "0"), |
| | "video_audio_score": mod.get("video_audio_score", "0"), |
| | "video_caption_score": mod.get("video_caption_score", "0"), |
| | "audio_caption_score": mod.get("audio_caption_score", "0"), |
| | "final_veracity_score": fin.get("veracity_score_total", "0"), |
| | "final_reasoning": fin.get("reasoning", ""), |
| | "tags": ", ".join(tags) |
| | } |
| | yield {"csv_row": row, "full_json": final_labels, "raw_toon": raw_toon} |
| |
|
| | except Exception as e: |
| | yield {"error": str(e)} |
| |
|
| | @app.get("/queue/list") |
| | async def get_queue_list(): |
| | queue_path = Path("data/batch_queue.csv") |
| | items = [] |
| | |
| | for row in robust_read_csv(queue_path): |
| | if len(row) > 0: |
| | link = row.get("link") |
| | if not link: continue |
| | status = "Processed" if check_if_processed(link) else "Pending" |
| | items.append({ |
| | "link": link, |
| | "timestamp": row.get("ingest_timestamp", ""), |
| | "status": status |
| | }) |
| | return items |
| |
|
| | @app.delete("/queue/delete") |
| | async def delete_queue_item(link: str): |
| | queue_path = Path("data/batch_queue.csv") |
| | if not queue_path.exists(): |
| | return {"status": "error", "message": "Queue file not found"} |
| | |
| | rows = [] |
| | deleted = False |
| | try: |
| | |
| | |
| | |
| | |
| | |
| | with open(queue_path, 'r', encoding='utf-8', errors='replace') as f: |
| | reader = csv.reader(f) |
| | try: header = next(reader) |
| | except StopIteration: header = ["link", "ingest_timestamp"] |
| | |
| | all_rows = list(robust_read_csv(queue_path)) |
| | new_rows = [] |
| | for row in all_rows: |
| | if row.get("link") == link: |
| | deleted = True |
| | else: |
| | new_rows.append(row) |
| | |
| | if deleted: |
| | with open(queue_path, 'w', newline='', encoding='utf-8') as f: |
| | writer = csv.DictWriter(f, fieldnames=header) |
| | writer.writeheader() |
| | writer.writerows(new_rows) |
| | return {"status": "success", "link": link} |
| | else: |
| | return {"status": "not_found", "message": "Link not found in queue"} |
| | |
| | except Exception as e: |
| | return {"status": "error", "message": str(e)} |
| |
|
| | @app.post("/queue/stop") |
| | async def stop_queue_processing(): |
| | global STOP_QUEUE_SIGNAL |
| | STOP_QUEUE_SIGNAL = True |
| | return {"status": "stopping"} |
| |
|
| | @app.post("/queue/upload_csv") |
| | async def upload_csv_to_queue(file: UploadFile = File(...)): |
| | try: |
| | content = await file.read() |
| | decoded = content.decode('utf-8').splitlines() |
| | reader = csv.reader(decoded) |
| | links_to_add = [] |
| | header = next(reader, None) |
| | if not header: return {"status": "empty file"} |
| |
|
| | link_idx = 0 |
| | header_lower = [h.lower() for h in header] |
| | if "link" in header_lower: link_idx = header_lower.index("link") |
| | elif "url" in header_lower: link_idx = header_lower.index("url") |
| | elif "http" in header[0]: |
| | links_to_add.append(header[0]) |
| | link_idx = 0 |
| | |
| | for row in reader: |
| | if len(row) > link_idx and row[link_idx].strip(): |
| | links_to_add.append(row[link_idx].strip()) |
| |
|
| | queue_path = Path("data/batch_queue.csv") |
| | existing_links = set() |
| | if queue_path.exists(): |
| | with open(queue_path, 'r', encoding='utf-8', errors='replace') as f: |
| | existing_links = set(f.read().splitlines()) |
| |
|
| | added_count = 0 |
| | with open(queue_path, 'a', newline='', encoding='utf-8') as f: |
| | writer = csv.writer(f) |
| | if not queue_path.exists() or queue_path.stat().st_size == 0: |
| | writer.writerow(["link", "ingest_timestamp"]) |
| | |
| | for link in links_to_add: |
| | duplicate = False |
| | for line in existing_links: |
| | if link in line: |
| | duplicate = True |
| | break |
| | if duplicate: continue |
| |
|
| | writer.writerow([link, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")]) |
| | added_count += 1 |
| | |
| | return {"status": "success", "added": added_count} |
| | except Exception as e: |
| | return JSONResponse(status_code=400, content={"error": str(e), "status": "failed"}) |
| |
|
| | @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(""), |
| | include_comments: bool = Form(False), |
| | reasoning_method: str = Form("cot") |
| | ): |
| | global STOP_QUEUE_SIGNAL |
| | STOP_QUEUE_SIGNAL = False |
| | gemini_config = {"api_key": gemini_api_key, "model_name": gemini_model_name} |
| | vertex_config = {"project_id": vertex_project_id, "location": vertex_location, "model_name": vertex_model_name, "api_key": vertex_api_key} |
| | |
| | async def queue_stream(): |
| | queue_path = Path("data/batch_queue.csv") |
| | items = [] |
| | for row in robust_read_csv(queue_path): |
| | l = row.get("link") |
| | if l: items.append(l) |
| | |
| | if not items: |
| | yield "data: Queue empty.\n\n" |
| | return |
| | |
| | processed_count = 0 |
| | total = len(items) |
| | |
| | for i, link in enumerate(items): |
| | if STOP_QUEUE_SIGNAL: |
| | yield "data: [SYSTEM] Stopped by user.\n\n" |
| | break |
| | |
| | if check_if_processed(link): |
| | yield f"data: [SKIP] {link} processed.\n\n" |
| | continue |
| | |
| | yield f"data: [START] {i+1}/{total}: {link}\n\n" |
| | final_data = None |
| | async for res in get_labels_for_link(link, gemini_config, vertex_config, model_selection, include_comments, reasoning_method): |
| | if isinstance(res, str): yield f"data: {res}\n\n" |
| | if isinstance(res, dict) and "csv_row" in res: final_data = res |
| | |
| | if final_data: |
| | row = final_data["csv_row"] |
| | vid_id = row["id"] |
| | ts = datetime.datetime.now().strftime('%Y%m%d%H%M%S') |
| | |
| | |
| | json_path = f"data/labels/{vid_id}_{ts}_labels.json" |
| | with open(json_path, 'w') as f: json.dump(final_data["full_json"], f, indent=2) |
| | |
| | |
| | with open(f"data/labels/{vid_id}_{ts}.toon", 'w') as f: f.write(final_data["raw_toon"]) |
| | |
| | |
| | prompt_content = final_data.get("full_json", {}).get("meta_info", {}).get("prompt_used", "") |
| | if prompt_content: |
| | with open(f"data/prompts/{vid_id}_{ts}_prompt.txt", 'w', encoding='utf-8') as f: |
| | f.write(prompt_content) |
| |
|
| | |
| | raw_response = final_data.get("raw_toon", "") |
| | if raw_response: |
| | with open(f"data/responses/{vid_id}.txt", 'w', encoding='utf-8') as f: |
| | f.write(raw_response) |
| |
|
| | row["metadatapath"] = await generate_and_save_croissant_metadata(row) |
| | row["json_path"] = json_path |
| | |
| | |
| | dpath = Path("data/dataset.csv") |
| | exists = dpath.exists() |
| | with open(dpath, 'a', newline='', encoding='utf-8') as f: |
| | writer = csv.DictWriter(f, fieldnames=list(row.keys()), extrasaction='ignore') |
| | if not exists: writer.writeheader() |
| | writer.writerow(row) |
| | |
| | processed_count += 1 |
| | yield f"data: [SUCCESS] Labeled.\n\n" |
| | else: |
| | yield f"data: [FAIL] Failed to label.\n\n" |
| | |
| | yield f"data: Batch Complete. +{processed_count} videos labeled.\n\n" |
| | yield "event: close\ndata: Done\n\n" |
| |
|
| | return StreamingResponse(queue_stream(), media_type="text/event-stream") |
| |
|
| | @app.post("/extension/ingest") |
| | async def extension_ingest(request: Request): |
| | try: |
| | data = await request.json() |
| | link = data.get("link") |
| | if not link: raise HTTPException(status_code=400, detail="No link") |
| | queue_path = Path("data/batch_queue.csv") |
| | file_exists = queue_path.exists() |
| | |
| | if file_exists: |
| | with open(queue_path, 'r', encoding='utf-8', errors='replace') as f: |
| | if link in f.read(): |
| | return {"status": "queued", "msg": "Duplicate"} |
| | |
| | with open(queue_path, 'a', newline='', encoding='utf-8') as f: |
| | writer = csv.writer(f) |
| | if not file_exists: writer.writerow(["link", "ingest_timestamp"]) |
| | writer.writerow([link, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")]) |
| | |
| | return {"status": "queued", "link": link} |
| | 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 or not comments: raise HTTPException(status_code=400, detail="Missing data") |
| | |
| | csv_path = Path("data/comments.csv") |
| | exists = csv_path.exists() |
| | |
| | |
| | fieldnames = ["link", "author", "comment_text", "timestamp"] |
| | |
| | with open(csv_path, 'a', newline='', encoding='utf-8') as f: |
| | writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore') |
| | if not exists: writer.writeheader() |
| | |
| | ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
| | for c in comments: |
| | row = {"link": link, "timestamp": ts} |
| | if isinstance(c, dict): |
| | row["author"] = c.get("author", "Unknown") |
| | row["comment_text"] = c.get("text", "").strip() |
| | else: |
| | |
| | row["author"] = "Unknown" |
| | row["comment_text"] = str(c).strip() |
| | |
| | if row["comment_text"]: |
| | writer.writerow(row) |
| | |
| | return {"status": "saved", "count": len(comments)} |
| | except Exception as e: |
| | raise HTTPException(status_code=500, detail=str(e)) |
| |
|
| | @app.post("/extension/save_manual") |
| | async def extension_save_manual(request: Request): |
| | try: |
| | data = await request.json() |
| | link = data.get("link") |
| | labels = data.get("labels", {}) |
| | stats = data.get("stats", {}) |
| | tags = data.get("tags", "") |
| | |
| | if not link: raise HTTPException(status_code=400, detail="No link") |
| | |
| | video_id = extract_tweet_id(link) or hashlib.md5(link.encode()).hexdigest()[:16] |
| | |
| | |
| | ensure_manual_dataset() |
| |
|
| | |
| | row_data = { |
| | "id": video_id, |
| | "link": link, |
| | "caption": data.get("caption", ""), |
| | "collecttime": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
| | "source": "manual_extension", |
| | |
| | |
| | "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", 0), |
| | |
| | |
| | "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", ""), |
| | |
| | |
| | "stats_likes": stats.get("likes", 0), |
| | "stats_shares": stats.get("shares", 0), |
| | "stats_comments": stats.get("comments", 0), |
| | "stats_platform": stats.get("platform", "unknown"), |
| | "tags": tags |
| | } |
| | |
| | |
| | dpath = Path("data/manual_dataset.csv") |
| | rows = [] |
| | replaced = False |
| | |
| | |
| | if dpath.exists(): |
| | rows = list(robust_read_csv(dpath)) |
| | |
| | new_rows = [] |
| | for r in rows: |
| | if r.get('id') == video_id: |
| | new_rows.append(row_data) |
| | replaced = True |
| | else: |
| | new_rows.append(r) |
| | |
| | if not replaced: |
| | new_rows.append(row_data) |
| | |
| | |
| | with open(dpath, 'w', newline='', encoding='utf-8') as f: |
| | writer = csv.DictWriter(f, fieldnames=list(row_data.keys()), extrasaction='ignore') |
| | writer.writeheader() |
| | writer.writerows(new_rows) |
| |
|
| | |
| | ai_path = Path("data/dataset.csv") |
| | ai_data = None |
| | if ai_path.exists(): |
| | for row in robust_read_csv(ai_path): |
| | |
| | r_link = normalize_link(row.get('link', '')) |
| | t_link = normalize_link(link) |
| | |
| | if r_link == t_link or row.get('id') == video_id: |
| | ai_data = row |
| | break |
| | |
| | if ai_data: |
| | |
| | comp_path = Path("data/comparison.csv") |
| | comp_exists = comp_path.exists() |
| | |
| | |
| | def get_int(d, k): |
| | try: |
| | |
| | val = str(d.get(k, 0)) |
| | val = re.sub(r'[^\d]', '', val) |
| | return int(val) if val else 0 |
| | except: return 0 |
| |
|
| | comparison_row = { |
| | "id": video_id, |
| | "link": link, |
| | "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), |
| | |
| | |
| | "ai_visual": get_int(ai_data, "visual_integrity_score"), |
| | "manual_visual": row_data["visual_integrity_score"], |
| | "delta_visual": get_int(ai_data, "visual_integrity_score") - row_data["visual_integrity_score"], |
| |
|
| | |
| | "ai_audio": get_int(ai_data, "audio_integrity_score"), |
| | "manual_audio": row_data["audio_integrity_score"], |
| | "delta_audio": get_int(ai_data, "audio_integrity_score") - row_data["audio_integrity_score"], |
| | |
| | |
| | "ai_source": get_int(ai_data, "source_credibility_score"), |
| | "manual_source": row_data["source_credibility_score"], |
| | "delta_source": get_int(ai_data, "source_credibility_score") - row_data["source_credibility_score"], |
| |
|
| | |
| | "ai_logic": get_int(ai_data, "logical_consistency_score"), |
| | "manual_logic": row_data["logical_consistency_score"], |
| | "delta_logic": get_int(ai_data, "logical_consistency_score") - row_data["logical_consistency_score"], |
| |
|
| | |
| | "ai_final": get_int(ai_data, "final_veracity_score"), |
| | "manual_final": row_data["final_veracity_score"], |
| | "delta_final": get_int(ai_data, "final_veracity_score") - row_data["final_veracity_score"] |
| | } |
| |
|
| | |
| | comp_rows = [] |
| | if comp_exists: |
| | comp_rows = list(robust_read_csv(comp_path)) |
| | |
| | final_comp_rows = [] |
| | comp_replaced = False |
| | for cr in comp_rows: |
| | if cr.get('id') == video_id: |
| | final_comp_rows.append(comparison_row) |
| | comp_replaced = True |
| | else: |
| | final_comp_rows.append(cr) |
| | if not comp_replaced: |
| | final_comp_rows.append(comparison_row) |
| | |
| | with open(comp_path, 'w', newline='', encoding='utf-8') as f: |
| | writer = csv.DictWriter(f, fieldnames=list(comparison_row.keys()), extrasaction='ignore') |
| | writer.writeheader() |
| | writer.writerows(final_comp_rows) |
| | |
| | return {"status": "saved", "compared": True if ai_data else False} |
| | except Exception as e: |
| | raise HTTPException(status_code=500, detail=str(e)) |
| |
|
| | @app.get("/workflow/status") |
| | async def get_workflow_status(): |
| | """ |
| | Returns a list of all known links (from queue and AI dataset), |
| | indicating whether they have been Manually Labeled. |
| | Matches primarily on ID (Tweet ID or Hash), falling back to Link. |
| | """ |
| | all_links = {} |
| |
|
| | def get_canonical_key(link, rid=None): |
| | |
| | tid = extract_tweet_id(link) |
| | if tid: return tid |
| | |
| | if rid and str(rid).strip(): return str(rid).strip() |
| | |
| | return normalize_link(link) |
| |
|
| | |
| | qp = Path("data/batch_queue.csv") |
| | for row in robust_read_csv(qp): |
| | url = row.get("link", "").strip() |
| | if url: |
| | |
| | key = get_canonical_key(url) |
| | all_links[key] = { |
| | "link": url, "source": "queue", |
| | "ai_status": "Pending", "manual_status": "Pending", |
| | "ai_data": None |
| | } |
| |
|
| | |
| | dp = Path("data/dataset.csv") |
| | for row in robust_read_csv(dp): |
| | url = row.get("link", "").strip() |
| | rid = row.get("id", "").strip() |
| | |
| | |
| | key = get_canonical_key(url, rid) |
| | |
| | if key not in all_links: |
| | all_links[key] = {"link": url, "source": "dataset", "manual_status": "Pending"} |
| | |
| | all_links[key]["ai_status"] = "Labeled" |
| | all_links[key]["ai_data"] = { |
| | "visual": row.get("visual_integrity_score"), |
| | "final": row.get("final_veracity_score"), |
| | "reasoning": row.get("final_reasoning"), |
| | "tags": row.get("tags", "") |
| | } |
| |
|
| | |
| | mp = Path("data/manual_dataset.csv") |
| | for row in robust_read_csv(mp): |
| | url = row.get("link", "").strip() |
| | rid = row.get("id", "").strip() |
| | |
| | key = get_canonical_key(url, rid) |
| | |
| | if key in all_links: |
| | all_links[key]["manual_status"] = "Completed" |
| | all_links[key]["manual_tags"] = row.get("tags", "") |
| | else: |
| | |
| | all_links[key] = { |
| | "link": url, "source": "manual_only", |
| | "ai_status": "Unknown", "manual_status": "Completed", |
| | "manual_tags": row.get("tags", "") |
| | } |
| |
|
| | return list(all_links.values()) |
| |
|
| | @app.get("/manage/list") |
| | async def list_data(): |
| | data = [] |
| |
|
| | |
| | manual_index = set() |
| | mp = Path("data/manual_dataset.csv") |
| | for row in robust_read_csv(mp): |
| | if row.get('link'): manual_index.add(normalize_link(row['link'])) |
| | if row.get('id'): manual_index.add(row['id'].strip()) |
| |
|
| | def read_csv(path, source_type): |
| | for row in robust_read_csv(path): |
| | if not row.get('id') or row['id'].strip() == "": |
| | link = row.get('link', '') |
| | tid = extract_tweet_id(link) |
| | row['id'] = tid if tid else hashlib.md5(link.encode()).hexdigest()[:16] |
| | |
| | json_content = None |
| | if row.get('json_path') and os.path.exists(row['json_path']): |
| | try: |
| | with open(row['json_path'], 'r') as jf: json_content = json.load(jf) |
| | except: pass |
| | |
| | row['source_type'] = source_type |
| | row['json_data'] = json_content |
| | |
| | |
| | if source_type == "auto": |
| | lid = row.get('id') |
| | llink = normalize_link(row.get('link', '')) |
| | if lid in manual_index or llink in manual_index: |
| | row['manual_verification_status'] = "Verified" |
| | else: |
| | row['manual_verification_status'] = "Need Manual" |
| | |
| | data.append(row) |
| |
|
| | read_csv(Path("data/dataset.csv"), "auto") |
| | read_csv(Path("data/manual_dataset.csv"), "manual") |
| | data.sort(key=lambda x: x.get('collecttime', ''), reverse=True) |
| | return data |
| |
|
| | @app.get("/manage/comparison_data") |
| | async def get_comparison_data(): |
| | """ |
| | Returns an aggregated dataset joining AI Labels vs Manual Labels for visualization. |
| | """ |
| | ai_data = {} |
| | |
| | for row in robust_read_csv(Path("data/dataset.csv")): |
| | |
| | key = row.get("id") |
| | if not key: key = normalize_link(row.get("link")) |
| | ai_data[key] = row |
| | |
| | comparisons = [] |
| | |
| | |
| | for manual in robust_read_csv(Path("data/manual_dataset.csv")): |
| | key = manual.get("id") |
| | if not key: key = normalize_link(manual.get("link")) |
| | |
| | if key in ai_data: |
| | ai = ai_data[key] |
| | |
| | def get_score(d, k): |
| | try: |
| | val = str(d.get(k, 0)) |
| | val = re.sub(r'[^\d]', '', val) |
| | return int(val) if val else 0 |
| | except: return 0 |
| | |
| | item = { |
| | "id": key, |
| | "link": manual.get("link"), |
| | "scores": { |
| | "visual": {"ai": get_score(ai, "visual_integrity_score"), "manual": get_score(manual, "visual_integrity_score")}, |
| | "audio": {"ai": get_score(ai, "audio_integrity_score"), "manual": get_score(manual, "audio_integrity_score")}, |
| | "final": {"ai": get_score(ai, "final_veracity_score"), "manual": get_score(manual, "final_veracity_score")} |
| | } |
| | } |
| | |
| | item["deltas"] = { |
| | "visual": item["scores"]["visual"]["ai"] - item["scores"]["visual"]["manual"], |
| | "audio": item["scores"]["audio"]["ai"] - item["scores"]["audio"]["manual"], |
| | "final": item["scores"]["final"]["ai"] - item["scores"]["final"]["manual"] |
| | } |
| | comparisons.append(item) |
| | |
| | return comparisons |
| |
|
| | @app.delete("/manage/delete") |
| | async def delete_data(id: str = "", link: str = ""): |
| | if not id and not link: raise HTTPException(status_code=400, detail="Must provide ID or Link") |
| | deleted_count = 0 |
| | target_id = id |
| | |
| | def remove_from_csv(path): |
| | nonlocal deleted_count, target_id |
| | if not path.exists(): return |
| | |
| | |
| | rows = list(robust_read_csv(path)) |
| | |
| | |
| | fieldnames = [] |
| | with open(path, 'r', encoding='utf-8', errors='replace') as f: |
| | reader = csv.DictReader(f) |
| | fieldnames = reader.fieldnames |
| | |
| | new_rows = [] |
| | found_in_file = False |
| | for row in rows: |
| | is_match = False |
| | if id and row.get('id') == id: is_match = True |
| | elif link and normalize_link(row.get('link', '')) == normalize_link(link): is_match = True |
| | |
| | if is_match: |
| | found_in_file = True |
| | deleted_count += 1 |
| | if not target_id: target_id = row.get('id') |
| | else: |
| | new_rows.append(row) |
| |
|
| | if found_in_file and fieldnames: |
| | with open(path, 'w', newline='', encoding='utf-8') as f: |
| | writer = csv.DictWriter(f, fieldnames=fieldnames) |
| | writer.writeheader() |
| | writer.writerows(new_rows) |
| |
|
| | remove_from_csv(Path("data/dataset.csv")) |
| | remove_from_csv(Path("data/manual_dataset.csv")) |
| | if target_id: |
| | for p in Path("data/labels").glob(f"{target_id}_*"): p.unlink(missing_ok=True) |
| | for p in Path("metadata").glob(f"{target_id}_*"): p.unlink(missing_ok=True) |
| | return {"status": "deleted", "count": deleted_count} |
| |
|
| | @app.post("/label_video") |
| | async def label_video_endpoint( |
| | video_url: str = Form(...), 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(""), |
| | include_comments: bool = Form(False), |
| | reasoning_method: str = Form("cot") |
| | ): |
| | gemini_config = {"api_key": gemini_api_key, "model_name": gemini_model_name} |
| | vertex_config = {"project_id": vertex_project_id, "location": vertex_location, "model_name": vertex_model_name, "api_key": vertex_api_key} |
| | async def stream(): |
| | async for msg in get_labels_for_link(video_url, gemini_config, vertex_config, model_selection, include_comments, reasoning_method): |
| | if isinstance(msg, str): yield f"data: {msg}\n\n" |
| | if isinstance(msg, dict) and "csv_row" in msg: yield "data: Done. Labels generated.\n\n" |
| | yield "event: close\ndata: Done.\n\n" |
| | return StreamingResponse(stream(), media_type="text/event-stream") |
