DevPatel0611/TruthLens
1
1"""2stage1_ingestion.py — Load, unify, deduplicate, and persist all datasets.3 4Reads from five dataset sources (ISOT, LIAR, Kaggle Combined / News_dataset,5Multi-Domain / overall, and supplementary training folder), maps them into a6single canonical schema, performs Sentence-BERT deduplication, and writes the7result to ``data/processed/unified.csv`` together with label-distribution8statistics in ``data/processed/stats.json``.9 10Usage:11 python -m src.stage1_ingestion # from fake_news_detection/12 python src/stage1_ingestion.py # direct execution13"""14 15from __future__ import annotations16 17import json18import logging19import os20import sys21import time22import uuid23from pathlib import Path24from typing import Dict, List, Optional, Tuple25from urllib.parse import urlparse26 27import pandas as pd28import yaml29 30# ── Ensure project root is on sys.path when running directly ──31_SCRIPT_DIR = Path(__file__).resolve().parent32_PROJECT_ROOT = _SCRIPT_DIR.parent33if str(_PROJECT_ROOT) not in sys.path:34 sys.path.insert(0, str(_PROJECT_ROOT))35 36from src.utils.deduplication import deduplicate_dataframe # noqa: E40237from src.utils.text_utils import clean_empty_texts, build_full_text, word_count38 39# ═══════════════════════════════════════════════════════════40# Logger41# ═══════════════════════════════════════════════════════════42logging.basicConfig(43 level=logging.INFO,44 format="%(asctime)s │ %(levelname)-8s │ %(name)s │ %(message)s",45 datefmt="%H:%M:%S",46)47logger = logging.getLogger("stage1_ingestion")48 49 50# ═══════════════════════════════════════════════════════════51# Config loader52# ═══════════════════════════════════════════════════════════53def load_config(config_path: Optional[str] = None) -> dict:54 """Load the YAML configuration file.55 56 Args:57 config_path: Explicit path to ``config.yaml``. Falls back to58 ``<project_root>/config/config.yaml`` if not provided.59 60 Returns:61 Parsed configuration dictionary.62 """63 if config_path is None:64 config_path = str(_PROJECT_ROOT / "config" / "config.yaml")65 with open(config_path, "r", encoding="utf-8") as fh:66 cfg = yaml.safe_load(fh)67 return cfg68 69 70# ═══════════════════════════════════════════════════════════71# Schema constants72# ═══════════════════════════════════════════════════════════73UNIFIED_COLUMNS = [74 "article_id",75 "title",76 "text",77 "source_domain",78 "published_date",79 "has_date",80 "binary_label",81 "dataset_origin",82]83 84 85# ═══════════════════════════════════════════════════════════86# Helper: extract domain from URL87# ═══════════════════════════════════════════════════════════88def extract_domain(url: Optional[str]) -> str:89 """Extract the domain (netloc) from a URL string.90 91 Args:92 url: Raw URL (may be ``None`` or malformed).93 94 Returns:95 Domain string such as ``"reuters.com"`` or ``"unknown"``.96 """97 if not url or not isinstance(url, str):98 return "unknown"99 url = url.strip()100 if not url.startswith(("http://", "https://")):101 url = "http://" + url102 try:103 netloc = urlparse(url).netloc104 # Strip leading 'www.'105 if netloc.startswith("www."):106 netloc = netloc[4:]107 return netloc if netloc else "unknown"108 except Exception:109 return "unknown"110 111 112def _try_parse_date(val) -> pd.Timestamp:113 """Attempt to parse a value into a pandas Timestamp.114 115 Args:116 val: Any date-like value.117 118 Returns:119 ``pd.Timestamp`` or ``pd.NaT`` on failure.120 """121 if pd.isna(val):122 return pd.NaT123 try:124 return pd.to_datetime(val)125 except Exception:126 return pd.NaT127 128 129# ═══════════════════════════════════════════════════════════130# Dataset-specific loaders131# ═══════════════════════════════════════════════════════════132 133def load_isot(dataset_root: str) -> pd.DataFrame:134 """Load the ISOT Fake Real News dataset (``True.csv`` + ``Fake.csv``).135 136 Located at ``<dataset_root>/fake_real/``.137 138 Args:139 dataset_root: Path to the top-level Dataset folder.140 141 Returns:142 DataFrame in the unified schema.143 """144 t0 = time.perf_counter()145 logger.info("Loading ISOT dataset …")146 147 base = os.path.join(dataset_root, "fake_real")148 true_path = os.path.join(base, "True.csv")149 fake_path = os.path.join(base, "Fake.csv")150 151 df_true = pd.read_csv(true_path)152 df_true["binary_label"] = 1153 df_fake = pd.read_csv(fake_path)154 df_fake["binary_label"] = 0155 156 df = pd.concat([df_true, df_fake], ignore_index=True)157 158 # Columns: title, text, subject, date159 records: List[dict] = []160 for _, row in df.iterrows():161 pub_date = _try_parse_date(row.get("date"))162 records.append({163 "article_id": str(uuid.uuid4()),164 "title": str(row.get("title", "") or ""),165 "text": str(row.get("text", "") or ""),166 "source_domain": "unknown", # ISOT has no URL column167 "published_date": pub_date,168 "has_date": not pd.isna(pub_date),169 "binary_label": int(row["binary_label"]),170 "dataset_origin": "isot",171 })172 173 result = pd.DataFrame(records, columns=UNIFIED_COLUMNS)174 logger.info(175 "ISOT loaded: %d rows (True=%d, Fake=%d) in %.1fs",176 len(result),177 (result["binary_label"] == 1).sum(),178 (result["binary_label"] == 0).sum(),179 time.perf_counter() - t0,180 )181 return result182 183 184# ─────────────────────────────────────────────────────────185 186# LIAR label mapping187_LIAR_LABEL_MAP = {188 "true": 1,189 "mostly-true": 1,190 "half-true": 1,191 "false": 0,192 "barely-true": 0,193 "pants-fire": 0,194}195 196_LIAR_COLNAMES = [197 "id", "label", "statement", "subject", "speaker",198 "job_title", "state", "party",199 "barely_true_cnt", "false_cnt", "half_true_cnt",200 "mostly_true_cnt", "pants_fire_cnt",201 "context",202]203 204 205def load_liar(dataset_root: str) -> pd.DataFrame:206 """Load the LIAR dataset (``train.tsv``, ``valid.tsv``, ``test.tsv``).207 208 Six-class labels are mapped to binary via ``_LIAR_LABEL_MAP``.209 210 Args:211 dataset_root: Path to the top-level Dataset folder.212 213 Returns:214 DataFrame in the unified schema.215 """216 t0 = time.perf_counter()217 logger.info("Loading LIAR dataset …")218 219 base = os.path.join(dataset_root, "liar")220 frames: List[pd.DataFrame] = []221 for fname in ("train.tsv", "valid.tsv", "test.tsv"):222 fp = os.path.join(base, fname)223 if os.path.exists(fp):224 tmp = pd.read_csv(fp, sep="\t", header=None, names=_LIAR_COLNAMES)225 frames.append(tmp)226 logger.info(" %s: %d rows", fname, len(tmp))227 228 df = pd.concat(frames, ignore_index=True)229 230 records: List[dict] = []231 for _, row in df.iterrows():232 label_str = str(row.get("label", "")).strip().lower()233 binary = _LIAR_LABEL_MAP.get(label_str)234 if binary is None:235 continue # Skip rows with unrecognised labels236 237 records.append({238 "article_id": str(uuid.uuid4()),239 "title": "", # LIAR has no title240 "text": str(row.get("statement", "") or ""),241 "source_domain": "politifact.com", # All LIAR data from PolitiFact242 "published_date": pd.NaT,243 "has_date": False,244 "binary_label": binary,245 "dataset_origin": "liar",246 })247 248 result = pd.DataFrame(records, columns=UNIFIED_COLUMNS)249 logger.info(250 "LIAR loaded: %d rows (True=%d, Fake=%d) in %.1fs",251 len(result),252 (result["binary_label"] == 1).sum(),253 (result["binary_label"] == 0).sum(),254 time.perf_counter() - t0,255 )256 return result257 258 259# ─────────────────────────────────────────────────────────260 261def load_kaggle_combined(dataset_root: str) -> pd.DataFrame:262 """Load the Kaggle Combined / News_dataset folder.263 264 This folder mirrors the ISOT structure (``True.csv``, ``Fake.csv``).265 266 Args:267 dataset_root: Path to the top-level Dataset folder.268 269 Returns:270 DataFrame in the unified schema.271 """272 t0 = time.perf_counter()273 logger.info("Loading Kaggle Combined (News_dataset) …")274 275 # Note: The actual folder has a trailing space: "News _dataset"276 base = os.path.join(dataset_root, "News _dataset")277 if not os.path.isdir(base):278 # Fallback without space279 base = os.path.join(dataset_root, "News_dataset")280 281 frames: List[pd.DataFrame] = []282 283 for fname in os.listdir(base):284 fpath = os.path.join(base, fname)285 if not fname.lower().endswith(".csv"):286 continue287 try:288 tmp = pd.read_csv(fpath)289 except Exception as exc:290 logger.warning("Could not read %s: %s", fpath, exc)291 continue292 293 # Detect label294 name_lower = fname.lower()295 if "true" in name_lower or "real" in name_lower:296 tmp["binary_label"] = 1297 elif "fake" in name_lower:298 tmp["binary_label"] = 0299 elif "label" in [c.lower() for c in tmp.columns]:300 # Dynamic: if there's a label column, try to map301 label_col = [c for c in tmp.columns if c.lower() == "label"][0]302 tmp["binary_label"] = tmp[label_col].apply(303 lambda x: 1 if str(x).strip().lower() in ("1", "true", "real") else 0304 )305 else:306 logger.warning("Cannot determine label for %s — skipping.", fname)307 continue308 309 frames.append(tmp)310 logger.info(" %s: %d rows", fname, len(tmp))311 312 if not frames:313 logger.warning("No CSV files found in Kaggle Combined folder.")314 return pd.DataFrame(columns=UNIFIED_COLUMNS)315 316 df = pd.concat(frames, ignore_index=True)317 318 # Detect column names dynamically319 col_map = {c.lower().strip(): c for c in df.columns}320 321 title_col = col_map.get("title")322 text_col = col_map.get("text") or col_map.get("article") or col_map.get("content")323 date_col = col_map.get("date") or col_map.get("published_date")324 325 records: List[dict] = []326 for _, row in df.iterrows():327 pub_date = _try_parse_date(row.get(date_col)) if date_col else pd.NaT328 records.append({329 "article_id": str(uuid.uuid4()),330 "title": str(row.get(title_col, "") or "") if title_col else "",331 "text": str(row.get(text_col, "") or "") if text_col else "",332 "source_domain": "unknown",333 "published_date": pub_date,334 "has_date": not pd.isna(pub_date),335 "binary_label": int(row["binary_label"]),336 "dataset_origin": "kaggle_combined",337 })338 339 result = pd.DataFrame(records, columns=UNIFIED_COLUMNS)340 logger.info(341 "Kaggle Combined loaded: %d rows (True=%d, Fake=%d) in %.1fs",342 len(result),343 (result["binary_label"] == 1).sum(),344 (result["binary_label"] == 0).sum(),345 time.perf_counter() - t0,346 )347 return result348 349 350# ─────────────────────────────────────────────────────────351 352def _load_txt_folder(folder: str, label: int) -> List[dict]:353 """Read all ``.txt`` files in *folder* and return a list of record dicts.354 355 The first non-empty line is treated as the title; the remainder is the356 body text.357 358 Args:359 folder: Directory containing ``.txt`` article files.360 label: Binary label (0 = Fake, 1 = True) to assign.361 362 Returns:363 List of dicts suitable for DataFrame construction.364 """365 records: List[dict] = []366 if not os.path.isdir(folder):367 return records368 for fname in sorted(os.listdir(folder)):369 if not fname.endswith(".txt"):370 continue371 fpath = os.path.join(folder, fname)372 try:373 with open(fpath, "r", encoding="utf-8", errors="replace") as fh:374 lines = fh.read().strip().splitlines()375 except Exception:376 continue377 title = lines[0].strip() if lines else ""378 body = "\n".join(lines[1:]).strip() if len(lines) > 1 else ""379 records.append({380 "article_id": str(uuid.uuid4()),381 "title": title,382 "text": body,383 "source_domain": "unknown",384 "published_date": pd.NaT,385 "has_date": False,386 "binary_label": label,387 "dataset_origin": "multi_domain",388 })389 return records390 391 392def load_multi_domain(dataset_root: str) -> pd.DataFrame:393 """Load the Multi-Domain Fake News dataset (``overall/`` folder).394 395 Structure::396 397 overall/overall/398 fake/ → .txt files (label 0)399 real/ → .txt files (label 1)400 celebrityDataset/401 fake/ → .txt files (label 0)402 legit/ → .txt files (label 1)403 404 Args:405 dataset_root: Path to the top-level Dataset folder.406 407 Returns:408 DataFrame in the unified schema.409 """410 t0 = time.perf_counter()411 logger.info("Loading Multi-Domain dataset …")412 413 base = os.path.join(dataset_root, "overall", "overall")414 records: List[dict] = []415 416 # Main fake / real folders417 records.extend(_load_txt_folder(os.path.join(base, "fake"), label=0))418 records.extend(_load_txt_folder(os.path.join(base, "real"), label=1))419 420 # Celebrity sub-dataset421 celeb = os.path.join(base, "celebrityDataset")422 records.extend(_load_txt_folder(os.path.join(celeb, "fake"), label=0))423 records.extend(_load_txt_folder(os.path.join(celeb, "legit"), label=1))424 425 result = pd.DataFrame(records, columns=UNIFIED_COLUMNS)426 logger.info(427 "Multi-Domain loaded: %d rows (True=%d, Fake=%d) in %.1fs",428 len(result),429 (result["binary_label"] == 1).sum(),430 (result["binary_label"] == 0).sum(),431 time.perf_counter() - t0,432 )433 return result434 435 436# ─────────────────────────────────────────────────────────437 438def load_training_folder(dataset_root: str) -> pd.DataFrame:439 """Load supplementary training data from ``training/training/``.440 441 Structure mirrors multi-domain with sub-datasets ``celebrityDataset``442 and ``fakeNewsDataset``, each containing ``fake/`` and ``legit/`` folders.443 444 Args:445 dataset_root: Path to the top-level Dataset folder.446 447 Returns:448 DataFrame in the unified schema.449 """450 t0 = time.perf_counter()451 logger.info("Loading supplementary training folder …")452 453 base = os.path.join(dataset_root, "training", "training")454 records: List[dict] = []455 456 for subdir in ("celebrityDataset", "fakeNewsDataset"):457 sub_path = os.path.join(base, subdir)458 if not os.path.isdir(sub_path):459 continue460 fake_recs = _load_txt_folder(os.path.join(sub_path, "fake"), label=0)461 legit_recs = _load_txt_folder(os.path.join(sub_path, "legit"), label=1)462 for r in fake_recs + legit_recs:463 r["dataset_origin"] = f"training_{subdir}"464 records.extend(fake_recs + legit_recs)465 logger.info(" %s: %d fake + %d legit", subdir, len(fake_recs), len(legit_recs))466 467 result = pd.DataFrame(records, columns=UNIFIED_COLUMNS)468 logger.info(469 "Training folder loaded: %d rows (True=%d, Fake=%d) in %.1fs",470 len(result),471 (result["binary_label"] == 1).sum(),472 (result["binary_label"] == 0).sum(),473 time.perf_counter() - t0,474 )475 return result476 477 478# ─────────────────────────────────────────────────────────479 480def load_testing_dataset(dataset_root: str) -> pd.DataFrame:481 """Load the sacred hold-out Testing_dataset (never used for training).482 483 Structure::484 485 Testing_dataset/testingSet/486 fake/ → .txt files (label 0)487 real/ → .txt files (label 1)488 489 The catalog CSVs in this folder provide metadata; the actual article490 bodies live in the ``fake/`` and ``real/`` sub-folders.491 492 Args:493 dataset_root: Path to the top-level Dataset folder.494 495 Returns:496 DataFrame in the unified schema with ``dataset_origin = "testing"``.497 """498 t0 = time.perf_counter()499 logger.info("Loading Testing dataset (hold-out) …")500 501 base = os.path.join(dataset_root, "Testing_dataset", "testingSet")502 records: List[dict] = []503 504 fake_recs = _load_txt_folder(os.path.join(base, "fake"), label=0)505 real_recs = _load_txt_folder(os.path.join(base, "real"), label=1)506 for r in fake_recs + real_recs:507 r["dataset_origin"] = "testing"508 records.extend(fake_recs + real_recs)509 510 # Optionally enrich with catalog metadata511 for catalog_name, label in [("Catalog - Fake Articles.csv", 0), ("Catalog - Real Articles.csv", 1)]:512 cat_path = os.path.join(base, catalog_name)513 if os.path.exists(cat_path):514 try:515 cat = pd.read_csv(cat_path)516 logger.info(" Catalog %s: %d entries", catalog_name, len(cat))517 except Exception as exc:518 logger.warning(" Could not read catalog %s: %s", catalog_name, exc)519 520 result = pd.DataFrame(records, columns=UNIFIED_COLUMNS)521 logger.info(522 "Testing dataset loaded: %d rows (True=%d, Fake=%d) in %.1fs",523 len(result),524 (result["binary_label"] == 1).sum(),525 (result["binary_label"] == 0).sum(),526 time.perf_counter() - t0,527 )528 return result529 530 531# ═══════════════════════════════════════════════════════════532# Main ingestion pipeline533# ═══════════════════════════════════════════════════════════534 535def run_ingestion(cfg: dict) -> pd.DataFrame:536 """Execute the full Stage 1 ingestion pipeline.537 538 Steps:539 1. Load all five dataset sources.540 2. Concatenate into a single DataFrame.541 3. Run Sentence-BERT deduplication.542 4. Persist ``unified.csv`` and ``stats.json``.543 544 Args:545 cfg: Parsed config dictionary (from ``config.yaml``).546 547 Returns:548 The final unified (deduplicated) DataFrame.549 """550 pipeline_t0 = time.perf_counter()551 logger.info("═" * 60)552 logger.info(" STAGE 1 — INGESTION START")553 logger.info("═" * 60)554 555 dataset_root = os.path.abspath(556 os.path.join(str(_PROJECT_ROOT), cfg["paths"]["dataset_root"])557 )558 logger.info("Dataset root resolved to: %s", dataset_root)559 560 # ── Step 1 : Load each dataset ───────────────────────────561 t0 = time.perf_counter()562 df_isot = load_isot(dataset_root)563 df_liar = load_liar(dataset_root)564 df_kaggle = load_kaggle_combined(dataset_root)565 df_multi = load_multi_domain(dataset_root)566 df_training = load_training_folder(dataset_root)567 df_testing = load_testing_dataset(dataset_root)568 load_time = time.perf_counter() - t0569 logger.info("All datasets loaded in %.1fs", load_time)570 571 # ── Step 2 : Concatenate ─────────────────────────────────572 t0 = time.perf_counter()573 all_frames = [df_isot, df_liar, df_kaggle, df_multi, df_training, df_testing]574 df_unified = pd.concat(all_frames, ignore_index=True)575 logger.info(576 "Unified dataset: %d rows (concat took %.1fs)",577 len(df_unified), time.perf_counter() - t0,578 )579 580 # Log per-origin counts581 origin_counts = df_unified["dataset_origin"].value_counts()582 for origin, cnt in origin_counts.items():583 logger.info(" %-30s %6d rows", origin, cnt)584 585 # ── Prep ─────────────────────────────────────────────────586 # FIX 1: Exclude Sacred Hold-out from Dedup587 test_mask = df_unified["dataset_origin"] == "testing"588 test_df = df_unified.loc[test_mask].copy()589 train_pool_df = df_unified.loc[~test_mask].copy()590 591 # FIX 3: Remove empty/near-empty texts from training ONLY592 min_word_count = cfg.get("preprocessing", {}).get("min_word_count", 3)593 train_before = len(train_pool_df)594 train_pool_df = clean_empty_texts(train_pool_df, min_word_count=min_word_count)595 empty_dropped = train_before - len(train_pool_df)596 597 # Flag short texts in test_df instead of dropping them598 test_full = test_df.apply(lambda r: build_full_text(r.get("title", ""), r.get("text", "")), axis=1)599 test_df["short_text_flag"] = test_full.apply(word_count) < min_word_count600 short_test_flagged = int(test_df["short_text_flag"].sum())601 602 logger.info("Sacred test rows preserved: %d (flagged %d short texts)", len(test_df), short_test_flagged)603 604 # ── Step 3 : Deduplication ───────────────────────────────605 dedup_cfg = cfg.get("dataset", {})606 threshold = dedup_cfg.get("dedup_threshold", 0.92)607 batch_size = dedup_cfg.get("dedup_batch_size", 64)608 609 train_pool_df["_dedup_text"] = (610 train_pool_df["title"].fillna("") + " " + train_pool_df["text"].fillna("")611 ).str.strip()612 613 mask_has_text = train_pool_df["_dedup_text"].str.len() > 10614 df_with_text = train_pool_df.loc[mask_has_text].copy()615 df_no_text = train_pool_df.loc[~mask_has_text].copy()616 617 logger.info(618 "Dedup candidates (train pool): %d rows with text, %d skipped (too short)",619 len(df_with_text), len(df_no_text),620 )621 622 if len(df_with_text) > 0:623 exact_counts = len(df_with_text) - len(df_with_text.drop_duplicates(subset=["_dedup_text"]))624 df_deduped, dedup_stats = deduplicate_dataframe(625 df_with_text,626 text_column="_dedup_text",627 threshold=threshold,628 batch_size=batch_size,629 origin_column="dataset_origin",630 )631 total_removed = len(df_with_text) - len(df_deduped)632 semantic_counts = total_removed - exact_counts633 else:634 df_deduped = df_with_text635 dedup_stats = {}636 exact_counts = 0637 semantic_counts = 0638 639 train_pool_deduped = pd.concat([df_deduped, df_no_text], ignore_index=True)640 train_pool_deduped.drop(columns=["_dedup_text"], inplace=True, errors="ignore")641 642 # FIX 2: Stratified Holdout Carve-out643 holdout_cfg = cfg.get("holdout", {})644 stratified_test_size = holdout_cfg.get("stratified_test_size", 0.10)645 random_state = holdout_cfg.get("random_state", 42)646 647 from sklearn.model_selection import StratifiedShuffleSplit648 sss = StratifiedShuffleSplit(n_splits=1, test_size=stratified_test_size, random_state=random_state)649 650 train_pool_deduped = train_pool_deduped.reset_index(drop=True)651 train_idx, held_idx = next(sss.split(train_pool_deduped, train_pool_deduped['binary_label']))652 653 stratified_holdout = train_pool_deduped.iloc[held_idx].copy()654 train_pool_final = train_pool_deduped.iloc[train_idx].copy()655 656 stratified_holdout['dataset_origin'] = 'stratified_holdout'657 658 logger.info("Train pool after carve-out: %d", len(train_pool_final))659 logger.info("Stratified holdout: %d", len(stratified_holdout))660 logger.info("Sacred test set: %d", len(test_df))661 662 train_pool_final['short_text_flag'] = False663 stratified_holdout['short_text_flag'] = False664 665 df_final = pd.concat([666 train_pool_final,667 stratified_holdout,668 test_df669 ], ignore_index=True)670 671 logger.info("Post-dedup and split total: %d rows", len(df_final))672 673 # ── Step 4 : Ensure types ────────────────────────────────674 df_final["published_date"] = pd.to_datetime(675 df_final["published_date"], errors="coerce"676 )677 df_final["has_date"] = df_final["published_date"].notna()678 df_final["binary_label"] = df_final["binary_label"].astype(int)679 680 # ── Step 5 : Save unified CSV + stats ────────────────────681 processed_dir = os.path.join(str(_PROJECT_ROOT), cfg["paths"]["processed_dir"])682 os.makedirs(processed_dir, exist_ok=True)683 684 csv_path = os.path.join(processed_dir, "unified.csv")685 df_final.to_csv(csv_path, index=False)686 logger.info("Saved unified CSV → %s (%d rows)", csv_path, len(df_final))687 688 # Stats689 stats = {690 "total_rows": len(df_final),691 "train_pool_rows": len(train_pool_final),692 "stratified_holdout_rows": len(stratified_holdout),693 "sacred_test_rows": len(test_df),694 "fake_count": int((df_final["binary_label"] == 0).sum()),695 "true_count": int((df_final["binary_label"] == 1).sum()),696 "has_date_ratio": float(df_final["has_date"].mean()),697 "empty_texts_dropped": empty_dropped,698 "short_text_flagged_in_test": short_test_flagged,699 "dedup_removed_exact": exact_counts,700 "dedup_removed_semantic": semantic_counts,701 "per_origin": df_final["dataset_origin"].value_counts().to_dict(),702 "dedup_stats": {k: int(v) for k, v in dedup_stats.items()}703 }704 stats_path = os.path.join(processed_dir, "stats.json")705 with open(stats_path, "w", encoding="utf-8") as fh:706 json.dump(stats, fh, indent=2, default=str)707 logger.info("Saved stats → %s", stats_path)708 709 pipeline_elapsed = time.perf_counter() - pipeline_t0710 logger.info("═" * 60)711 logger.info(" STAGE 1 — INGESTION COMPLETE (%.1fs total)", pipeline_elapsed)712 logger.info("═" * 60)713 714 return df_final715 716 717# ═══════════════════════════════════════════════════════════718# __main__ block for standalone testing719# ═══════════════════════════════════════════════════════════720if __name__ == "__main__":721 cfg = load_config()722 df = run_ingestion(cfg)723 print("\n=== Final Unified Dataset ===")724 print(f"Shape: {df.shape}")725 print(f"\nLabel distribution:\n{df['binary_label'].value_counts()}")726 print(f"\nOrigin distribution:\n{df['dataset_origin'].value_counts()}")727 print(f"\nhas_date ratio: {df['has_date'].mean():.2%}")728 print(f"\nSample rows:\n{df.head(3).to_string()}")729