Elmanda1/Add New Engine

How to Add a New Engine

This skill documents the complete process for adding a new data source engine to DataPipeline OS, following the established generator pattern and pluggable architecture.

Prerequisites

• Understanding of the unified 5-column output schema: keyword, source, date, content, engagement_score
• The new data source must be accessible via API, RSS, or web scraping

Steps

1. Create the Engine File

Create engines/<source>_engine.py using this template:

"""
engines/<source>_engine.py

<Source Name> scraper.
- Generator pattern: chunk → yield → clear() — RAM flat O(chunk_size)
- Dual mode: set <ENV_VAR> in .env for live, blank = simulation
"""

import os
import time
import random
from typing import Generator, List


class <Source>Engine:

    SAFE_COLUMNS = {"keyword", "source", "date", "content", "engagement_score"}

    def __init__(
        self,
        keywords:   List[str],
        start_date: str,
        end_date:   str,
        language:   str = "id",
        chunk_size: int = 500,
    ):
        self.keywords   = keywords
        self.start_date = start_date
        self.end_date   = end_date
        self.language   = language
        self.chunk_size = chunk_size
        self.all_data:  List[dict] = []

    def fetch(self) -> Generator[List[dict], None, None]:
        api_key = os.getenv("<ENV_VAR>", "").strip()
        if api_key:
            yield from self._fetch_live(api_key)
        else:
            yield from self._fetch_simulation()

    def _fetch_live(self, api_key: str) -> Generator[List[dict], None, None]:
        buffer: List[dict] = []

        for keyword in self.keywords:
            # ... fetch data from API ...
            for item in api_results:
                buffer.append({
                    "keyword":          keyword,
                    "source":           "<source_name>",
                    "date":             "YYYY-MM-DD",
                    "content":          "text content here",
                    "engagement_score": 0,
                })
                if len(buffer) >= self.chunk_size:
                    self.all_data.extend(buffer)
                    yield buffer
                    buffer.clear()

            if buffer:
                self.all_data.extend(buffer)
                yield buffer
                buffer.clear()

    def _fetch_simulation(self) -> Generator[List[dict], None, None]:
        for keyword in self.keywords:
            for _ in range(random.randint(3, 7)):
                buffer: List[dict] = []
                size = random.randint(int(self.chunk_size * 0.5), self.chunk_size)
                for _ in range(size):
                    buffer.append({
                        "keyword":          keyword,
                        "source":           "<source_name>",
                        "date":             f"2023-{random.randint(1,12):02d}-{random.randint(1,28):02d}",
                        "content":          f"Simulated content about {keyword}",
                        "engagement_score": random.randint(0, 10000),
                    })
                self.all_data.extend(buffer)
                time.sleep(0.15)
                yield buffer
                buffer.clear()

2. Register in app.py

Add the engine to the ENGINE_REGISTRY list in app.py:

from engines.<source>_engine import <Source>Engine

ENGINE_REGISTRY = [
    # ... existing engines ...
    ("<source_name>", <Source>Engine, "<eng_key>", "<4CHAR>"),
]

And add the engine state in pipeline_state["engines"]:

"<eng_key>": {"status": "idle", "rows": 0, "ram_mb": 0},

3. Update Frontend

In frontend/templates/index.html:

• Add a checkbox in the platform selection form with value="<source_name>"
• Add an engine row in the monitor panel with corresponding data-engine="<eng_key>"

4. Update Environment

• Add the API key variable to .env.example with instructions
• Add any new pip dependencies to requirements.txt

5. Test

1. Run with simulation mode first (no API key)
2. Verify output CSV has correct 5-column schema
3. Set API key in .env and test live mode
4. Verify generator pattern works (RAM stays flat)

Critical Rules

• Always use generator pattern — yield buffer then buffer.clear()
• Always accumulate to self.all_data — needed by schema mapper
• Only output 5 columns — extra columns will be stripped by cleaner
• Include simulation fallback — engine must work without API keys
• Source name in output must be unique and recognizable