News pipeline

News pipeline#

The operator’s working tool for collecting, extracting, analyzing, and publishing open-source reporting on a topic, entity, or region. Five libraries cover the stages, scrapy, newspaper, nltk, pandas, and django.

This project assumes the operator has worked through Setup, Tooling, Syntax, Patterns, and Data Structures.

The pipeline#

The pipeline takes a list of seed URLs (a target’s site list, RSS feeds, sitemaps) and produces a searchable, scored corpus you can query.

        flowchart LR
    A["Seeds<br/>URLs / RSS / sitemaps"] --> B["Crawl<br/>scrapy"]
    B --> C["Extract<br/>newspaper"]
    C --> D["Analyze<br/>nltk"]
    D --> E["Tabulate<br/>pandas"]
    E --> F["Publish<br/>django"]

Each stage is a separate module you can run on its own. The stages communicate through files on disk (JSON-Lines, Parquet) or through a Django model, depending on scale and durability requirements.

Stage	Library	Operator concern
Crawl	`scrapy`	Discovery, rate-limit, robots, deduplication, retries.
Extract	`newspaper` (`newspaper3k`)	Strip nav and ads to clean article body, author, date.
Analyze	`nltk`	Tokenize, POS-tag, named-entity recognition, sentiment.
Tabulate	`pandas`	Filter, aggregate, time-window, score, dedup, export.
Publish	`django`	Search, dashboards, alerting, RBAC, audit log.

Project layout#

One repository with a package per stage and one Django project that consumes the output of the earlier stages.

news-pipeline/
├── pyproject.toml
├── uv.lock
├── README.md
│
├── src/
│   ├── crawl/                  # stage 1: scrapy project
│   │   ├── __init__.py
│   │   ├── settings.py
│   │   ├── pipelines.py
│   │   └── spiders/
│   │       └── articles.py
│   │
│   ├── extract/                # stage 2: newspaper
│   │   ├── __init__.py
│   │   └── extract.py
│   │
│   ├── analyze/                # stage 3: nltk
│   │   ├── __init__.py
│   │   ├── tokens.py
│   │   ├── entities.py
│   │   └── sentiment.py
│   │
│   ├── tabulate/               # stage 4: pandas
│   │   ├── __init__.py
│   │   └── frames.py
│   │
│   └── site/                   # stage 5: django project
│       ├── manage.py
│       ├── site/
│       │   ├── settings.py
│       │   └── urls.py
│       └── corpus/
│           ├── models.py
│           ├── views.py
│           └── admin.py
│
├── data/                       # local artifacts (gitignored)
│   ├── raw/                    # scrapy output, .jsonl
│   ├── extracted/              # newspaper output, .jsonl
│   ├── analyzed/               # nltk output, .jsonl
│   └── tabulated/              # pandas output, .parquet
│
└── tests/

Stage 1: Crawl#

scrapy is the operator’s standard for crawls beyond single-page fetches. It handles concurrency, retries, robots, duplicate filtering, and depth-limited spidering out of the box.

A minimal article spider that follows links on a domain and emits one JSON record per page.

# src/crawl/spiders/articles.py
import scrapy
from datetime import datetime
from urllib.parse import urlparse

class ArticleSpider(scrapy.Spider):
    name = "articles"
    custom_settings = {
        "DOWNLOAD_DELAY": 1.0,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 4,
        "ROBOTSTXT_OBEY": True,
        "USER_AGENT": "news-pipeline/0.1 (+ops@example.com)",
    }

    def __init__(self, seeds_file: str, **kw):
        super().__init__(**kw)
        with open(seeds_file) as f:
            self.start_urls = [line.strip() for line in f if line.strip()]
        self.allowed_domains = [
            urlparse(u).netloc for u in self.start_urls
        ]

    def parse(self, response):
        yield {
            "url": response.url,
            "fetched_at": datetime.utcnow().isoformat(),
            "status": response.status,
            "html": response.text,
        }
        for href in response.css("a::attr(href)").getall():
            yield response.follow(href, callback=self.parse)

$ uv run scrapy crawl articles \
    -a seeds_file=seeds.txt \
    -O data/raw/articles.jsonl

Stage 2: Extract#

newspaper3k strips boilerplate (nav, ads, sidebars) and returns clean article text plus metadata (title, authors, publish date, summary).

# src/extract/extract.py
import json, sys
from newspaper import Article

def extract(record: dict) -> dict | None:
    art = Article(record["url"])
    art.set_html(record["html"])
    try:
        art.parse()
    except Exception:
        return None
    return {
        "url": record["url"],
        "fetched_at": record["fetched_at"],
        "title": art.title,
        "authors": art.authors,
        "publish_date": (
            art.publish_date.isoformat() if art.publish_date else None
        ),
        "text": art.text,
    }

