import asyncio import inspect import re from img.core import analyze_saved_images from text_module.pipeline import verify_text_logic from text_module.TextAnalysisResult import TextAnalysisResult # --- HELPER: CHECK FAKE USING ONLY ASSESSMENT LINE --- def is_verdict_fake(assessment_string): if not assessment_string: return False s = assessment_string.lower().strip() # If it explicitly says "real photo", it's definitely not fake. if "real photo" in s: return False # Keywords that indicate a fake verdict fake_keywords = ["not real", "fake", "manipulated", "generated", "artificial", "synthetic"] for kw in fake_keywords: if kw in s: return True return False # --- HELPER: PARSE REPORT --- def parse_child_report(report_text): """ Use regex to extract each specific line value. This version is more robust to handle markdown and emojis. """ data = { "auth": "❓ **UNCLEAR**", "tools": "Unknown", "synth": "N/A", "artifacts": "" } if not report_text: return data # 1. Extract Authenticity Assessment (most important) # This regex is designed to be more flexible auth_match = re.search(r"Authenticity Assessment:\s*(.*)", report_text) if auth_match: auth_text = auth_match.group(1).strip() if auth_text: data["auth"] = auth_text # 2. Extract Tools tools_match = re.search(r"Verification Tools & Methods:\s*(.+)", report_text) if tools_match: data["tools"] = tools_match.group(1).strip() # 3. Extract Synthetic Type synth_match = re.search(r"Synthetic Type \(if applicable\):\s*(.+)", report_text) if synth_match: data["synth"] = synth_match.group(1).strip() # 4. Extract Artifacts (from that line through the end) art_match = re.search(r"Other Artifacts:\s*(.*)", report_text, re.DOTALL) if art_match: data["artifacts"] = art_match.group(1).strip() return data # --- HELPER: CHECK FAKE USING ONLY ASSESSMENT LINE --- def is_verdict_fake(assessment_string): if not assessment_string: return False s = assessment_string.lower().strip() # If it explicitly says "real photo", it's definitely not fake. if "real photo" in s: return False # Keywords that indicate a fake verdict fake_keywords = ["not real", "fake", "manipulated", "generated", "artificial", "synthetic"] for kw in fake_keywords: if kw in s: return True return False # --- HTML STATUS BAR (KEEP LOGIC) --- def create_status_html(label, status, message): color = "#9ca3af"; percent = 5; icon = "⏳"; bg_pulse = ""; text_color = "#374151" if status == 'processing': color = "#2563eb"; percent = 60; icon = "⚙️"; bg_pulse = "animation: pulse 2s cubic-bezier(0.4, 0, 0.6, 1) infinite;" elif status == 'done': color = "#16a34a"; percent = 100; icon = "✅" elif status == 'error': color = "#dc2626"; percent = 100; icon = "❌" html = f"""
{icon} {label} {message}
""" return html # --- TASK 1: PROCESS IMAGES --- async def run_image_task(shared_state, image_input): shared_state['img_status'] = 'processing' shared_state['img_msg'] = "Scanning artifacts..." img_result_obj = TextAnalysisResult() try: final_json = {} final_report_md = "" gen = analyze_saved_images(image_input) if inspect.isasyncgen(gen): async for res in gen: final_json, final_report_md = res else: for res in gen: final_json, final_report_md = res # Save the full report string into artifacts img_result_obj.set_other_artifacts(final_report_md) # Parse the Auth line to update status (used for possible short-circuit) parsed = parse_child_report(final_report_md) img_result_obj.set_authenticity_assessment(parsed["auth"]) shared_state['img_status'] = 'done' shared_state['img_msg'] = "Done" except Exception as e: shared_state['img_status'] = 'error' shared_state['img_msg'] = "Error" img_result_obj.set_authenticity_assessment("Error") return img_result_obj # --- TASK 2: PROCESS TEXT --- async def run_text_task(shared_state, text_input): shared_state['txt_status'] = 'processing' shared_state['txt_msg'] = "Verifying logic..." txt_result_obj = TextAnalysisResult() try: await asyncio.to_thread(verify_text_logic, text_input, txt_result_obj) shared_state['txt_status'] = 'done' shared_state['txt_msg'] = "Done" except Exception as e: shared_state['txt_status'] = 'error' shared_state['txt_msg'] = str(e) txt_result_obj.set_authenticity_assessment("Error") return txt_result_obj # --- MAIN ORCHESTRATOR --- async def verify_multimodal_logic(image_state, text_input): shared_state = {'img_status': 'waiting', 'img_msg': 'Ready...', 'txt_status': 'waiting', 'txt_msg': 'Ready...'} def get_ui(): return create_status_html("Image Analysis", shared_state['img_status'], shared_state['img_msg']), create_status_html("Text Analysis", shared_state['txt_status'], shared_state['txt_msg']), "..." yield get_ui() task_img = asyncio.create_task(run_image_task(shared_state, image_state)) task_txt = asyncio.create_task(run_text_task(shared_state, text_input)) img_res, txt_res = None, None # Loop short-circuit while not (task_img.done() and task_txt.done()): yield get_ui() if task_img.done() and img_res is None: try: img_res = task_img.result() # Check fake verdict using only the Assessment line (short-circuit) if is_verdict_fake(img_res.get_authenticity_assessment()): if not task_txt.done(): task_txt.cancel(); shared_state['txt_msg'] = "Stopped (Image is Fake)" break except: pass if task_txt.done() and txt_res is None: try: txt_res = task_txt.result() # Check fake verdict using only the Assessment line if is_verdict_fake(txt_res.get_authenticity_assessment()): if not task_img.done(): task_img.cancel(); shared_state['img_msg'] = "Stopped (Text is Fake)" break except: pass await asyncio.sleep(0.1) if img_res is None and task_img.done(): img_res = task_img.result() if txt_res is None and task_txt.done(): txt_res = task_txt.result() if not img_res: img_res = TextAnalysisResult(authenticity_assessment="Skipped") if not txt_res: txt_res = TextAnalysisResult(authenticity_assessment="Skipped") # ========================================================================= # MERGE LOGIC: BASED ONLY ON THE ASSESSMENT LINE # ========================================================================= # 1. Parse Image Report to extract a clean "Authenticity Assessment" line img_data_parsed = parse_child_report(img_res.get_other_artifacts()) img_auth_line = img_data_parsed["auth"] # VD: "🧑 REAL PHOTO" # 2. Get the Assessment line for Text txt_auth_line = txt_res.get_authenticity_assessment() # VD: "REAL (Authentic)" # 3. Determine fake/real based on those two lines img_is_fake = is_verdict_fake(img_auth_line) txt_is_fake = is_verdict_fake(txt_auth_line) # --- FIELD 1: Authenticity Assessment --- if img_is_fake or txt_is_fake: final_auth = "🤖 NOT REAL (Fake, Manipulated, or AI)" color_hex = "#dc2626" else: final_auth = " REAL (Authentic)" color_hex = "#16a34a" # --- FIELD 2: Verification Tools --- final_tools = f"Verified by our model using algorithms SearchLLM and ImageForensics." # --- FIELD 3: Synthetic Type --- final_synth_list = [] # Only pull Synthetic Type from Image module if Image is deemed Fake if img_is_fake: s_type = img_data_parsed["synth"] if img_data_parsed["synth"] != "N/A" else "Manipulated Image" final_synth_list.append(f"**Image:** {s_type}") # Only pull Synthetic Type from Text module if Text is deemed Fake if txt_is_fake: s_type = txt_res.get_synthetic_type() if not s_type or s_type == "N/A": s_type = "Generated Content" final_synth_list.append(f"**Text:** {s_type}") final_synth_str = "\n".join(final_synth_list) if final_synth_list else "N/A" # --- FIELD 4: Other Artifacts (Display source/artifacts logic) --- final_artifacts_str = "" # Case: both are Fake -> show both if img_is_fake and txt_is_fake: final_artifacts_str = f"**[Image Evidence]**\n{img_data_parsed['artifacts']}\n\n**[Text Evidence]**\n{txt_res.get_other_artifacts()}" # Case: only Image is Fake -> show image evidence elif img_is_fake: final_artifacts_str = f"{img_data_parsed['artifacts']}" # Case: only Text is Fake -> show text evidence elif txt_is_fake: final_artifacts_str = f"{txt_res.get_other_artifacts()}" # Case: both are REAL -> show source if available else: final_artifacts_str = "Both image and text are verified as authentic by our multi-modal pipeline." # Check image source (non-empty and not N/A) img_src = img_data_parsed.get('artifacts', '').strip() if img_src and img_src != "N/A" and "No details" not in img_src: final_artifacts_str += f"\n\n**For Image:** {img_src}" # Check text source txt_src = txt_res.get_other_artifacts().strip() if txt_src and txt_src != "N/A": final_artifacts_str += f"\n\n**For Text:** {txt_src}" # BUILD FINAL MARKDOWN final_report_md = f""" ### 📋 Final Verification Report **1. Authenticity Assessment:** {final_auth} **2. Verification Tools & Methods:** {final_tools} **3. Synthetic Type (if applicable):** {final_synth_str} **4. Other Artifacts:** {final_artifacts_str} """ yield get_ui()[0], get_ui()[1], final_report_md