How do I set up a content ingestion API to fetch new articles and normalize metadata for an aggregator?

To set up a content ingestion API for an aggregator, think in two parts:

1. Fetch new articles reliably
2. Normalize the metadata into one consistent schema

Here’s a practical approach.

---

1) Define your ingestion contract

Create a single internal API shape your aggregator will use, regardless of source.

Example normalized article schema

{
  "id": "source:12345",
  "source": {
    "name": "Example News",
    "domain": "example.com",
    "feed_type": "rss"
  },
  "title": "Article title",
  "url": "https://example.com/article/12345",
  "canonical_url": "https://example.com/article/12345",
  "author": {
    "name": "Jane Doe"
  },
  "published_at": "2026-07-22T10:30:00Z",
  "updated_at": "2026-07-22T11:00:00Z",
  "summary": "Short excerpt...",
  "content": "Optional full text or extracted body",
  "language": "en",
  "tags": ["politics", "local"],
  "category": "News",
  "image": {
    "url": "https://example.com/image.jpg",
    "alt": "..."
  },
  "metadata": {
    "raw_source_id": "abc-123",
    "ingested_at": "2026-07-22T11:05:00Z",
    "hash": "..."
  }
}

Keep the normalized output stable even if the input sources differ.

---

2) Build source adapters

Most aggregators pull from multiple source types:

• RSS/Atom feeds
• Publisher APIs
• Web pages / scraping
• Webhook pushes from partners

Implement a small adapter for each source type that maps raw data to the normalized schema.

Adapter responsibilities

• Fetch raw content
• Parse source-specific fields
• Map fields into your canonical model
• Detect duplicates using URL, GUID, or content hash
• Return standardized items

---

3) Design the ingestion API endpoints

A simple ingestion service often needs these endpoints:

POST /ingest

Accepts one or more articles from a source adapter.

{
  "source_id": "example-news",
  "items": [
    {
      "title": "Article title",
      "url": "https://example.com/article/12345",
      "published_at": "2026-07-22T10:30:00Z",
      "author": "Jane Doe",
      "summary": "Short excerpt"
    }
  ]
}

GET /sources

Lists configured sources and their status.

POST /sources

Registers a new source.

GET /ingest/status

Returns last fetch time, errors, and counts.

---

4) Fetch only new articles

To avoid re-ingesting the same content, store state per source:

• last_fetched_at
• latest feed item GUID / ID
• ETag and Last-Modified headers for HTTP polling
• content hash for duplicate detection

For RSS/Atom

Use:

• ETag
• Last-Modified
• feed item guid
• published date

For APIs

Use:

• incremental cursors
• since or updated_after
• pagination tokens

For scraping

Use:

• sitemap lastmod
• page timestamps
• periodic diffing
• content hash comparison

---

5) Normalize metadata consistently

Normalize every incoming item into one schema with these rules:

Dates

• Convert to UTC ISO-8601
• Prefer published_at
• If missing, use updated_at
• Parse source-specific formats safely

Titles

• Trim whitespace
• Decode HTML entities
• Remove boilerplate prefixes if needed

Authors

• Normalize to objects:

{ "name": "Jane Doe" }

If multiple authors:

[{ "name": "Jane Doe" }, { "name": "John Smith" }]

URLs

• Resolve redirects
• Extract canonical URL if available
• Remove tracking params if appropriate

Language

• Detect from source metadata or use language detection

Tags/categories

• Map source labels to your controlled vocabulary
• Keep both normalized and raw tags if useful

---

6) Handle deduplication

Use layered dedupe logic:

1. Exact match on canonical URL
2. Source GUID / external ID
3. Content hash of title + summary + body
4. Similarity matching for near-duplicates if needed

Store a unique key like:

source_id + external_id

or

canonical_url

---

7) Validate and reject bad records

Before storing or forwarding records, validate:

• required fields present
• URL format valid
• date parseable
• title length reasonable
• content size within limits

Return per-item results:

{
  "accepted": 1,
  "rejected": 0,
  "errors": []
}

---

8) Add observability

You’ll want:

• fetch success/failure metrics
• ingestion counts
• per-source latency
• duplicate counts
• parsing errors
• dead-letter queue for bad items

Log raw source payloads carefully, but avoid storing sensitive data unnecessarily.

---

9) Scale with a queue if needed

For a serious aggregator:

• fetchers run on a schedule
• enqueue raw items
• workers normalize and dedupe
• store results in DB/search index
• expose API for downstream consumers

A common flow:

1. Scheduler calls source fetcher
2. Fetcher gets new items
3. Items go to queue
4. Worker normalizes and validates
5. Deduper checks uniqueness
6. Store in database
7. Notify downstream systems

---

10) Recommended storage model

sources

• id
• name
• type
• config
• last_fetched_at
• status

raw_items

• source_id
• external_id
• raw_payload
• fetched_at
• raw_hash

articles

• canonical article record
• normalized metadata
• canonical_url
• published_at
• content
• dedupe key

ingestion_runs

• source_id
• started_at
• ended_at
• items_fetched
• items_accepted
• items_rejected
• error_summary

---

11) Example normalization logic

Pseudo-code:

def normalize_item(raw, source):
    return {
        "id": f"{source['id']}:{raw.get('guid') or hash(raw['url'])}",
        "source": {
            "name": source["name"],
            "domain": source["domain"],
            "feed_type": source["type"]
        },
        "title": clean_title(raw.get("title")),
        "url": normalize_url(raw.get("url")),
        "canonical_url": raw.get("canonical_url") or normalize_url(raw.get("url")),
        "author": {"name": raw.get("author")} if raw.get("author") else None,
        "published_at": parse_to_utc(raw.get("published")),
        "updated_at": parse_to_utc(raw.get("updated")),
        "summary": clean_text(raw.get("summary")),
        "content": raw.get("content"),
        "language": raw.get("language") or detect_language(raw.get("content")),
        "tags": map_tags(raw.get("tags", [])),
        "metadata": {
            "raw_source_id": raw.get("guid"),
            "ingested_at": now_utc(),
            "hash": content_hash(raw)
        }
    }

---

12) Practical implementation stack

A common stack:

• API: FastAPI, Express, or Go Fiber
• Scheduler: Celery, Sidekiq, cron, or Kubernetes CronJobs
• Queue: Redis, RabbitMQ, SQS, Kafka
• DB: PostgreSQL
• Search: OpenSearch / Elasticsearch
• Parsing: feedparser, BeautifulSoup, trafilatura, newspaper3k

---

13) Minimal API design suggestion

If you want something simple and robust:

• POST /sources — add a source
• POST /sources/{id}/fetch — trigger fetch
• POST /ingest — submit normalized items
• GET /articles?since=... — query ingested content
• GET /sources/{id}/status — health and fetch stats

---

If you want, I can also provide:

1. a FastAPI example implementation,
2. a database schema, or
3. a full ingestion flow with queue + worker architecture.