| | |
| | import re |
| | import logging |
| | import csv |
| | from io import StringIO |
| |
|
| | logger = logging.getLogger(__name__) |
| |
|
| | def parse_toon_line(line_def, data_line): |
| | if not data_line or data_line.isspace(): |
| | return {} |
| |
|
| | try: |
| | reader = csv.reader(StringIO(data_line), skipinitialspace=True) |
| | try: |
| | values = next(reader) |
| | except StopIteration: |
| | values = [] |
| | |
| | cleaned_values = [] |
| | for v in values: |
| | v_str = v.strip() |
| | v_str = v_str.replace('(', '').replace(')', '') |
| | if '/' in v_str and any(c.isdigit() for c in v_str): |
| | parts = v_str.split('/') |
| | if parts[0].strip().isdigit(): |
| | v_str = parts[0].strip() |
| | cleaned_values.append(v_str) |
| |
|
| | headers = line_def.get('headers', []) |
| | |
| | if len(cleaned_values) < len(headers): |
| | cleaned_values += [""] * (len(headers) - len(cleaned_values)) |
| | elif len(cleaned_values) > len(headers): |
| | cleaned_values = cleaned_values[:len(headers)] |
| |
|
| | return dict(zip(headers, cleaned_values)) |
| | except Exception as e: |
| | logger.error(f"Error parsing TOON line '{data_line}': {e}") |
| | return {} |
| |
|
| | def fuzzy_extract_scores(text: str) -> dict: |
| | scores = { |
| | 'visual': '0', 'audio': '0', 'source': '0', 'logic': '0', 'emotion': '0', |
| | 'video_audio': '0', 'video_caption': '0', 'audio_caption': '0' |
| | } |
| | |
| | mappings = [ |
| | ('visual', 'visual'), |
| | ('visual.*?integrity', 'visual'), |
| | ('accuracy', 'visual'), |
| | ('audio', 'audio'), |
| | ('source', 'source'), |
| | ('logic', 'logic'), |
| | ('emotion', 'emotion'), |
| | (r'video.*?audio', 'video_audio'), |
| | (r'video.*?caption', 'video_caption'), |
| | (r'audio.*?caption', 'audio_caption') |
| | ] |
| |
|
| | for pattern_str, key in mappings: |
| | pattern = re.compile(fr'(?i){pattern_str}.*?[:=\-\s\(]+(\b10\b|\b\d\b)(?:/10)?') |
| | match = pattern.search(text) |
| | if match: |
| | if scores[key] == '0': |
| | scores[key] = match.group(1) |
| | |
| | return scores |
| |
|
| | def parse_veracity_toon(text: str) -> dict: |
| | if not text: |
| | return {} |
| |
|
| | text = re.sub(r'```\w*', '', text) |
| | text = re.sub(r'```', '', text) |
| | text = text.strip() |
| |
|
| | parsed_sections = {} |
| |
|
| | block_pattern = re.compile( |
| | r'([a-zA-Z0-9_]+)\s*:\s*(?:\w+\s*)?(?:\[\s*(\d+)\s*\])?\s*\{\s*(.*?)\s*\}\s*:\s*', |
| | re.MULTILINE |
| | ) |
| | |
| | matches = list(block_pattern.finditer(text)) |
| | |
| | for i, match in enumerate(matches): |
| | key = match.group(1).lower() |
| | count = int(match.group(2)) if match.group(2) else 1 |
| | headers_str = match.group(3) |
| | headers = [h.strip().lower() for h in headers_str.split(',')] |
| | |
| | start_idx = match.end() |
| | end_idx = matches[i+1].start() if i + 1 < len(matches) else len(text) |
| | block_content = text[start_idx:end_idx].strip() |
| | |
| | lines = [line.strip() for line in block_content.splitlines() if line.strip()] |
| | |
| | data_items = [] |
| | valid_lines = [l for l in lines if len(l) > 1] |
| | |
| | for line in valid_lines[:count]: |
| | item = parse_toon_line({'key': key, 'headers': headers}, line) |
| | data_items.append(item) |
| | |
| | if count == 1 and data_items: |
| | parsed_sections[key] = data_items[0] |
| | else: |
| | parsed_sections[key] = data_items |
| |
|
| | flat_result = { |
| | 'veracity_vectors': { |
| | 'visual_integrity_score': '0', |
| | 'audio_integrity_score': '0', |
| | 'source_credibility_score': '0', |
| | 'logical_consistency_score': '0', |
| | 'emotional_manipulation_score': '0' |
| | }, |
| | 'modalities': { |
| | 'video_audio_score': '0', |
| | 'video_caption_score': '0', |
| | 'audio_caption_score': '0' |
| | }, |
| | 'video_context_summary': '', |
| | 'factuality_factors': {}, |
| | 'disinformation_analysis': {}, |
| | 'final_assessment': {} |
| | } |
| | |
| | got_vectors = False |
| | got_modalities = False |
| |
|
| | vectors_data = parsed_sections.get('vectors', []) |
| | if isinstance(vectors_data, dict): |
| | v = vectors_data |
| | if any(val and val != '0' for val in v.values()): |
| | if 'visual' in v: flat_result['veracity_vectors']['visual_integrity_score'] = v['visual'] |
| | if 'audio' in v: flat_result['veracity_vectors']['audio_integrity_score'] = v['audio'] |
| | if 'source' in v: flat_result['veracity_vectors']['source_credibility_score'] = v['source'] |
| | if 'logic' in v: flat_result['veracity_vectors']['logical_consistency_score'] = v['logic'] |
| | if 'emotion' in v: flat_result['veracity_vectors']['emotional_manipulation_score'] = v['emotion'] |
| | got_vectors = True |
| |
|
| | elif isinstance(vectors_data, list): |
| | for item in vectors_data: |
| | cat = item.get('category', '').lower() |
| | score = item.get('score', '0') |
| | if score and score != '0': |
| | got_vectors = True |
| | if 'visual' in cat: flat_result['veracity_vectors']['visual_integrity_score'] = score |
| | elif 'audio' in cat: flat_result['veracity_vectors']['audio_integrity_score'] = score |
| | elif 'source' in cat: flat_result['veracity_vectors']['source_credibility_score'] = score |
| | elif 'logic' in cat: flat_result['veracity_vectors']['logical_consistency_score'] = score |
| | elif 'emotion' in cat: flat_result['veracity_vectors']['emotional_manipulation_score'] = score |
| |
|
| | modalities_data = parsed_sections.get('modalities', []) |
| | if isinstance(modalities_data, dict): |
| | m = modalities_data |
| | for k, v in m.items(): |
| | k_clean = k.lower().replace(' ', '').replace('-', '').replace('_', '') |
| | if 'videoaudio' in k_clean: flat_result['modalities']['video_audio_score'] = v |
| | elif 'videocaption' in k_clean: flat_result['modalities']['video_caption_score'] = v |
| | elif 'audiocaption' in k_clean: flat_result['modalities']['audio_caption_score'] = v |
| | if v and v != '0': got_modalities = True |
| |
|
| | elif isinstance(modalities_data, list): |
| | for item in modalities_data: |
| | cat = item.get('category', '').lower().replace(' ', '').replace('-', '').replace('_', '') |
| | score = item.get('score', '0') |
| | if score and score != '0': |
| | got_modalities = True |
| | if 'videoaudio' in cat: flat_result['modalities']['video_audio_score'] = score |
| | elif 'videocaption' in cat: flat_result['modalities']['video_caption_score'] = score |
| | elif 'audiocaption' in cat: flat_result['modalities']['audio_caption_score'] = score |
| |
|
| | if not got_vectors or not got_modalities: |
| | fuzzy_scores = fuzzy_extract_scores(text) |
| | |
| | if not got_vectors: |
| | flat_result['veracity_vectors']['visual_integrity_score'] = fuzzy_scores['visual'] |
| | flat_result['veracity_vectors']['audio_integrity_score'] = fuzzy_scores['audio'] |
| | flat_result['veracity_vectors']['source_credibility_score'] = fuzzy_scores['source'] |
| | flat_result['veracity_vectors']['logical_consistency_score'] = fuzzy_scores['logic'] |
| | flat_result['veracity_vectors']['emotional_manipulation_score'] = fuzzy_scores['emotion'] |
| | |
| | if not got_modalities: |
| | flat_result['modalities']['video_audio_score'] = fuzzy_scores['video_audio'] |
| | flat_result['modalities']['video_caption_score'] = fuzzy_scores['video_caption'] |
| | flat_result['modalities']['audio_caption_score'] = fuzzy_scores['audio_caption'] |
| |
|
| | f = parsed_sections.get('factuality', {}) |
| | if isinstance(f, list): f = f[0] if f else {} |
| | flat_result['factuality_factors'] = { |
| | 'claim_accuracy': f.get('accuracy', 'Unverifiable'), |
| | 'evidence_gap': f.get('gap', ''), |
| | 'grounding_check': f.get('grounding', '') |
| | } |
| |
|
| | d = parsed_sections.get('disinfo', {}) |
| | if isinstance(d, list): d = d[0] if d else {} |
| | flat_result['disinformation_analysis'] = { |
| | 'classification': d.get('class', 'None'), |
| | 'intent': d.get('intent', 'None'), |
| | 'threat_vector': d.get('threat', 'None') |
| | } |
| |
|
| | fn = parsed_sections.get('final', {}) |
| | if isinstance(fn, list): fn = fn[0] if fn else {} |
| | flat_result['final_assessment'] = { |
| | 'veracity_score_total': fn.get('score', '0'), |
| | 'reasoning': fn.get('reasoning', '') |
| | } |
| |
|
| | s = parsed_sections.get('summary', {}) |
| | if isinstance(s, list): s = s[0] if s else {} |
| | flat_result['video_context_summary'] = s.get('text', '') |
| |
|
| | flat_result['raw_parsed_structure'] = parsed_sections |
| | |
| | return flat_result |