def main(in_path: str, out_path: str) -> None:
    with open(in_path) as src, open(out_path, "w") as dst:
        for line in src:
            record = json.loads(line)
            extracted = extract(record)
            if extracted and extracted["text"]:
                dst.write(json.dumps(extracted) + "\n")

if __name__ == "__main__":
    main(*sys.argv[1:])

$ uv run python -m extract.extract \
    data/raw/articles.jsonl \
    data/extracted/articles.jsonl

Stage 3: Analyze#

nltk handles tokenization, POS tagging, named-entity recognition, and (with VADER) sentiment.

$ uv run python -c "
import nltk
for pkg in ('punkt', 'averaged_perceptron_tagger',
            'maxent_ne_chunker', 'words', 'vader_lexicon'):
    nltk.download(pkg)
"

# src/analyze/entities.py
from nltk import word_tokenize, pos_tag, ne_chunk
from nltk.tree import Tree

def entities(text: str) -> list[tuple[str, str]]:
    chunks = ne_chunk(pos_tag(word_tokenize(text)))
    out = []
    for node in chunks:
        if isinstance(node, Tree):
            out.append((" ".join(w for w, _ in node.leaves()), node.label()))
    return out

# src/analyze/sentiment.py
from nltk.sentiment.vader import SentimentIntensityAnalyzer
_sia = SentimentIntensityAnalyzer()

def sentiment(text: str) -> dict:
    return _sia.polarity_scores(text)

Stage 4: Tabulate#

pandas collapses the per-article JSONL into a single tabular corpus you can filter, group, and aggregate.

# src/tabulate/frames.py
import json
import pandas as pd

def load(jsonl_path: str) -> pd.DataFrame:
    rows = []
    with open(jsonl_path) as f:
        for line in f:
            r = json.loads(line)
            rows.append({
                "url": r["url"],
                "fetched_at": pd.to_datetime(r["fetched_at"]),
                "publish_date": pd.to_datetime(r.get("publish_date")),
                "title": r["title"],
                "text_len": len(r["text"]),
                "sentiment": r["sentiment"]["compound"],
                "entity_orgs": [
                    e for e, lbl in r["entities"] if lbl == "ORGANIZATION"
                ],
                "entity_persons": [
                    e for e, lbl in r["entities"] if lbl == "PERSON"
                ],
            })
    return pd.DataFrame(rows)

Stage 5: Publish#

django serves the corpus to the operator and their team. The full Django architecture lives in Django site; here, the minimal model that consumes the tabulated corpus.

# src/site/corpus/models.py
from django.db import models

class Article(models.Model):
    url           = models.URLField(max_length=2048, unique=True)
    title         = models.CharField(max_length=512)
    text          = models.TextField()
    fetched_at    = models.DateTimeField()
    publish_date  = models.DateTimeField(null=True, blank=True)
    sentiment     = models.FloatField()
    organizations = models.JSONField(default=list)
    persons       = models.JSONField(default=list)

    class Meta:
        indexes = [
            models.Index(fields=["publish_date"]),
            models.Index(fields=["sentiment"]),
        ]

Common Tasks#

Run the full pipeline against a seeds file.

$ uv run scrapy crawl articles -a seeds_file=seeds.txt \
    -O data/raw/articles.jsonl
$ uv run python -m extract.extract \
    data/raw/articles.jsonl data/extracted/articles.jsonl
$ uv run python -m analyze \
    data/extracted/articles.jsonl data/analyzed/articles.jsonl
$ uv run python -c "from tabulate.frames import load; \
    load('data/analyzed/articles.jsonl') \
        .to_parquet('data/tabulated/corpus.parquet')"

Query the corpus for an entity, last 30 days.

import pandas as pd
df = pd.read_parquet("data/tabulated/corpus.parquet")
cutoff = pd.Timestamp.utcnow() - pd.Timedelta(days=30)
recent = df[df["publish_date"] >= cutoff]
recent[recent["entity_orgs"].apply(lambda xs: "Anthropic" in xs)]

Resume a crawl after a failure.

$ uv run scrapy crawl articles -a seeds_file=seeds.txt \
    -s JOBDIR=data/crawls/articles-1 \
    -O data/raw/articles.jsonl

Schedule the pipeline daily.

$ crontab -e
# 0 3 * * * cd /home/operator/news-pipeline && ./run_pipeline.sh

References#

Django site for the Django publishing layer end-to-end.
Libraries for the libraries this project leans on.
Frameworks for Django’s architecture and full surface.
Networking for the protocols the crawl works over.
Scrapy documentation
Newspaper3k documentation
NLTK documentation
pandas documentation