Keyword-driven public opinion and sentiment tracker for Indonesian content, built as a data engineering portfolio project.
Voice is a data engineering system that answers a simple question: what is the public saying about a topic right now?
Enter a keyword (e.g. makan bergizi gratis, korupsi, demo mahasiswa) and Voice:
- Scrapes news articles and social media comments from 6 sources (no login, no API keys for most sources)
- Cleans raw HTML/JSON into structured rows in a data warehouse
- Classifies sentiment using a hybrid local AI approach (IndoBERT + Gemini fallback)
- Serves the results through a REST API and an interactive dashboard
The pipeline is fully decoupled. Each stage runs independently, connected by a message queue. All raw data is preserved so re-processing is always possible without re-scraping.
Screenshots below show real data. Filter set to positive sentiment in Feed view.
Overview
Sentiment Trend Over Time
Sources Breakdown
Top Words by Sentiment
Comment Feed (Positive Filter)
flowchart LR
A([REST API\nor Dashboard]) -->|keyword + date_range| B[Redpanda / Kafka]
B --> C[Scraper Worker]
C -->|raw HTML / JSON| D[(MongoDB\nBronze Layer)]
D --> E[Cleaner]
E -->|structured rows| F[(PostgreSQL\nGold Layer)]
F --> G[AI Processor\nIndoBERT + Gemini]
G -->|sentiment labels| F
F --> H[FastAPI]
H --> A
V[(Valkey\nCache)] -. dedup by URL hash .-> C
V -. checkpoint cursor .-> E
Data flow in 6 steps:
- A keyword job is published to Redpanda (Kafka-compatible), triggered via REST API
- The Scraper Worker consumes the job and dispatches it to all registered source plugins concurrently
- Each plugin fetches articles and comments; a Valkey cache keyed on URL MD5 prevents re-scraping the same content across jobs
- Raw HTML/JSON is stored as-is in MongoDB (bronze layer, insert-only, never mutated)
- The Cleaner reads MongoDB, parses raw payloads per source, and writes structured rows to PostgreSQL (gold layer)
- The AI Processor reads unclassified comments, runs IndoBERT locally for bulk classification, and falls back to Gemini API only for low-confidence or stance-ambiguous cases
| Source | Content | Comments | Method |
|---|---|---|---|
| Detik | ✅ Articles | ✅ (GraphQL API) | SSR scraping + GraphQL |
| CNN Indonesia | ✅ Articles | ✅ (GraphQL API) | JSON search API + GraphQL |
| Liputan6 | ✅ Articles | No | Tag page SSR |
| Merdeka | ✅ Articles | No | Google CSE API (JSONP) |
| YouTube | ✅ Videos | ✅ Comments | Piped (self-hosted, no key) |
| ✅ Posts | ✅ (PullPush archive) | Public RSS + PullPush API |
No login or paid API keys required for any source.
Each tool was chosen deliberately, not to collect buzzwords, but to fit the actual problem.
Raw HTML and JSON vary wildly between sources. A Detik article body is structured HTML while a Reddit post is a JSON blob with nested fields. A document store handles schema-per-source naturally. More importantly, keeping raw data immutable means you can always re-parse after fixing a bug in the cleaner, with no re-scraping needed.
Why not MinIO (object storage)? Evaluated and rejected. MongoDB makes it easy to query and inspect raw documents during development (useful when debugging a parser). MinIO would require downloading files to inspect them.
The cleaned, structured data fits a relational model well: comments have a fixed schema, keywords and sources are discrete dimensions. PostgreSQL handles both the OLTP queries the API needs (filtering, pagination) and the simple aggregations the dashboard uses (sentiment percentages, word counts) without needing a separate analytics database.
Why not ClickHouse? Planned for a later phase. PostgreSQL is more than sufficient for MVP scale and avoids adding infrastructure complexity before it's needed.
Decoupling the scraper from the API trigger means the pipeline is async and observable. You can inspect the queue, replay messages, and add new consumers without touching the producer. Redpanda runs as a single Docker container vs Kafka's multi-service setup, making local development much lighter, while remaining fully Kafka-API compatible for production upgrade.
Running a local model (IndoBERT via transformers) handles the bulk of classification. It is free, has no rate limit, and works offline. Gemini is invoked only for comments where the local model's confidence is below 75%, or where the text contains explicit stance markers (pro/kontra). This keeps cost at $0 for normal runs while improving accuracy on the hardest cases.
Why not Gemini-only? Rate limits (10 RPM free tier) would make classifying 5,000+ comments impractical. Why not local-only? Gemini meaningfully improves accuracy on ambiguous Indonesian slang and stance detection.
Two responsibilities: (1) dedup by URL MD5, which prevents re-scraping the same article across keyword runs; (2) checkpoint cursor for the cleaner, letting it resume from where it left off after a crash or restart. Valkey is a Redis fork with no license restrictions, drop-in compatible with redis-py.
YouTube's official Data API has strict quota limits. Piped proxies YouTube's internal APIs with no account required, returning video metadata, upload dates, and comments. Running it self-hosted means no shared quota and no rate limits imposed by a third-party instance.
Raw-first pipeline: raw data is always preserved in MongoDB before any transformation. Cleaning and AI are separate idempotent passes that can be re-run without re-fetching. This is the data lakehouse pattern at small scale.
Plugin-based source registry: adding a new source is one new file with no changes to the pipeline. Each source inherits BaseSource and registers itself via a decorator. The worker discovers and runs all registered plugins automatically.
Broker flexibility: switch between Redpanda and Kafka by changing BROKER_TYPE in .env. The same kafka-python client works with both.
date_range="all": scraping and querying support "all" as a special value meaning no date restriction. Useful for initial data collection or historical analysis.
voice/
├── api/
│ ├── main.py # FastAPI app + CORS
│ ├── utils.py # Shared query helpers
│ └── routers/
│ ├── opinions.py # /summary, /trend, /trend-detail, /sources, /words, /feed, /prokontra
│ └── jobs.py # /jobs/trigger (REST pipeline trigger)
├── worker/
│ └── main.py # Kafka consumer, scraper pipeline
├── processor/
│ ├── main.py # Orchestrator: cleaner + AI loop (every 30s)
│ ├── cleaner.py # MongoDB, parse per source, PostgreSQL
│ └── ai.py # IndoBERT local + Gemini fallback classifier
├── sources/
│ ├── __init__.py # BaseSource abstract class
│ ├── registry.py # @register_source decorator + plugin registry
│ ├── news/
│ │ ├── detik.py # SSR scraping + GraphQL comments
│ │ ├── cnn.py # JSON search API + GraphQL comments
│ │ ├── liputan6.py # Tag page SSR scraping
│ │ └── merdeka.py # Google CSE API (JSONP)
│ └── socmed/
│ ├── youtube.py # Piped API (search + comments)
│ └── reddit.py # Public RSS + PullPush comment archive
├── shared/
│ ├── config.py # pydantic-settings, all config from .env
│ ├── kafka/ # Producer + Consumer wrappers
│ ├── mongodb/ # Connection + insert/find helpers
│ ├── postgresql/ # ThreadedConnectionPool + DDL + typed helpers
│ ├── redys/ # Valkey client: URL cache, checkpoint, job lock
│ └── utils/
│ ├── logger.py # Loguru + Rich, two-sink (console + rotating file)
│ ├── network.py # Async HTTP with retry + backoff
│ ├── datetime.py # Interval parsing (7d / 30d / 90d / all)
│ └── endecode.py # MD5, URL encode, base64 utilities
├── dashboard/ # React + Vite + Tailwind + Recharts
├── bot/ # Telegram bot (coming soon)
├── docker-compose.yml # Infra: MongoDB, PostgreSQL, Valkey, Redpanda/Kafka, Piped
├── requirements.txt
└── .env.example
-
Docker Desktop for all infrastructure services
-
Python 3.11+ for the pipeline services
-
Node.js 18+ for the dashboard (optional)
-
Gemini API key — optional, pipeline works with IndoBERT only. To get one:
- Go to aistudio.google.com and sign in with a Google account
- Click Get API key in the left sidebar
- Click Create API key, select a project (or create one), and copy the key
- Paste it into
GEMINI_API_KEYin your.envfile and setUSE_GEMINI=true
The free tier includes 1,500 requests/day and 10 requests/minute, which is enough for fallback use.
git clone https://github.com/yourusername/voice.git
cd voice
cp .env.example .envEdit .env with your values:
| Variable | Description | Default |
|---|---|---|
BROKER_TYPE |
redpanda or kafka |
redpanda |
BROKER_ADDRESS |
Kafka/Redpanda address | localhost:19092 |
MONGO_URI |
MongoDB connection string | mongodb://voice:voice@localhost:27017/voice_datalake?authSource=admin |
POSTGRES_DSN |
PostgreSQL connection string | postgresql://voice:voice@localhost:5432/voice_warehouse |
VALKEY_URL |
Valkey/Redis URL | redis://localhost:6379/0 |
PIPED_BASE_URL |
Piped backend (for YouTube) | http://localhost:8888 |
GEMINI_API_KEY |
Gemini API key (optional) | (empty) |
USE_GEMINI |
Enable Gemini fallback | true |
# Redpanda + all core services (MongoDB, PostgreSQL, Valkey)
docker compose --profile redpanda up -d
# Add Piped for YouTube support
docker compose --profile redpanda --profile piped up -dVerify everything is running:
docker compose pspip install -r requirements.txtOpen three separate terminals:
# Terminal 1: Scraper worker (listens for keyword jobs)
python -m worker.main
# Terminal 2: Processor (cleaner + AI classifier, runs every 30s)
python -m processor.main
# Terminal 3: REST API
python -m api.maincurl -X POST http://localhost:8000/api/v1/jobs/trigger \
-H "Content-Type: application/json" \
-d '{"keyword": "makan bergizi gratis", "date_range": "30d"}'The job flows through: Redpanda, scraper worker, MongoDB, cleaner, PostgreSQL, AI processor, and is then available via the API.
cd dashboard
npm install
npm run devOpen http://localhost:5173, enter your keyword, and explore the results.
Base URL: http://localhost:8000
All GET endpoints accept keyword (required), date_range (7d | 30d | 90d | all, default 30d), date_from, and date_to.
| Endpoint | Description |
|---|---|
GET /api/v1/summary |
Overall sentiment percentages, article + comment counts |
GET /api/v1/trend |
Daily sentiment breakdown over time |
GET /api/v1/trend-detail |
Comments for a specific date (drill-down) |
GET /api/v1/sources |
Per-source article and comment counts |
GET /api/v1/words |
Top words by sentiment (positive / negative / neutral) |
GET /api/v1/feed |
Paginated comment + article feed with filters |
GET /api/v1/prokontra |
Pro/kontra vote breakdown (Detik and CNN only) |
GET /api/v1/pipeline-stats |
AI classification stats (IndoBERT vs Gemini ratio) |
POST /api/v1/jobs/trigger |
Trigger a scraping + classification job |
Interactive docs available at http://localhost:8000/docs.
Adding a new data source requires one file, with no changes to the pipeline:
from sources import BaseSource
from sources.registry import register_source
@register_source("mysource")
class MySource(BaseSource):
async def collect_urls(self, keyword: str, interval: str) -> list[dict]:
"""Return [{url, title, media, desc, date}, ...]"""
...
async def fetch_detail(self, content: dict) -> dict:
"""Return content with content_id and raw fields added."""
...
async def fetch_comments(self, content: dict) -> list[dict]:
"""Return raw comment objects (one per MongoDB document)."""
...BaseSource.process() handles concurrency, dedup, MongoDB writes, and progress display automatically.
| Phase | Status | Name | Description |
|---|---|---|---|
| 0 | ✅ Done | Foundation | Scaffolding, tooling, local dev environment |
| 1 | ✅ Done | Scraping Core | BaseSource plugin registry, all 6 sources |
| 2 | ✅ Done | Pipeline | Redpanda, MongoDB bronze layer, Valkey dedup |
| 3 | ✅ Done | Cleaning & AI | Cleaner, IndoBERT + Gemini hybrid classifier |
| 4 | ✅ Done | API & Dashboard | FastAPI endpoints, React dashboard |
| 5 | 🔄 In Progress | Bot | Telegram bot interface |
| 6 | ⏳ Planned | Hardening | Retries, observability, test coverage |
| 7 | ⏳ Planned | Scheduling | Airflow DAGs for periodic keyword re-runs |
Coverage is partial: only public, no-login-required sources are scraped. Results represent indexed opinion, not population-representative opinion.
AI limitations: classification reflects biases in the IndoBERT training data and Gemini prompt design. Confidence scores are exposed in the API so consumers can filter by threshold.
Privacy: no user-identifying information is stored after the cleaning stage. Author handles from Reddit/YouTube are dropped before the API layer.
Responsible use: this system must not be used for mass surveillance, targeted harassment, or political manipulation. Scraping respects source rate limits.
MIT. See LICENSE