Spaces:
Runtime error
Runtime error
File size: 8,129 Bytes
e0f2d0e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 |
from schemas.text_schemas import AITextDetector
from schemas.vision_schemas import FakeFaceDetector
from schemas.vision_schemas import FaceDetector
from schemas.text_schemas import FakeNewsDetector
from schemas.vision_schemas import AIImageDetector
from schemas.text_schemas import EmotionDetector
from schemas.text_schemas import SearchQueryExtractor
from schemas.fake_manager_schemas import News, AggregatedNewsAnalysis, ImageAnalysis
from schemas.vision_schemas import FaceMainPoints
from services.fact_search_service import FactCheckService
from utils.utils import open_image
from typing import List, Optional, Union
from PIL import Image as PILImage
from pathlib import Path
import uuid
from datetime import datetime
class FakeNewsManager:
"""Manager that aggregates multiple detectors and services to analyze a single
news item (text + images) and produce an aggregated analysis.
Responsibilities:
- Run text-based detectors (AI text detector, fake-news/text classifier).
- Extract search queries and run fact-checking.
- Run emotion analysis on the text.
- Run image-level detectors (face detection, AI-image detection, deepfake
face detection) and crop faces for per-face analysis.
Attributes:
ai_text_detector: Optional AI text detector; must provide `.detect(text) -> bool|float|None`.
fake_face_detector: Optional face-level deepfake detector; must provide `.detect(pil_image) -> bool|float|None`.
face_detector: Optional face detector; must provide `.detect(pil_image) -> list[FaceMainPoints]`.
news_detector: Optional fake-news/text detector; must provide `.detect(text) -> bool|float|None`.
ai_image_detector: Optional AI-image detector; must provide `.detect(pil_image) -> bool|float|None`.
query_extractor: Optional extractor that returns list[str] from text.
emotion_detector: Optional emotion detector; must provide `.analyze(text)`.
fact_checker: Optional fact-check service; must provide `.verify_claim(query)`.
"""
ai_text_detector: Optional[AITextDetector]
fake_face_detector: Optional[FakeFaceDetector]
face_detector: Optional[FaceDetector]
news_detector: Optional[FakeNewsDetector]
ai_image_detector: Optional[AIImageDetector]
query_extractor: Optional[SearchQueryExtractor]
emotion_detector: Optional[EmotionDetector]
fact_checker: Optional[FactCheckService]
def __init__(
self,
*,
ai_text_detector: Optional[AITextDetector] = None,
fake_face_detector: Optional[FakeFaceDetector] = None,
face_detector: Optional[FaceDetector] = None,
news_detector: Optional[FakeNewsDetector] = None,
ai_image_detector: Optional[AIImageDetector] = None,
query_extractor: Optional[SearchQueryExtractor] = None,
emotion_detector: Optional[EmotionDetector] = None,
fact_checker: Optional[FactCheckService] = None,
) -> None:
"""Create a FakeNewsManager.
All parameters are optional; missing detectors/services are simply skipped
during analysis. Types are intentionally permissive to accommodate a
variety of detector implementations used in this project.
"""
self.ai_text_detector = ai_text_detector
self.fake_face_detector = fake_face_detector
self.face_detector = face_detector
self.news_detector = news_detector
self.ai_image_detector = ai_image_detector
self.query_extractor = query_extractor
self.emotion_detector = emotion_detector
self.fact_checker = fact_checker
def test(self) -> None:
"""Lightweight method used for quick smoke tests.
Intended for interactive debugging only; it prints a short marker.
"""
print("test")
def _crop_face(self, img: PILImage, face_mp: FaceMainPoints) -> PILImage:
"""Crop a face region from a PIL image using coordinates from
a `FaceMainPoints` object.
Args:
img: PIL.Image instance to crop from.
face_mp: FaceMainPoints providing `box_start_point` and
`box_end_point` coordinates as (x, y) tuples.
Returns:
A new PIL.Image containing only the cropped face region.
"""
x1, y1 = face_mp.box_start_point
x2, y2 = face_mp.box_end_point
return img.crop((x1, y1, x2, y2))
def analyze(
self,
news: News,
fakeness_score_threshold: float = 0.6,
) -> AggregatedNewsAnalysis:
"""Analyze a `News` item and return an `AggregatedNewsAnalysis`.
The method coordinates text and image analyzers, runs optional
fact-checking on extracted queries, and constructs an
`AggregatedNewsAnalysis` object that summarizes all results.
Args:
news: `News` object containing `text` (str) and `images` (list of
paths or file-like objects) to analyze.
fakeness_score_threshold: Float threshold in [0, 1] used by the
aggregated analysis to decide the final `is_fake_final_decision`.
Returns:
AggregatedNewsAnalysis populated with detector outputs and a
computed final decision.
"""
# Text detectors
is_ai_text = self.ai_text_detector.detect(news.text) if self.ai_text_detector else None
is_fake_text = self.news_detector.detect(news.text) if self.news_detector else None
# Query extraction & emotion
queries: List[str] = self.query_extractor.extract(news.text) if self.query_extractor else []
emotion = self.emotion_detector.analyze(news.text) if self.emotion_detector else None
# Run fact-checking for each extracted query; if no queries, fall back to full text
fact_check: Optional[List[object]] = None
if self.fact_checker:
fact_check = []
targets = queries if queries else [news.text]
for q in targets:
res = self.fact_checker.verify_claim(q)
if res is not None:
fact_check.append(res)
# Image-level analysis
images_analysis: List[ImageAnalysis] = []
for img_in in news.images:
img = open_image(img_in)
faces = self.face_detector.detect(img) if self.face_detector else []
is_ai_image = self.ai_image_detector.detect(img) if self.ai_image_detector else False
deepfake_faces: List[bool] = []
if self.fake_face_detector and faces:
for f in faces:
face_img = self._crop_face(img, f)
deepfake_faces.append(bool(self.fake_face_detector.detect(face_img)))
# Ensure image_path is a string as required by schema
if isinstance(img_in, (str, Path)):
image_path = str(img_in)
else:
image_path = ""
images_analysis.append(
ImageAnalysis(
image_path=image_path,
is_ai_image=is_ai_image,
faces=faces,
deepfake_faces=deepfake_faces,
)
)
aggregated_news_analysis = AggregatedNewsAnalysis(
is_fake_final_decision=None,
analysis_timestamp=datetime.now().isoformat(),
analysis_id=str(uuid.uuid4()),
text=news.text,
is_ai_text=is_ai_text,
is_fake_text=is_fake_text,
queries=queries,
emotion=emotion,
fact_check=fact_check,
images=images_analysis,
)
# Compute final decision using the AggregatedNewsAnalysis helper
aggregated_news_analysis.is_fake_final_decision = (
aggregated_news_analysis.compute_final_decision(fakeness_score_threshold)
)
return aggregated_news_analysis
|