TruthTrace: Israel Narrative Monitor

TruthTrace has one purpose: track how different news websites frame Israel over time.

The pipeline supports 3 languages (en, he, ar) and produces:

• Israel relevance detection
• framing classification (security, diplomacy, humanitarian, domestic, legal, economy, mixed)
• tone classification (supportive, neutral, critical)
• story-level cross-source comparison and daily trends
• story divergence scoring (low / medium / high)

Data Engineering Highlights

• Robust ETL pipeline: idempotent RSS ingestion + analysis upserts.
• Pipeline observability:
  • run log table: pipeline_runs
  • per-source run metrics: source_ingest_metrics
  • anomaly events: pipeline_anomalies
  • feed diagnostics (feed_ok, feed_status, parse warnings, fetch latency)
• Anomaly detection:
  • source volume drops/spikes
  • no-new-items detection
  • consecutive feed failures / stale-source streaks
  • relevance ratio drops/spikes
• Containerized runtime:
  • Docker Compose local stack
  • continuous collector service
• Orchestration path (optional but real):
  • Airflow scheduler/web UI
  • task DAGs for ingestion, health checks, and daily ML prep
• CI/CD checks:
  • lint, tests, compose validation, image builds

Architecture

• ingestor/main.py: RSS ingestion + Israel analysis + run/anomaly tracking
• api/main.py: dashboard + operations APIs
• api/static/: monitoring UI
• db/init.sql: single source-of-truth schema (used by Postgres init and ingestor schema ensure step)
• sources/sources.yaml: expanded source registry in 3 languages
• airflow/dags/: orchestrated pipelines (optional)

Flow (Runtime)

flowchart LR
  A[sources/sources.yaml RSS feeds] --> B[Ingestor: ingestor/main.py]
  B --> C[(PostgreSQL)]
  C --> D[API: /dashboard + /ops/*]
  D --> E[UI: api/static/*]
  B --> F[pipeline_runs]
  B --> G[source_ingest_metrics]
  B --> H[pipeline_anomalies]

Loading

Flow (Airflow Optional)

flowchart LR
  I[truthtrace_ingest_15m DAG] --> J[ingest_rss]
  J --> K[check_pipeline_health]
  J --> L[check_source_health]

  M[truthtrace_ml_daily_prep DAG] --> N[benchmark_export]
  N --> O[backfill_tone]

Loading

What Is Core Vs Optional

Core runtime path (used continuously):

• collector + ingestor/main.py
• api/main.py GET /dashboard
• api/static/* dashboard UI

Optional but intentional modules:

• ingestor/ml/* and scripts/run_local_ml_pipeline.sh (training/benchmark workflow)
• Ops-only scripts: scripts/check_pipeline_health.sh, scripts/check_source_health.sh
• Ops API endpoints (/ops/summary, /ops/source-health) used by monitoring scripts and automation
• Airflow DAGs and service (airflow/*) for scheduled orchestration

Quick Start

1. Start services and build images:

docker compose up -d postgres
docker compose build api ingestor

2. Start continuous collection and API:

docker compose up -d collector api

3. Open:

• http://127.0.0.1:8000

Collector Tuning

Default collector cadence is every 15 minutes (900 seconds).

Override interval and per-source fetch size when starting collector:

INGEST_INTERVAL_SECONDS=600 INGEST_MAX_ENTRIES=60 docker compose up -d collector

Useful commands:

docker compose logs -f collector
docker compose run --rm ingestor python -u /app/ingestor/main.py --max-entries-per-source 60

Airflow Orchestration (Optional)

Run Airflow locally (real scheduler + web UI):

docker compose build airflow
docker compose up -d airflow

Open:

• http://127.0.0.1:8080
• default local login: admin / admin

Included DAGs:

• truthtrace_ingest_15m:
  • runs ingestion every 15 minutes
  • then runs pipeline/source health checks
• truthtrace_ml_daily_prep:
  • daily benchmark export
  • tone backfill refresh

Useful commands:

docker compose exec airflow airflow dags list
docker compose exec airflow airflow tasks list truthtrace_ingest_15m
docker compose exec airflow airflow dags trigger truthtrace_ml_daily_prep

API Endpoints

Core analytics:

• GET /health
• GET /dashboard?days=7&lang=all|en|he|ar

Ops/monitoring:

• GET /ops/summary?days=7&run_limit=12&anomaly_limit=25
• GET /ops/source-health?days=14&runs_per_source=12&limit=120

Linux Ops Script

Quick pipeline health check:

./scripts/check_pipeline_health.sh
# or
./scripts/check_pipeline_health.sh http://127.0.0.1:8000

Strict mode (exit non-zero if SLA-like thresholds are violated):

STRICT=1 MIN_SUCCESS_RATE=80 MAX_AVG_ANOMALIES=10 ./scripts/check_pipeline_health.sh

Source health check (find down/stale/low-yield feeds):

./scripts/check_source_health.sh
# strict gate
STRICT=1 MAX_UNHEALTHY_PCT=40 MAX_DOWN_SOURCES=3 ./scripts/check_source_health.sh

Notes

• This is Israel-only by design.
• Historical comparisons come from continuous ingestion over time.
• Source list is editable in sources/sources.yaml.

Local ML Upgrade (No API Needed)

You can bootstrap a stronger local stance model pipeline without paid APIs.

1. Export benchmark records and baseline report:

docker compose run --rm ingestor python -m ingestor.ml.benchmark --days 45 --limit 30000 --out-dir /app/ingestor/ml_artifacts

2. Run local labeling (ML deps are included in the ingestor image):

docker compose run --rm ingestor python -m ingestor.ml.auto_label_local --input /app/ingestor/ml_artifacts/tone_records.jsonl --out-dir /app/ingestor/ml_artifacts

requirements-ml.txt uses CPU Torch wheels and is installed in truthtrace/ingestor:local.

3. Build balanced silver dataset for training:

docker compose run --rm ingestor python -m ingestor.ml.build_silver --records /app/ingestor/ml_artifacts/tone_records.jsonl --labels /app/ingestor/ml_artifacts/tone_ml_labels.jsonl --out-dir /app/ingestor/ml_artifacts/silver

The silver builder auto-bootstraps neutral examples and caps neutral volume per language so the model does not collapse into polarized-only predictions.

4. Train a local multilingual classifier:

docker compose run --rm ingestor python -m ingestor.ml.train_tone_model --data-dir /app/ingestor/ml_artifacts/silver --out-dir /app/ingestor/ml_artifacts/models/tone_xlmr_v2 --epochs 6

5. Backfill recent rows with model+rule fusion:

docker compose run --rm ingestor python -m ingestor.ml.backfill_tone --days 30 --min-confidence 0.76 --min-margin 0.18

Or run all steps in one command:

./scripts/run_local_ml_pipeline.sh
# include training:
TRAIN_MODEL=1 TRAIN_DATASET=silver ./scripts/run_local_ml_pipeline.sh
TRAIN_MODEL=1 TRAIN_DATASET=silver TRAIN_OUT_DIR=models/tone_xlmr_v2 ./scripts/run_local_ml_pipeline.sh

Outputs:

• ingestor/ml_artifacts/benchmark_report.json
• ingestor/ml_artifacts/audit_low_conf_samples.jsonl
• ingestor/ml_artifacts/tone_ml_labels.jsonl
• ingestor/ml_artifacts/silver/train.jsonl + val.jsonl + test.jsonl

Runtime inference loads TONE_MODEL_DIR when set, otherwise prefers:

1. ingestor/ml_artifacts/models/tone_xlmr_v2
2. ingestor/ml_artifacts/models/tone_xlmr