Data Processing Pipeline

Relevant source files

The following files were used as context for generating this wiki page:

• monitor/main.py
• monitor/config.py
• monitor/extractor/article_parser.py
• monitor/models/article.py
• monitor/feeds/base.py
• WARP.md
• test_basic.py
• README.md

The Data Processing Pipeline is the core engine of the Technical Blog Monitor. It is an asynchronous, multi-stage system designed to discover new blog posts from configured feeds, fetch and render their content, extract meaningful text and metadata, generate vector embeddings, and store the results in a vector database for semantic search. The pipeline is highly modular, allowing for different components (e.g., embedding models, vector databases) to be swapped via configuration.

The entire process is orchestrated by the main application entry point, which can run either as a continuous daemon or as a one-time process. It manages the lifecycle of all necessary components, including browser pools, cache clients, and database connections, ensuring graceful startup and shutdown.

Sources: README.md, WARP.md, monitor/main.py:382-414

Architecture Overview

The pipeline follows a linear flow, transforming a URL from a feed into a searchable, embedded record in a vector database. Each stage is handled by a specialized component, with data flowing from one to the next.

This diagram illustrates the high-level stages of the data processing pipeline.

Pipeline Orchestration

The pipeline is managed by the AppContext class and the app_lifecycle context manager. This central orchestrator handles the initialization, dependency injection, and graceful shutdown of all services.

Sources: monitor/main.py:32-238

Application Context (AppContext)

The AppContext class holds all initialized components required for the pipeline to run. On startup, it initializes the following services in order:

1. Cache Client: Based on CacheConfig, can be Redis or in-memory.
2. Browser Pool: Manages a pool of Playwright browser instances for concurrent rendering.
3. Embedding Client: A pluggable client for generating embeddings (e.g., OpenAI, Ollama, HuggingFace).
4. Vector DB Client: A pluggable client for a vector database (e.g., Qdrant, Chroma, PGVector).
5. Scheduler: An AsyncIOScheduler instance that triggers feed processing jobs at configured intervals.

The AppContext uses an AsyncExitStack to ensure all these resources are cleaned up properly on shutdown.

Sources: monitor/main.py:32-187

Execution Flow

The main execution flow is initiated in monitor/main.py. The application can run in two modes:

• Daemon Mode (run_daemon): Runs continuously, relying on the APScheduler to trigger process_feed jobs based on each feed's check_interval_minutes.
• Once Mode (run_once): Iterates through all enabled feeds, processes them once, and then exits.

The core processing logic for a single feed is encapsulated in the process_feed function, which is called by the scheduler or the run_once loop.

This sequence diagram shows the interactions between the main components during a single feed processing job.

Pipeline Stages

1. Feed Discovery

This is the first stage of the pipeline, where the system checks a given feed URL for new posts.

• Function: process_feed_posts
• Logic:
  1. A FeedProcessor (e.g., RSSFeedProcessor) is selected based on the feed URL and content.
  2. The feed is fetched and parsed to extract a list of entries.
  3. Each entry's unique ID is checked against the cache to see if it has been processed before.
  4. Entries that are not found in the cache are considered new and are converted into BlogPost objects.
• Configuration: FeedConfig in monitor/config.py defines the URL, check interval, and other parameters for each feed.

Sources: monitor/feeds/base.py, monitor/config.py:28-35

2. Content Extraction

Once a new post is identified, its full content is fetched and parsed to extract structured data. This stage is controlled by the full_content_capture setting.

• Function: extract_article_content
• Logic:
  1. The function first checks a cache for the ArticleContent object to avoid re-processing.
  2. If not cached, the article URL is fetched. The system can use the BrowserPool to render the page fully, executing JavaScript and capturing a screenshot.
  3. The raw HTML is passed to readability-lxml to extract the main article body, title, and summary.
  4. BeautifulSoup is used for more granular parsing to extract metadata like author, publication date, and tags. It looks for common HTML patterns, meta tags (Open Graph, Twitter Cards), and JSON-LD scripts.
  5. The extracted HTML is cleaned to produce plain text, and a word count is calculated.
  6. All extracted data is compiled into an ArticleContent object.

Sources: monitor/extractor/article_parser.py:70-202, monitor/main.py:307-312

ArticleContent Data Model

The ArticleContent model is a Pydantic class that serves as the structured representation of a processed article.

Field	Type	Description
url	HttpUrl	The canonical URL of the article.
title	str	The main title of the article.
text	str	The cleaned, plain text content.
html	str	The cleaned HTML of the main article body.
author	Optional[str]	The extracted author's name.
publish_date	Optional[datetime]	The publication date of the article.
word_count	int	The total number of words in the text.
tags	List[str]	A list of tags or keywords.
screenshot_paths	List[str]	Filesystem paths to captured screenshots.
metadata	Dict[str, Any]	A dictionary for any other extracted metadata.

Sources: monitor/models/article.py:16-43

3. Embedding Generation

After content extraction, the text and optional screenshot are converted into numerical vector representations (embeddings).

• Function: process_post calls embedding_client.embed_text() and embed_image().
• Logic:
  1. The text from the ArticleContent object is passed to the configured text embedding model.
  2. If an image model is configured and a screenshot was taken, the screenshot is passed to the image embedding model.
  3. The system uses a pluggable EmbeddingClient which supports various providers like OpenAI, HuggingFace, and Ollama.
• Configuration: EmbeddingConfig in monitor/config.py controls the model types, names, API keys, and dimensions.

Sources: monitor/main.py:347-354, monitor/config.py:105-143

4. Vector Storage

The final stage is to store the post's data and its newly generated embeddings in a vector database.

• Function: vector_db_client.upsert()
• Logic:
  1. An EmbeddingRecord is created, which contains the post's ID, URL, title, metadata, and the text/image embeddings.
  2. This record is "upserted" into the vector database. The upsert operation will either create a new entry or update an existing one based on the record's ID.
• Configuration: VectorDBConfig in monitor/config.py specifies the database type (Qdrant, Chroma, etc.), connection string, and collection name.

Sources: monitor/main.py:372-374, monitor/config.py:155-181

Configuration

The pipeline's behavior is heavily controlled by the Settings model, which is populated from environment variables or a .env file. Key configuration sections for the pipeline are:

Section	Pydantic Model	Purpose	Source File
Feeds	FeedConfig	Defines the blog sources to monitor.	monitor/config.py:28-35
Browser	BrowserConfig	Configures the headless browser for rendering.	monitor/config.py:38-60
Article Processing	ArticleProcessingConfig	Controls content capture, concurrency, and archival.	monitor/config.py:214-266
Embeddings	EmbeddingConfig	Selects embedding models and API keys.	monitor/config.py:105-143
Vector DB	VectorDBConfig	Configures the connection to the vector database.	monitor/config.py:155-181
Cache	CacheConfig	Configures the caching layer (Redis or filesystem).	monitor/config.py:63-75

Sources: monitor/config.py

Summary

The Data Processing Pipeline is a robust and extensible system for ingesting content from technical blogs. It leverages an asynchronous architecture to efficiently handle I/O-bound tasks like fetching and CPU-bound tasks like parsing. Its modular design, driven by Pydantic configuration, allows developers to easily adapt the pipeline for different environments, data sources, and machine learning models without changing the core orchestration logic. The clear separation of concerns between stages—discovery, extraction, embedding, and storage—makes the system maintainable and scalable.