File size: 5,136 Bytes
3abf967
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""
liar_loader.py - LIAR Dataset Loader and Parser for TruthLens AI.

This module handles loading, parsing, and analyzing the LIAR benchmark dataset
(Wang, ACL 2017) across train, valid, and test splits. It robustly parses the
14 tab-delimited columns and computes statistics on statements, speakers,
metadata, and 6-class truth labels.
"""

import csv
import os
from typing import Dict, List, Any, Optional, Tuple
import pandas as pd
import numpy as np

LIAR_COLUMNS = [
    "id",
    "label",
    "statement",
    "subject",
    "speaker",
    "speaker_job_title",
    "state_info",
    "party_affiliation",
    "barely_true_counts",
    "false_counts",
    "half_true_counts",
    "mostly_true_counts",
    "pants_on_fire_counts",
    "context",
]

NUMERIC_COUNT_COLS = [
    "barely_true_counts",
    "false_counts",
    "half_true_counts",
    "mostly_true_counts",
    "pants_on_fire_counts",
]

# Standard 6-point scale defined by PolitiFact and LIAR
LIAR_6_LABELS = [
    "pants-fire",
    "false",
    "barely-true",
    "half-true",
    "mostly-true",
    "true",
]


def load_liar_split(filepath: str) -> Tuple[pd.DataFrame, int]:
    """
    Loads a single LIAR TSV split file.
    
    Uses csv.QUOTE_NONE to ensure unescaped quotes inside speech statements
    do not cause row misalignment or multi-line consumption.
    
    Args:
        filepath: Path to train.tsv, valid.tsv, or test.tsv.
        
    Returns:
        Tuple of (DataFrame of parsed records, count of malformed lines).
    """
    if not os.path.exists(filepath):
        raise FileNotFoundError(f"LIAR file not found: {filepath}")
        
    malformed_count = 0
    valid_rows = []
    
    with open(filepath, "r", encoding="utf-8", errors="replace") as f:
        reader = csv.reader(f, delimiter="\t", quoting=csv.QUOTE_NONE)
        for line_idx, row in enumerate(reader, start=1):
            if len(row) == 14:
                valid_rows.append(row)
            elif len(row) > 14:
                # Tab within context or statement; merge trailing elements into context
                merged_row = row[:13] + ["\t".join(row[13:])]
                valid_rows.append(merged_row)
                malformed_count += 1
            else:
                malformed_count += 1
                
    df = pd.DataFrame(valid_rows, columns=LIAR_COLUMNS)
    
    # Cast count columns to numeric
    for col in NUMERIC_COUNT_COLS:
        df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0).astype(int)
        
    return df, malformed_count


def load_all_liar_splits(data_dir: str) -> Dict[str, pd.DataFrame]:
    """
    Loads all available LIAR splits from a directory (train, valid, test).
    """
    splits = {}
    for split_name, filename in [("train", "train.tsv"), ("valid", "valid.tsv"), ("test", "test.tsv")]:
        full_path = os.path.join(data_dir, filename)
        if os.path.exists(full_path):
            df, _ = load_liar_split(full_path)
            splits[split_name] = df
    return splits


def get_liar_statistics(df: pd.DataFrame, split_name: str = "dataset") -> Dict[str, Any]:
    """
    Calculates detailed statistics for a LIAR DataFrame.
    """
    total_records = len(df)
    if total_records == 0:
        return {"total_records": 0}
        
    statement_words = df["statement"].astype(str).apply(lambda s: len(s.split()))
    statement_chars = df["statement"].astype(str).apply(len)
    
    # Missing / empty string counts
    missing_counts = {}
    for col in df.columns:
        null_cnt = int(df[col].isnull().sum())
        empty_cnt = int((df[col].astype(str).str.strip() == "").sum())
        missing_counts[col] = null_cnt + empty_cnt
        
    # Duplicates
    unique_stmts = df["statement"].nunique()
    duplicate_stmts = total_records - unique_stmts
    
    # Label counts
    label_counts = df["label"].value_counts().to_dict()
    
    # Speaker & party metrics
    top_speakers = df["speaker"].value_counts().head(10).to_dict()
    top_parties = df["party_affiliation"].value_counts().head(5).to_dict()
    top_subjects = df["subject"].value_counts().head(10).to_dict()
    
    return {
        "split": split_name,
        "total_records": total_records,
        "columns": list(df.columns),
        "label_counts": label_counts,
        "missing_or_empty_values": missing_counts,
        "unique_statements": unique_stmts,
        "duplicate_statements": duplicate_stmts,
        "statement_word_length": {
            "min": int(statement_words.min()),
            "max": int(statement_words.max()),
            "mean": float(round(statement_words.mean(), 2)),
            "median": float(round(statement_words.median(), 2)),
            "std": float(round(statement_words.std(), 2)),
        },
        "statement_char_length": {
            "min": int(statement_chars.min()),
            "max": int(statement_chars.max()),
            "mean": float(round(statement_chars.mean(), 2)),
            "median": float(round(statement_chars.median(), 2)),
        },
        "top_speakers": top_speakers,
        "top_parties": top_parties,
        "top_subjects": top_subjects,
    }