feat: Add long-running observability and integration features
- Job history tracking with SQLite table - Dry-run mode (--dry-run) to test scrape rules - Plugin system with 3 built-in plugins (sentiment, dedupe, keywords) - REST API server (--api) for Metabase/Grafana integration - Parquet export (--export-parquet) for DuckDB/warehouses - SQLite maintenance (--backup, --vacuum) - Dashboard Integrations tab with external tools guides - Updated Dockerfile and docker-compose.yml for cloud deployment - Comprehensive README documentation
This commit is contained in:
Sanjeev Kumar
parent
a623e5c12d
commit
f65b35f881
15 changed files with 1879 additions and 182 deletions
43
Dockerfile
43
Dockerfile
|
|
@ -1,8 +1,45 @@
|
|||
FROM python:3.9-slim
|
||||
FROM python:3.11-slim
|
||||
|
||||
# Set environment variables
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
ENV PYTHONDONTWRITEBYTECODE=1
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
# Install system dependencies (for some Python packages)
|
||||
RUN apt-get update && apt-get install -y --no-install-recommends \
|
||||
curl \
|
||||
&& rm -rf /var/lib/apt/lists/*
|
||||
|
||||
# Copy requirements first for better caching
|
||||
COPY requirements.txt .
|
||||
COPY main.py .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
RUN mkdir data
|
||||
|
||||
# Copy all source code
|
||||
COPY main.py .
|
||||
COPY config.py .
|
||||
COPY analytics/ ./analytics/
|
||||
COPY alerts/ ./alerts/
|
||||
COPY dashboard/ ./dashboard/
|
||||
COPY export/ ./export/
|
||||
COPY scheduler/ ./scheduler/
|
||||
COPY scraper/ ./scraper/
|
||||
COPY search/ ./search/
|
||||
COPY plugins/ ./plugins/
|
||||
COPY api/ ./api/
|
||||
COPY docs/ ./docs/
|
||||
|
||||
# Create data directory with subdirectories
|
||||
RUN mkdir -p data/backups data/parquet
|
||||
|
||||
# Expose ports
|
||||
# 8501 = Streamlit Dashboard
|
||||
# 8000 = REST API
|
||||
EXPOSE 8501 8000
|
||||
|
||||
# Health check for API mode
|
||||
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
|
||||
CMD curl -f http://localhost:8000/health || exit 1
|
||||
|
||||
# Default: show help
|
||||
ENTRYPOINT ["python", "main.py"]
|
||||
|
|
|
|||
289
README.md
289
README.md
|
|
@ -2,21 +2,25 @@
|
|||
|
||||
[](https://github.com/ksanjeev284/reddit-universal-scraper/actions/workflows/docker-publish.yml)
|
||||
|
||||
A **full-featured** Reddit scraper suite with analytics dashboard, sentiment analysis, scheduled scraping, notifications, and more!
|
||||
A **full-featured** Reddit scraper with analytics dashboard, REST API, scheduled scraping, plugins, and more. **No API keys required!**
|
||||
|
||||
## ✨ Features
|
||||
|
||||
| Feature | Description |
|
||||
|---------|-------------|
|
||||
| 📊 **Full Scraping** | Posts, comments, images, videos, galleries |
|
||||
| 📈 **Analytics Dashboard** | Beautiful Streamlit web UI |
|
||||
| 📈 **Web Dashboard** | Beautiful Streamlit UI with 7 tabs |
|
||||
| 🚀 **REST API** | Connect Metabase, Grafana, DuckDB |
|
||||
| 🔌 **Plugin System** | Extensible post-processing (sentiment, dedupe, keywords) |
|
||||
| 📋 **Job Tracking** | Full history with status, duration, errors |
|
||||
| 🧪 **Dry Run Mode** | Test scrape rules without saving data |
|
||||
| 📦 **Parquet Export** | Analytics-ready format for DuckDB/warehouses |
|
||||
| 😀 **Sentiment Analysis** | Analyze post/comment sentiment |
|
||||
| ☁️ **Keyword Extraction** | Generate word clouds |
|
||||
| 🔍 **Search & Filter** | Query scraped data with filters |
|
||||
| 📅 **Scheduled Scraping** | Cron-style job scheduling |
|
||||
| 📧 **Notifications** | Discord & Telegram alerts |
|
||||
| 🗄️ **SQLite Database** | Structured data storage |
|
||||
| 📤 **Multiple Exports** | CSV, JSON, Excel |
|
||||
| 🗄️ **SQLite Database** | Structured storage with auto-backup |
|
||||
|
||||
---
|
||||
|
||||
## 🚀 Quick Start
|
||||
|
||||
|
|
@ -24,22 +28,25 @@ A **full-featured** Reddit scraper suite with analytics dashboard, sentiment ana
|
|||
# Install dependencies
|
||||
pip install -r requirements.txt
|
||||
|
||||
# Scrape a subreddit (posts + media + comments)
|
||||
python main.py delhi --mode full --limit 100
|
||||
# Scrape a subreddit
|
||||
python main.py python --mode full --limit 100
|
||||
|
||||
# Launch analytics dashboard
|
||||
# Launch dashboard
|
||||
python main.py --dashboard
|
||||
# Opens at http://localhost:8501
|
||||
```
|
||||
|
||||
## 📖 Usage Guide
|
||||
---
|
||||
|
||||
### 🔄 Scraping Modes
|
||||
## 📖 All Commands
|
||||
|
||||
### 🔄 Scraping
|
||||
|
||||
```bash
|
||||
# Full scrape with everything
|
||||
# Full scrape (posts + media + comments)
|
||||
python main.py delhi --mode full --limit 100
|
||||
|
||||
# History only (no media/comments - faster)
|
||||
# Fast history-only (no media/comments)
|
||||
python main.py delhi --mode history --limit 500
|
||||
|
||||
# Live monitor (checks every 5 min)
|
||||
|
|
@ -49,46 +56,95 @@ python main.py delhi --mode monitor
|
|||
python main.py spez --user --mode full --limit 50
|
||||
|
||||
# Skip media or comments
|
||||
python main.py delhi --mode full --no-media --limit 200
|
||||
python main.py delhi --mode full --no-comments --limit 200
|
||||
python main.py delhi --no-media --limit 200
|
||||
python main.py delhi --no-comments --limit 200
|
||||
```
|
||||
|
||||
### 📊 Analytics Dashboard
|
||||
### 🧪 Dry Run Mode
|
||||
|
||||
Test scrape rules without saving any data:
|
||||
|
||||
```bash
|
||||
# Launch the web dashboard
|
||||
python main.py --dashboard
|
||||
python main.py python --mode full --limit 50 --dry-run
|
||||
```
|
||||
|
||||
Output:
|
||||
```
|
||||
🧪 DRY RUN MODE - No data will be saved
|
||||
🧪 DRY RUN COMPLETE!
|
||||
📊 Would scrape: 100 posts
|
||||
💬 Would scrape: 245 comments
|
||||
```
|
||||
|
||||
### 🔌 Plugins
|
||||
|
||||
Enable post-processing plugins:
|
||||
|
||||
```bash
|
||||
# List available plugins
|
||||
python main.py --list-plugins
|
||||
|
||||
# Run with plugins enabled
|
||||
python main.py python --mode full --plugins
|
||||
```
|
||||
|
||||
**Built-in Plugins:**
|
||||
| Plugin | Description |
|
||||
|--------|-------------|
|
||||
| `sentiment_tagger` | Adds sentiment scores to posts |
|
||||
| `deduplicator` | Removes duplicate posts |
|
||||
| `keyword_extractor` | Extracts top keywords |
|
||||
|
||||
Create custom plugins in `plugins/` folder.
|
||||
|
||||
### 📊 Dashboard
|
||||
|
||||
```bash
|
||||
python main.py --dashboard
|
||||
# Opens at http://localhost:8501
|
||||
```
|
||||
|
||||
**Dashboard Features:**
|
||||
- 📈 Post statistics & charts
|
||||
- 😀 Sentiment analysis
|
||||
- ☁️ Keyword extraction
|
||||
- 🔍 Search & filter interface
|
||||
- 📤 Export data
|
||||
**Dashboard Tabs:**
|
||||
- 📊 Overview - Stats & charts
|
||||
- 📈 Analytics - Sentiment & keywords
|
||||
- 🔍 Search - Query scraped data
|
||||
- 💬 Comments - Comment analysis
|
||||
- ⚙️ Scraper - Start new scrapes
|
||||
- 📋 Job History - View all jobs
|
||||
- 🔌 Integrations - API, export, plugins
|
||||
|
||||
### 🔍 Search Data
|
||||
### 🚀 REST API
|
||||
|
||||
```bash
|
||||
# Search all scraped data
|
||||
python main.py --search "credit card"
|
||||
|
||||
# Search with filters
|
||||
python main.py --search "laptop" --min-score 100
|
||||
python main.py --search "advice" --author username
|
||||
python main.py --search "help" --subreddit delhi
|
||||
python main.py --api
|
||||
# API at http://localhost:8000
|
||||
# Docs at http://localhost:8000/docs
|
||||
```
|
||||
|
||||
### 😀 Analytics
|
||||
**Endpoints:**
|
||||
| Endpoint | Description |
|
||||
|----------|-------------|
|
||||
| `GET /posts` | List posts with filters |
|
||||
| `GET /comments` | List comments |
|
||||
| `GET /subreddits` | All scraped subreddits |
|
||||
| `GET /jobs` | Job history |
|
||||
| `GET /query?sql=...` | Raw SQL queries |
|
||||
| `GET /grafana/query` | Grafana time-series |
|
||||
|
||||
### 📦 Export & Maintenance
|
||||
|
||||
```bash
|
||||
# Run sentiment analysis
|
||||
python main.py --analyze delhi --sentiment
|
||||
# Export to Parquet (for DuckDB/warehouses)
|
||||
python main.py --export-parquet python
|
||||
|
||||
# Extract top keywords
|
||||
python main.py --analyze delhi --keywords
|
||||
# View job history
|
||||
python main.py --job-history
|
||||
|
||||
# Backup database
|
||||
python main.py --backup
|
||||
|
||||
# Optimize database
|
||||
python main.py --vacuum
|
||||
```
|
||||
|
||||
### 📅 Scheduled Scraping
|
||||
|
|
@ -97,51 +153,125 @@ python main.py --analyze delhi --keywords
|
|||
# Scrape every 60 minutes
|
||||
python main.py --schedule delhi --every 60
|
||||
|
||||
# Scrape with options
|
||||
# With options
|
||||
python main.py --schedule delhi --every 30 --mode full --limit 50
|
||||
```
|
||||
|
||||
### 📧 Notifications (Discord/Telegram)
|
||||
### 🔍 Search & Analytics
|
||||
|
||||
**Discord:**
|
||||
```bash
|
||||
python main.py delhi --mode monitor --discord-webhook "YOUR_WEBHOOK_URL"
|
||||
# Search scraped data
|
||||
python main.py --search "credit card" --min-score 100
|
||||
|
||||
# Run sentiment analysis
|
||||
python main.py --analyze delhi --sentiment
|
||||
|
||||
# Extract keywords
|
||||
python main.py --analyze delhi --keywords
|
||||
```
|
||||
|
||||
**Telegram:**
|
||||
---
|
||||
|
||||
## 🐳 Docker
|
||||
|
||||
### Quick Start
|
||||
|
||||
```bash
|
||||
python main.py delhi --mode monitor \
|
||||
--telegram-token "YOUR_BOT_TOKEN" \
|
||||
--telegram-chat "YOUR_CHAT_ID"
|
||||
# Build
|
||||
docker build -t reddit-scraper .
|
||||
|
||||
# Run scrape
|
||||
docker run -v ./data:/app/data reddit-scraper python --limit 100
|
||||
|
||||
# Run with plugins
|
||||
docker run -v ./data:/app/data reddit-scraper python --plugins
|
||||
```
|
||||
|
||||
### Docker Compose (Full Stack)
|
||||
|
||||
```bash
|
||||
# Start API + Dashboard
|
||||
docker-compose up -d
|
||||
|
||||
# Access:
|
||||
# Dashboard: http://localhost:8501
|
||||
# API: http://localhost:8000/docs
|
||||
```
|
||||
|
||||
### Deploy to AWS/VPS
|
||||
|
||||
```bash
|
||||
# SSH into your server
|
||||
ssh user@your-server-ip
|
||||
|
||||
# Clone repo
|
||||
git clone https://github.com/ksanjeev284/reddit-universal-scraper.git
|
||||
cd reddit-universal-scraper
|
||||
|
||||
# Start services
|
||||
docker-compose up -d
|
||||
|
||||
# Open firewall ports
|
||||
sudo ufw allow 8000
|
||||
sudo ufw allow 8501
|
||||
```
|
||||
|
||||
Access:
|
||||
- `http://your-server-ip:8501` → Dashboard
|
||||
- `http://your-server-ip:8000/docs` → API
|
||||
|
||||
---
|
||||
|
||||
## 🔗 External Integrations
|
||||
|
||||
### Metabase
|
||||
|
||||
1. Start API: `python main.py --api`
|
||||
2. Add HTTP datasource: `http://localhost:8000`
|
||||
3. Query: `/posts?subreddit=python&limit=100`
|
||||
|
||||
### Grafana
|
||||
|
||||
1. Install "JSON API" or "Infinity" plugin
|
||||
2. Add datasource: `http://localhost:8000`
|
||||
3. Use `/grafana/query` for time-series
|
||||
|
||||
### DuckDB
|
||||
|
||||
```python
|
||||
import duckdb
|
||||
|
||||
# Export to Parquet first
|
||||
# python main.py --export-parquet python
|
||||
|
||||
# Query directly
|
||||
duckdb.query("SELECT * FROM 'data/parquet/*.parquet'").df()
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 📁 Project Structure
|
||||
|
||||
```
|
||||
reddit-scraper/
|
||||
├── main.py # Main CLI entry point
|
||||
├── config.py # Configuration settings
|
||||
├── analytics/ # Sentiment & keyword analysis
|
||||
│ └── sentiment.py
|
||||
├── alerts/ # Discord & Telegram notifications
|
||||
│ └── notifications.py
|
||||
├── dashboard/ # Streamlit web UI
|
||||
│ └── app.py
|
||||
├── export/ # Database & export functions
|
||||
│ └── database.py
|
||||
├── scheduler/ # Cron-style scheduling
|
||||
│ └── cron.py
|
||||
├── search/ # Search & filter engine
|
||||
│ └── query.py
|
||||
└── data/ # Scraped data
|
||||
└── r_subreddit/
|
||||
├── posts.csv
|
||||
├── comments.csv
|
||||
└── media/
|
||||
├── images/
|
||||
└── videos/
|
||||
├── main.py # CLI entry point
|
||||
├── config.py # Settings
|
||||
├── analytics/ # Sentiment & keywords
|
||||
├── alerts/ # Discord/Telegram
|
||||
├── api/ # REST API server
|
||||
├── dashboard/ # Streamlit UI
|
||||
├── export/ # Database & exports
|
||||
├── plugins/ # Post-processing plugins
|
||||
├── scheduler/ # Cron scheduling
|
||||
├── search/ # Search engine
|
||||
└── data/
|
||||
├── r_subreddit/ # Scraped data
|
||||
├── backups/ # DB backups
|
||||
└── parquet/ # Parquet exports
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 📊 Data Output
|
||||
|
||||
### posts.csv
|
||||
|
|
@ -154,9 +284,7 @@ reddit-scraper/
|
|||
| num_comments | Comment count |
|
||||
| post_type | text/image/video/gallery |
|
||||
| selftext | Post body |
|
||||
| flair | Post flair |
|
||||
| is_nsfw | NSFW flag |
|
||||
| created_utc | Timestamp |
|
||||
| sentiment_score | -1.0 to 1.0 (with plugins) |
|
||||
|
||||
### comments.csv
|
||||
| Column | Description |
|
||||
|
|
@ -166,31 +294,20 @@ reddit-scraper/
|
|||
| author | Username |
|
||||
| body | Comment text |
|
||||
| score | Upvotes |
|
||||
| depth | Nesting level |
|
||||
|
||||
## 🐳 Docker
|
||||
|
||||
```bash
|
||||
# Build
|
||||
docker build -t reddit-scraper .
|
||||
|
||||
# Full scrape
|
||||
docker run -v $(pwd)/data:/app/data reddit-scraper delhi --mode full --limit 100
|
||||
|
||||
# Monitor mode
|
||||
docker run -d -v $(pwd)/data:/app/data reddit-scraper delhi --mode monitor
|
||||
```
|
||||
|
||||
## ⚙️ Configuration
|
||||
|
||||
Edit `config.py` or use environment variables:
|
||||
---
|
||||
|
||||
## ⚙️ Environment Variables
|
||||
|
||||
```bash
|
||||
# Notifications
|
||||
export DISCORD_WEBHOOK_URL="https://discord.com/api/webhooks/..."
|
||||
export TELEGRAM_BOT_TOKEN="123456:ABC..."
|
||||
export TELEGRAM_CHAT_ID="987654321"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 📜 License
|
||||
|
||||
MIT License - Feel free to use, modify, and distribute.
|
||||
|
|
|
|||
2
api/__init__.py
Normal file
2
api/__init__.py
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
"""Reddit Scraper REST API"""
|
||||
from .server import app
|
||||
244
api/server.py
Normal file
244
api/server.py
Normal file
|
|
@ -0,0 +1,244 @@
|
|||
"""
|
||||
REST API Module - Expose Reddit Scraper data as a REST API
|
||||
For integration with Metabase, Grafana, DreamFactory, and other tools.
|
||||
|
||||
Start with: python api/server.py
|
||||
Or: uvicorn api.server:app --reload --port 8000
|
||||
"""
|
||||
from fastapi import FastAPI, Query, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from typing import Optional, List
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add parent to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from export.database import (
|
||||
get_connection, search_posts, search_comments,
|
||||
get_subreddit_stats, get_all_subreddits,
|
||||
get_job_history, get_job_stats, get_database_info
|
||||
)
|
||||
|
||||
# Create FastAPI app
|
||||
app = FastAPI(
|
||||
title="Reddit Scraper API",
|
||||
description="REST API for Reddit Scraper data. Use with Metabase, Grafana, or any tool.",
|
||||
version="1.0.0",
|
||||
docs_url="/docs",
|
||||
redoc_url="/redoc"
|
||||
)
|
||||
|
||||
# Enable CORS for external tools
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"], # Allow all origins for local tools
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
|
||||
# --- HEALTH & INFO ---
|
||||
|
||||
@app.get("/", tags=["Info"])
|
||||
def root():
|
||||
"""API root - basic info."""
|
||||
return {
|
||||
"name": "Reddit Scraper API",
|
||||
"version": "1.0.0",
|
||||
"docs": "/docs",
|
||||
"endpoints": ["/posts", "/comments", "/subreddits", "/jobs", "/stats"]
|
||||
}
|
||||
|
||||
|
||||
@app.get("/health", tags=["Info"])
|
||||
def health_check():
|
||||
"""Health check endpoint."""
|
||||
try:
|
||||
info = get_database_info()
|
||||
return {"status": "healthy", "database": info}
|
||||
except Exception as e:
|
||||
return {"status": "unhealthy", "error": str(e)}
|
||||
|
||||
|
||||
@app.get("/info", tags=["Info"])
|
||||
def database_info():
|
||||
"""Get database info and table counts."""
|
||||
return get_database_info()
|
||||
|
||||
|
||||
# --- POSTS ---
|
||||
|
||||
@app.get("/posts", tags=["Posts"])
|
||||
def list_posts(
|
||||
q: Optional[str] = Query(None, description="Search query"),
|
||||
subreddit: Optional[str] = Query(None, description="Filter by subreddit"),
|
||||
author: Optional[str] = Query(None, description="Filter by author"),
|
||||
min_score: Optional[int] = Query(None, description="Minimum score"),
|
||||
post_type: Optional[str] = Query(None, description="Post type filter"),
|
||||
limit: int = Query(100, ge=1, le=1000, description="Max results")
|
||||
):
|
||||
"""
|
||||
Get posts with optional filters.
|
||||
|
||||
Use for Grafana dashboards, Metabase queries, or custom integrations.
|
||||
"""
|
||||
return search_posts(
|
||||
query=q,
|
||||
subreddit=subreddit,
|
||||
author=author,
|
||||
min_score=min_score,
|
||||
post_type=post_type,
|
||||
limit=limit
|
||||
)
|
||||
|
||||
|
||||
@app.get("/posts/{post_id}", tags=["Posts"])
|
||||
def get_post(post_id: str):
|
||||
"""Get a single post by ID."""
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT * FROM posts WHERE id = ?", (post_id,))
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
|
||||
if not row:
|
||||
raise HTTPException(status_code=404, detail="Post not found")
|
||||
return dict(row)
|
||||
|
||||
|
||||
# --- COMMENTS ---
|
||||
|
||||
@app.get("/comments", tags=["Comments"])
|
||||
def list_comments(
|
||||
q: Optional[str] = Query(None, description="Search in comment body"),
|
||||
post_id: Optional[str] = Query(None, description="Filter by post ID"),
|
||||
author: Optional[str] = Query(None, description="Filter by author"),
|
||||
min_score: Optional[int] = Query(None, description="Minimum score"),
|
||||
limit: int = Query(100, ge=1, le=1000, description="Max results")
|
||||
):
|
||||
"""Get comments with optional filters."""
|
||||
return search_comments(
|
||||
query=q,
|
||||
post_id=post_id,
|
||||
author=author,
|
||||
min_score=min_score,
|
||||
limit=limit
|
||||
)
|
||||
|
||||
|
||||
# --- SUBREDDITS ---
|
||||
|
||||
@app.get("/subreddits", tags=["Subreddits"])
|
||||
def list_subreddits():
|
||||
"""Get all scraped subreddits with post counts."""
|
||||
return get_all_subreddits()
|
||||
|
||||
|
||||
@app.get("/subreddits/{subreddit}/stats", tags=["Subreddits"])
|
||||
def subreddit_stats(subreddit: str):
|
||||
"""Get detailed statistics for a subreddit."""
|
||||
stats = get_subreddit_stats(subreddit)
|
||||
if not stats.get('total_posts'):
|
||||
raise HTTPException(status_code=404, detail=f"No data for r/{subreddit}")
|
||||
return stats
|
||||
|
||||
|
||||
# --- JOBS ---
|
||||
|
||||
@app.get("/jobs", tags=["Jobs"])
|
||||
def list_jobs(
|
||||
status: Optional[str] = Query(None, description="Filter by status"),
|
||||
target: Optional[str] = Query(None, description="Filter by target"),
|
||||
limit: int = Query(50, ge=1, le=200)
|
||||
):
|
||||
"""Get job history."""
|
||||
return get_job_history(limit=limit, target=target, status=status)
|
||||
|
||||
|
||||
@app.get("/jobs/stats", tags=["Jobs"])
|
||||
def job_stats():
|
||||
"""Get aggregated job statistics."""
|
||||
return get_job_stats()
|
||||
|
||||
|
||||
# --- RAW SQL (for advanced users) ---
|
||||
|
||||
@app.get("/query", tags=["Advanced"])
|
||||
def raw_query(
|
||||
sql: str = Query(..., description="SQL SELECT query"),
|
||||
limit: int = Query(100, ge=1, le=1000)
|
||||
):
|
||||
"""
|
||||
Execute a raw SQL SELECT query.
|
||||
|
||||
⚠️ Only SELECT queries allowed. Use for custom Grafana/Metabase queries.
|
||||
|
||||
Example: /query?sql=SELECT title, score FROM posts ORDER BY score DESC
|
||||
"""
|
||||
# Security: Only allow SELECT
|
||||
if not sql.strip().upper().startswith("SELECT"):
|
||||
raise HTTPException(status_code=400, detail="Only SELECT queries allowed")
|
||||
|
||||
# Add limit if not present
|
||||
if "LIMIT" not in sql.upper():
|
||||
sql = f"{sql} LIMIT {limit}"
|
||||
|
||||
try:
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(sql)
|
||||
results = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
return {"query": sql, "count": len(results), "results": results}
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=400, detail=f"Query error: {e}")
|
||||
|
||||
|
||||
# --- GRAFANA COMPATIBLE ENDPOINTS ---
|
||||
|
||||
@app.get("/grafana/search", tags=["Grafana"])
|
||||
def grafana_search():
|
||||
"""Grafana SimpleJSON datasource - search endpoint."""
|
||||
subs = get_all_subreddits()
|
||||
return [s['subreddit'] for s in subs]
|
||||
|
||||
|
||||
@app.post("/grafana/query", tags=["Grafana"])
|
||||
def grafana_query(body: dict):
|
||||
"""Grafana SimpleJSON datasource - query endpoint."""
|
||||
# Return time series data for Grafana
|
||||
results = []
|
||||
|
||||
for target in body.get('targets', []):
|
||||
subreddit = target.get('target')
|
||||
if subreddit:
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT date(created_utc) as time, COUNT(*) as value
|
||||
FROM posts WHERE subreddit = ?
|
||||
GROUP BY date(created_utc)
|
||||
ORDER BY time
|
||||
""", (subreddit,))
|
||||
|
||||
datapoints = [[row['value'], row['time']] for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
|
||||
results.append({
|
||||
"target": subreddit,
|
||||
"datapoints": datapoints
|
||||
})
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# --- CLI ---
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
print("🚀 Starting Reddit Scraper API...")
|
||||
print(" 📖 Docs: http://localhost:8000/docs")
|
||||
print(" 📊 Use with Metabase, Grafana, or any REST client")
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
240
dashboard/app.py
240
dashboard/app.py
|
|
@ -115,8 +115,8 @@ def main():
|
|||
comments_df = data.get('comments', pd.DataFrame())
|
||||
|
||||
# Main content tabs
|
||||
tab1, tab2, tab3, tab4, tab5 = st.tabs([
|
||||
"📊 Overview", "📈 Analytics", "🔍 Search", "💬 Comments", "⚙️ Scraper"
|
||||
tab1, tab2, tab3, tab4, tab5, tab6, tab7 = st.tabs([
|
||||
"📊 Overview", "📈 Analytics", "🔍 Search", "💬 Comments", "⚙️ Scraper", "📋 Job History", "🔌 Integrations"
|
||||
])
|
||||
|
||||
with tab1:
|
||||
|
|
@ -359,6 +359,242 @@ def main():
|
|||
f"{selected_sub}_posts.json",
|
||||
"application/json"
|
||||
)
|
||||
|
||||
with tab6:
|
||||
st.header("📋 Job History")
|
||||
|
||||
try:
|
||||
from export.database import get_job_history, get_job_stats
|
||||
|
||||
# Job stats
|
||||
stats = get_job_stats()
|
||||
|
||||
col1, col2, col3, col4 = st.columns(4)
|
||||
with col1:
|
||||
st.metric("Total Jobs", stats.get('total_jobs', 0))
|
||||
with col2:
|
||||
st.metric("Completed", stats.get('completed', 0))
|
||||
with col3:
|
||||
st.metric("Failed", stats.get('failed', 0))
|
||||
with col4:
|
||||
avg_dur = stats.get('avg_duration')
|
||||
st.metric("Avg Duration", f"{avg_dur:.1f}s" if avg_dur else "-")
|
||||
|
||||
st.divider()
|
||||
|
||||
# Job history table
|
||||
st.subheader("Recent Jobs")
|
||||
|
||||
col1, col2 = st.columns(2)
|
||||
with col1:
|
||||
filter_status = st.selectbox("Filter by Status", ['All', 'completed', 'failed', 'running'])
|
||||
with col2:
|
||||
limit = st.number_input("Show last N jobs", min_value=10, max_value=100, value=20)
|
||||
|
||||
status_filter = None if filter_status == 'All' else filter_status
|
||||
jobs = get_job_history(limit=limit, status=status_filter)
|
||||
|
||||
if jobs:
|
||||
jobs_df = pd.DataFrame(jobs)
|
||||
# Format for display
|
||||
display_cols = ['job_id', 'target', 'mode', 'status', 'posts_scraped',
|
||||
'comments_scraped', 'duration_seconds', 'started_at', 'dry_run']
|
||||
display_cols = [c for c in display_cols if c in jobs_df.columns]
|
||||
st.dataframe(jobs_df[display_cols], use_container_width=True)
|
||||
|
||||
# Success rate chart
|
||||
st.subheader("Success Rate")
|
||||
if 'status' in jobs_df.columns:
|
||||
status_counts = jobs_df['status'].value_counts()
|
||||
st.bar_chart(status_counts)
|
||||
else:
|
||||
st.info("No job history found. Run some scrapes first!")
|
||||
|
||||
except Exception as e:
|
||||
st.error(f"Failed to load job history: {e}")
|
||||
st.info("Make sure the database is initialized.")
|
||||
|
||||
with tab7:
|
||||
st.header("🔌 Integrations & Settings")
|
||||
|
||||
# REST API Section
|
||||
st.subheader("🚀 REST API")
|
||||
|
||||
col1, col2 = st.columns(2)
|
||||
with col1:
|
||||
st.markdown("""
|
||||
**Start the API server:**
|
||||
```bash
|
||||
python main.py --api
|
||||
```
|
||||
""")
|
||||
with col2:
|
||||
api_port = st.number_input("API Port", value=8000, min_value=1000, max_value=65535)
|
||||
st.code(f"http://localhost:{api_port}/docs")
|
||||
|
||||
st.markdown("""
|
||||
**Available Endpoints:**
|
||||
| Endpoint | Description |
|
||||
|----------|-------------|
|
||||
| `/posts` | List posts with filters |
|
||||
| `/comments` | List comments |
|
||||
| `/subreddits` | All scraped subreddits |
|
||||
| `/jobs` | Job history |
|
||||
| `/query?sql=...` | Raw SQL queries |
|
||||
| `/docs` | Interactive Swagger UI |
|
||||
""")
|
||||
|
||||
st.divider()
|
||||
|
||||
# External Tools
|
||||
st.subheader("📊 External Tools Integration")
|
||||
|
||||
tool_tabs = st.tabs(["📈 Metabase", "📊 Grafana", "🔗 DreamFactory", "🧦 DuckDB"])
|
||||
|
||||
with tool_tabs[0]:
|
||||
st.markdown("""
|
||||
**Metabase Setup:**
|
||||
1. Start API: `python main.py --api`
|
||||
2. In Metabase: New Question → Native Query
|
||||
3. Use HTTP datasource with `http://localhost:8000`
|
||||
4. Query: `/posts?subreddit=python&limit=100`
|
||||
|
||||
**Or use raw SQL:**
|
||||
```
|
||||
/query?sql=SELECT title, score FROM posts ORDER BY score DESC
|
||||
```
|
||||
""")
|
||||
|
||||
with tool_tabs[1]:
|
||||
st.markdown("""
|
||||
**Grafana Setup:**
|
||||
1. Install "JSON API" or "Infinity" plugin
|
||||
2. Add datasource: `http://localhost:8000`
|
||||
3. Use `/grafana/query` for time-series
|
||||
|
||||
**Example Panel Query:**
|
||||
```sql
|
||||
SELECT date(created_utc) as time, COUNT(*) as posts
|
||||
FROM posts GROUP BY date(created_utc)
|
||||
```
|
||||
""")
|
||||
|
||||
with tool_tabs[2]:
|
||||
st.markdown("""
|
||||
**DreamFactory Setup:**
|
||||
1. Point to SQLite file: `data/reddit_scraper.db`
|
||||
2. Or use REST API: `http://localhost:8000`
|
||||
3. Auto-generates API for all tables
|
||||
""")
|
||||
|
||||
with tool_tabs[3]:
|
||||
st.markdown("""
|
||||
**DuckDB (Analytics):**
|
||||
1. Export to Parquet first (see below)
|
||||
2. Query directly:
|
||||
```python
|
||||
import duckdb
|
||||
duckdb.query("SELECT * FROM 'data/parquet/*.parquet'").df()
|
||||
```
|
||||
""")
|
||||
|
||||
st.divider()
|
||||
|
||||
# Parquet Export
|
||||
st.subheader("📦 Parquet Export")
|
||||
|
||||
col1, col2 = st.columns(2)
|
||||
with col1:
|
||||
export_sub = st.selectbox("Select subreddit to export", subreddits, key="parquet_export")
|
||||
with col2:
|
||||
if st.button("📦 Export to Parquet"):
|
||||
st.info(f"Run: `python main.py --export-parquet {export_sub.replace('r_', '').replace('u_', '')}`")
|
||||
|
||||
# List existing parquet files
|
||||
from pathlib import Path
|
||||
parquet_dir = Path("data/parquet")
|
||||
if parquet_dir.exists():
|
||||
parquet_files = list(parquet_dir.glob("*.parquet"))
|
||||
if parquet_files:
|
||||
st.write("**Existing Parquet files:**")
|
||||
for f in parquet_files[:10]:
|
||||
size_mb = f.stat().st_size / (1024 * 1024)
|
||||
st.text(f" • {f.name} ({size_mb:.2f} MB)")
|
||||
|
||||
st.divider()
|
||||
|
||||
# Database Maintenance
|
||||
st.subheader("🛠️ Database Maintenance")
|
||||
|
||||
col1, col2, col3 = st.columns(3)
|
||||
|
||||
with col1:
|
||||
if st.button("💾 Backup Database"):
|
||||
st.info("Run: `python main.py --backup`")
|
||||
|
||||
with col2:
|
||||
if st.button("🧹 Vacuum/Optimize"):
|
||||
st.info("Run: `python main.py --vacuum`")
|
||||
|
||||
with col3:
|
||||
try:
|
||||
from export.database import get_database_info
|
||||
db_info = get_database_info()
|
||||
st.metric("DB Size", f"{db_info.get('size_mb', 0):.2f} MB")
|
||||
except:
|
||||
st.metric("DB Size", "N/A")
|
||||
|
||||
# Show backup files
|
||||
backup_dir = Path("data/backups")
|
||||
if backup_dir.exists():
|
||||
backups = sorted(backup_dir.glob("*.db"), reverse=True)[:5]
|
||||
if backups:
|
||||
st.write("**Recent Backups:**")
|
||||
for b in backups:
|
||||
size_mb = b.stat().st_size / (1024 * 1024)
|
||||
st.text(f" • {b.name} ({size_mb:.2f} MB)")
|
||||
|
||||
st.divider()
|
||||
|
||||
# Plugin Configuration
|
||||
st.subheader("🔌 Plugins")
|
||||
|
||||
try:
|
||||
from plugins import load_plugins
|
||||
plugins = load_plugins()
|
||||
|
||||
if plugins:
|
||||
st.write("**Available Plugins:**")
|
||||
for plugin in plugins:
|
||||
status = "✅" if plugin.enabled else "❌"
|
||||
st.markdown(f"{status} **{plugin.name}** - {plugin.description}")
|
||||
|
||||
st.info("💡 Enable plugins when scraping: `python main.py <target> --plugins`")
|
||||
else:
|
||||
st.warning("No plugins found in plugins/ directory")
|
||||
except Exception as e:
|
||||
st.error(f"Plugin loading error: {e}")
|
||||
|
||||
st.divider()
|
||||
|
||||
# Quick Commands Reference
|
||||
st.subheader("📋 Quick Commands")
|
||||
st.code("""
|
||||
# Start REST API
|
||||
python main.py --api
|
||||
|
||||
# Export to Parquet
|
||||
python main.py --export-parquet <subreddit>
|
||||
|
||||
# Backup database
|
||||
python main.py --backup
|
||||
|
||||
# Scrape with plugins
|
||||
python main.py <target> --plugins
|
||||
|
||||
# Dry run (test without saving)
|
||||
python main.py <target> --dry-run
|
||||
""", language="bash")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
98
docker-compose.yml
Normal file
98
docker-compose.yml
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
version: '3.8'
|
||||
|
||||
# Reddit Scraper Suite - Full Stack
|
||||
# Start with: docker-compose up -d
|
||||
|
||||
services:
|
||||
# Main Scraper (run scrape jobs)
|
||||
scraper:
|
||||
build: .
|
||||
volumes:
|
||||
- ./data:/app/data # Persist scraped data
|
||||
command: ["--help"] # Override with your scrape command
|
||||
profiles: ["scrape"] # Only run when explicitly requested
|
||||
|
||||
# REST API Server (for Metabase/Grafana integration)
|
||||
api:
|
||||
build: .
|
||||
ports:
|
||||
- "8000:8000"
|
||||
volumes:
|
||||
- ./data:/app/data
|
||||
command: ["--api"]
|
||||
restart: unless-stopped
|
||||
healthcheck:
|
||||
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
|
||||
interval: 30s
|
||||
timeout: 10s
|
||||
retries: 3
|
||||
|
||||
# Streamlit Dashboard
|
||||
dashboard:
|
||||
build: .
|
||||
ports:
|
||||
- "8501:8501"
|
||||
volumes:
|
||||
- ./data:/app/data
|
||||
command: ["--dashboard"]
|
||||
restart: unless-stopped
|
||||
depends_on:
|
||||
- api
|
||||
|
||||
# Scheduled Scraper (optional - uncomment and configure)
|
||||
# scheduler:
|
||||
# build: .
|
||||
# volumes:
|
||||
# - ./data:/app/data
|
||||
# command: ["--schedule", "python", "--every", "60"]
|
||||
# restart: unless-stopped
|
||||
|
||||
# Optional: Add Metabase for data visualization
|
||||
# Uncomment to enable
|
||||
#
|
||||
# metabase:
|
||||
# image: metabase/metabase:latest
|
||||
# ports:
|
||||
# - "3000:3000"
|
||||
# environment:
|
||||
# MB_DB_TYPE: h2
|
||||
# volumes:
|
||||
# - metabase-data:/metabase-data
|
||||
# depends_on:
|
||||
# - api
|
||||
|
||||
# ===========================================
|
||||
# PRODUCTION DEPLOYMENT (AWS/VPS)
|
||||
# ===========================================
|
||||
# Uncomment the nginx service below for:
|
||||
# - HTTPS/SSL termination
|
||||
# - Basic authentication
|
||||
# - Single port exposure (80/443)
|
||||
|
||||
# nginx:
|
||||
# image: nginx:alpine
|
||||
# ports:
|
||||
# - "80:80"
|
||||
# - "443:443"
|
||||
# volumes:
|
||||
# - ./nginx.conf:/etc/nginx/nginx.conf:ro
|
||||
# - ./ssl:/etc/nginx/ssl:ro # Add your SSL certs
|
||||
# depends_on:
|
||||
# - api
|
||||
# - dashboard
|
||||
|
||||
# volumes:
|
||||
# metabase-data:
|
||||
|
||||
# ===========================================
|
||||
# QUICK DEPLOY TO AWS/VPS:
|
||||
# ===========================================
|
||||
# 1. SSH into your server
|
||||
# 2. git clone <your-repo>
|
||||
# 3. docker-compose up -d
|
||||
# 4. Open firewall: ports 8000, 8501
|
||||
#
|
||||
# Access:
|
||||
# http://<your-server-ip>:8501 (Dashboard)
|
||||
# http://<your-server-ip>:8000 (API)
|
||||
# ===========================================
|
||||
93
docs/INTEGRATION.md
Normal file
93
docs/INTEGRATION.md
Normal file
|
|
@ -0,0 +1,93 @@
|
|||
# External Tools Integration Guide
|
||||
|
||||
Connect Metabase, Grafana, DreamFactory, or any REST client to your Reddit scraper data.
|
||||
|
||||
---
|
||||
|
||||
## Quick Start
|
||||
|
||||
```powershell
|
||||
# Install dependencies
|
||||
pip install fastapi uvicorn
|
||||
|
||||
# Start the API server
|
||||
python main.py --api
|
||||
```
|
||||
|
||||
The API will be available at `http://localhost:8000`
|
||||
|
||||
---
|
||||
|
||||
## API Endpoints
|
||||
|
||||
| Endpoint | Description |
|
||||
|----------|-------------|
|
||||
| `GET /posts` | List posts with filters (q, subreddit, author, min_score) |
|
||||
| `GET /posts/{id}` | Get single post |
|
||||
| `GET /comments` | List comments with filters |
|
||||
| `GET /subreddits` | List all scraped subreddits |
|
||||
| `GET /subreddits/{name}/stats` | Get subreddit statistics |
|
||||
| `GET /jobs` | View job history |
|
||||
| `GET /jobs/stats` | Job statistics |
|
||||
| `GET /query?sql=...` | Raw SQL SELECT queries |
|
||||
| `GET /docs` | Interactive API documentation |
|
||||
|
||||
---
|
||||
|
||||
## Metabase Setup
|
||||
|
||||
1. Start API: `python main.py --api`
|
||||
2. In Metabase, add a new "HTTP" question
|
||||
3. Use `http://localhost:8000/posts?limit=1000`
|
||||
4. Or use `/query?sql=SELECT * FROM posts` for custom queries
|
||||
|
||||
---
|
||||
|
||||
## Grafana Setup
|
||||
|
||||
1. Install "JSON API" or "Infinity" datasource plugin
|
||||
2. Add datasource with URL: `http://localhost:8000`
|
||||
3. Use `/grafana/query` for time-series data
|
||||
4. Or use `/query?sql=...` for custom queries
|
||||
|
||||
Example Grafana query:
|
||||
```sql
|
||||
SELECT date(created_utc) as time, COUNT(*) as posts
|
||||
FROM posts
|
||||
GROUP BY date(created_utc)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## DreamFactory / REST Clients
|
||||
|
||||
The API includes full CORS support. Connect any tool that speaks REST:
|
||||
|
||||
```bash
|
||||
# Get posts
|
||||
curl http://localhost:8000/posts?subreddit=python&limit=10
|
||||
|
||||
# Custom SQL query
|
||||
curl "http://localhost:8000/query?sql=SELECT title, score FROM posts ORDER BY score DESC LIMIT 5"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Docker Compose (All-in-One)
|
||||
|
||||
```yaml
|
||||
version: '3'
|
||||
services:
|
||||
scraper-api:
|
||||
build: .
|
||||
ports:
|
||||
- "8000:8000"
|
||||
volumes:
|
||||
- ./data:/app/data
|
||||
command: python main.py --api
|
||||
|
||||
metabase:
|
||||
image: metabase/metabase
|
||||
ports:
|
||||
- "3000:3000"
|
||||
```
|
||||
|
|
@ -112,6 +112,27 @@ def init_database():
|
|||
)
|
||||
""")
|
||||
|
||||
# Job history table for observability
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS job_history (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
job_id TEXT UNIQUE,
|
||||
target TEXT,
|
||||
is_user BOOLEAN DEFAULT 0,
|
||||
mode TEXT,
|
||||
status TEXT,
|
||||
started_at TEXT,
|
||||
completed_at TEXT,
|
||||
duration_seconds REAL,
|
||||
posts_scraped INTEGER DEFAULT 0,
|
||||
comments_scraped INTEGER DEFAULT 0,
|
||||
media_downloaded INTEGER DEFAULT 0,
|
||||
errors TEXT,
|
||||
error_count INTEGER DEFAULT 0,
|
||||
dry_run BOOLEAN DEFAULT 0
|
||||
)
|
||||
""")
|
||||
|
||||
# Create indexes
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_posts_subreddit ON posts(subreddit)")
|
||||
cursor.execute("CREATE INDEX IF NOT EXISTS idx_posts_created ON posts(created_utc)")
|
||||
|
|
@ -385,5 +406,237 @@ def get_all_subreddits():
|
|||
conn.close()
|
||||
return results
|
||||
|
||||
# --- JOB HISTORY FUNCTIONS ---
|
||||
|
||||
def start_job_record(target, mode, is_user=False, dry_run=False):
|
||||
"""
|
||||
Start tracking a new scrape job.
|
||||
|
||||
Returns:
|
||||
job_id: Unique identifier for the job
|
||||
"""
|
||||
import uuid
|
||||
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
job_id = str(uuid.uuid4())[:8]
|
||||
started_at = datetime.now().isoformat()
|
||||
|
||||
cursor.execute("""
|
||||
INSERT INTO job_history (job_id, target, is_user, mode, status, started_at, dry_run)
|
||||
VALUES (?, ?, ?, ?, 'running', ?, ?)
|
||||
""", (job_id, target, is_user, mode, started_at, dry_run))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
print(f"📋 Job started: {job_id}")
|
||||
return job_id
|
||||
|
||||
def complete_job_record(job_id, status, posts=0, comments=0, media=0, errors=None):
|
||||
"""
|
||||
Complete a job record with results.
|
||||
|
||||
Args:
|
||||
job_id: Job ID from start_job_record
|
||||
status: 'completed' or 'failed'
|
||||
posts: Number of posts scraped
|
||||
comments: Number of comments scraped
|
||||
media: Number of media files downloaded
|
||||
errors: Error message if failed
|
||||
"""
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
completed_at = datetime.now().isoformat()
|
||||
|
||||
# Calculate duration
|
||||
cursor.execute("SELECT started_at FROM job_history WHERE job_id = ?", (job_id,))
|
||||
row = cursor.fetchone()
|
||||
|
||||
duration = 0
|
||||
error_count = 0
|
||||
if row:
|
||||
started = datetime.fromisoformat(row['started_at'])
|
||||
duration = (datetime.now() - started).total_seconds()
|
||||
|
||||
if errors:
|
||||
error_count = 1
|
||||
|
||||
cursor.execute("""
|
||||
UPDATE job_history
|
||||
SET status = ?, completed_at = ?, duration_seconds = ?,
|
||||
posts_scraped = ?, comments_scraped = ?, media_downloaded = ?,
|
||||
errors = ?, error_count = ?
|
||||
WHERE job_id = ?
|
||||
""", (status, completed_at, duration, posts, comments, media, errors, error_count, job_id))
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
if status == 'completed':
|
||||
print(f"✅ Job {job_id} completed: {posts} posts, {comments} comments in {duration:.1f}s")
|
||||
else:
|
||||
print(f"❌ Job {job_id} failed: {errors}")
|
||||
|
||||
def get_job_history(limit=50, target=None, status=None):
|
||||
"""Get recent job history."""
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
sql = "SELECT * FROM job_history WHERE 1=1"
|
||||
params = []
|
||||
|
||||
if target:
|
||||
sql += " AND target = ?"
|
||||
params.append(target)
|
||||
|
||||
if status:
|
||||
sql += " AND status = ?"
|
||||
params.append(status)
|
||||
|
||||
sql += " ORDER BY started_at DESC LIMIT ?"
|
||||
params.append(limit)
|
||||
|
||||
cursor.execute(sql, params)
|
||||
results = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
return results
|
||||
|
||||
def get_job_stats():
|
||||
"""Get aggregated job statistics."""
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
stats = {}
|
||||
|
||||
# Overall counts
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
COUNT(*) as total_jobs,
|
||||
SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as completed,
|
||||
SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) as failed,
|
||||
SUM(CASE WHEN status = 'running' THEN 1 ELSE 0 END) as running,
|
||||
AVG(duration_seconds) as avg_duration,
|
||||
SUM(posts_scraped) as total_posts,
|
||||
SUM(comments_scraped) as total_comments
|
||||
FROM job_history
|
||||
""")
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
stats.update(dict(row))
|
||||
|
||||
# Recent jobs
|
||||
cursor.execute("""
|
||||
SELECT target, status, duration_seconds, posts_scraped, started_at
|
||||
FROM job_history ORDER BY started_at DESC LIMIT 10
|
||||
""")
|
||||
stats['recent_jobs'] = [dict(row) for row in cursor.fetchall()]
|
||||
|
||||
conn.close()
|
||||
return stats
|
||||
|
||||
def print_job_history(limit=20):
|
||||
"""Pretty print job history."""
|
||||
jobs = get_job_history(limit)
|
||||
|
||||
print("\n📋 Job History")
|
||||
print("-" * 80)
|
||||
print(f"{'ID':<10} {'Target':<15} {'Status':<10} {'Posts':<8} {'Duration':<10} {'Started':<20}")
|
||||
print("-" * 80)
|
||||
|
||||
for job in jobs:
|
||||
status_icon = "✅" if job['status'] == 'completed' else "❌" if job['status'] == 'failed' else "🔄"
|
||||
duration = f"{job['duration_seconds']:.1f}s" if job['duration_seconds'] else "-"
|
||||
started = job['started_at'][:19] if job['started_at'] else "-"
|
||||
dry = " (dry)" if job['dry_run'] else ""
|
||||
|
||||
print(f"{status_icon} {job['job_id']:<8} {job['target']:<15} {job['status']:<10} "
|
||||
f"{job['posts_scraped']:<8} {duration:<10} {started}{dry}")
|
||||
|
||||
print("-" * 80)
|
||||
|
||||
stats = get_job_stats()
|
||||
success_rate = (stats['completed'] / stats['total_jobs'] * 100) if stats['total_jobs'] else 0
|
||||
print(f"\n📊 Stats: {stats['total_jobs']} jobs | {success_rate:.0f}% success | "
|
||||
f"{stats['total_posts'] or 0} posts total")
|
||||
|
||||
# --- SQLITE MAINTENANCE FUNCTIONS ---
|
||||
|
||||
def enable_auto_vacuum():
|
||||
"""Enable incremental auto-vacuum on SQLite database."""
|
||||
conn = get_connection()
|
||||
try:
|
||||
conn.execute("PRAGMA auto_vacuum = INCREMENTAL")
|
||||
conn.execute("PRAGMA incremental_vacuum")
|
||||
conn.commit()
|
||||
print("✅ Auto-vacuum enabled")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def vacuum_database():
|
||||
"""Run VACUUM to optimize and compact the database."""
|
||||
conn = get_connection()
|
||||
try:
|
||||
print("🔧 Running VACUUM...")
|
||||
conn.execute("VACUUM")
|
||||
print("✅ Database optimized")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def backup_database(backup_path=None):
|
||||
"""
|
||||
Create a backup of the SQLite database.
|
||||
|
||||
Args:
|
||||
backup_path: Optional custom backup path
|
||||
|
||||
Returns:
|
||||
Path to the backup file
|
||||
"""
|
||||
import shutil
|
||||
|
||||
backup_dir = DATA_DIR / "backups"
|
||||
backup_dir.mkdir(exist_ok=True)
|
||||
|
||||
if backup_path is None:
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
backup_path = backup_dir / f"reddit_scraper_{timestamp}.db"
|
||||
|
||||
shutil.copy2(DB_PATH, backup_path)
|
||||
|
||||
# Get file size
|
||||
size_mb = Path(backup_path).stat().st_size / (1024 * 1024)
|
||||
print(f"✅ Backup created: {backup_path} ({size_mb:.2f} MB)")
|
||||
|
||||
return str(backup_path)
|
||||
|
||||
def get_database_info():
|
||||
"""Get database size and table info."""
|
||||
info = {}
|
||||
|
||||
# File size
|
||||
if DB_PATH.exists():
|
||||
info['size_mb'] = DB_PATH.stat().st_size / (1024 * 1024)
|
||||
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Table counts
|
||||
tables = ['posts', 'comments', 'job_history', 'alerts', 'subreddits']
|
||||
info['tables'] = {}
|
||||
|
||||
for table in tables:
|
||||
try:
|
||||
cursor.execute(f"SELECT COUNT(*) FROM {table}")
|
||||
info['tables'][table] = cursor.fetchone()[0]
|
||||
except:
|
||||
info['tables'][table] = 0
|
||||
|
||||
conn.close()
|
||||
return info
|
||||
|
||||
# Initialize on import
|
||||
init_database()
|
||||
|
||||
|
|
|
|||
181
export/parquet.py
Normal file
181
export/parquet.py
Normal file
|
|
@ -0,0 +1,181 @@
|
|||
"""
|
||||
Parquet Export Module - For DuckDB/Warehouse integration
|
||||
Export scraped data to Parquet format for analytics tools.
|
||||
"""
|
||||
import pandas as pd
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
def export_to_parquet(subreddit, output_dir=None, prefix="r"):
|
||||
"""
|
||||
Export subreddit data to Parquet format.
|
||||
|
||||
Args:
|
||||
subreddit: Subreddit name
|
||||
output_dir: Output directory (default: data/parquet)
|
||||
prefix: "r" for subreddit, "u" for user
|
||||
|
||||
Returns:
|
||||
Dictionary with paths to exported files
|
||||
"""
|
||||
try:
|
||||
import pyarrow
|
||||
except ImportError:
|
||||
raise ImportError("pyarrow required for Parquet export. Run: pip install pyarrow")
|
||||
|
||||
# Setup paths
|
||||
data_dir = Path(f"data/{prefix}_{subreddit}")
|
||||
output_path = Path(output_dir) if output_dir else Path("data/parquet")
|
||||
output_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if not data_dir.exists():
|
||||
print(f"❌ No data found for {prefix}/{subreddit}")
|
||||
return {}
|
||||
|
||||
exported = {}
|
||||
timestamp = datetime.now().strftime("%Y%m%d")
|
||||
|
||||
# Export posts
|
||||
posts_csv = data_dir / "posts.csv"
|
||||
if posts_csv.exists():
|
||||
print(f"📦 Converting posts to Parquet...")
|
||||
df = pd.read_csv(posts_csv)
|
||||
|
||||
# Convert datetime columns
|
||||
if 'created_utc' in df.columns:
|
||||
df['created_utc'] = pd.to_datetime(df['created_utc'], errors='coerce')
|
||||
|
||||
# Optimize dtypes
|
||||
for col in ['score', 'num_comments', 'num_crossposts', 'total_awards']:
|
||||
if col in df.columns:
|
||||
df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0).astype('int32')
|
||||
|
||||
for col in ['is_nsfw', 'is_spoiler', 'has_media', 'media_downloaded']:
|
||||
if col in df.columns:
|
||||
df[col] = df[col].astype(bool)
|
||||
|
||||
output_file = output_path / f"{subreddit}_posts_{timestamp}.parquet"
|
||||
df.to_parquet(output_file, engine="pyarrow", compression="snappy")
|
||||
|
||||
size_mb = output_file.stat().st_size / (1024 * 1024)
|
||||
print(f" ✅ {output_file.name} ({len(df)} rows, {size_mb:.2f} MB)")
|
||||
exported['posts'] = str(output_file)
|
||||
|
||||
# Export comments
|
||||
comments_csv = data_dir / "comments.csv"
|
||||
if comments_csv.exists():
|
||||
print(f"📦 Converting comments to Parquet...")
|
||||
df = pd.read_csv(comments_csv)
|
||||
|
||||
if 'created_utc' in df.columns:
|
||||
df['created_utc'] = pd.to_datetime(df['created_utc'], errors='coerce')
|
||||
|
||||
if 'score' in df.columns:
|
||||
df['score'] = pd.to_numeric(df['score'], errors='coerce').fillna(0).astype('int32')
|
||||
|
||||
output_file = output_path / f"{subreddit}_comments_{timestamp}.parquet"
|
||||
df.to_parquet(output_file, engine="pyarrow", compression="snappy")
|
||||
|
||||
size_mb = output_file.stat().st_size / (1024 * 1024)
|
||||
print(f" ✅ {output_file.name} ({len(df)} rows, {size_mb:.2f} MB)")
|
||||
exported['comments'] = str(output_file)
|
||||
|
||||
print(f"\n✅ Export complete! Files saved to: {output_path}")
|
||||
print(f" 💡 Query with DuckDB: duckdb.query(\"SELECT * FROM '{exported.get('posts', '')}' LIMIT 10\")")
|
||||
|
||||
return exported
|
||||
|
||||
|
||||
def export_database_to_parquet(output_dir=None):
|
||||
"""
|
||||
Export entire SQLite database to Parquet files.
|
||||
|
||||
Args:
|
||||
output_dir: Output directory
|
||||
|
||||
Returns:
|
||||
Dictionary with paths to exported files
|
||||
"""
|
||||
try:
|
||||
import pyarrow
|
||||
except ImportError:
|
||||
raise ImportError("pyarrow required. Run: pip install pyarrow")
|
||||
|
||||
from export.database import get_connection
|
||||
|
||||
output_path = Path(output_dir) if output_dir else Path("data/parquet")
|
||||
output_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
conn = get_connection()
|
||||
exported = {}
|
||||
timestamp = datetime.now().strftime("%Y%m%d")
|
||||
|
||||
tables = ['posts', 'comments', 'job_history']
|
||||
|
||||
for table in tables:
|
||||
try:
|
||||
print(f"📦 Exporting {table}...")
|
||||
df = pd.read_sql(f"SELECT * FROM {table}", conn)
|
||||
|
||||
if len(df) > 0:
|
||||
output_file = output_path / f"db_{table}_{timestamp}.parquet"
|
||||
df.to_parquet(output_file, engine="pyarrow", compression="snappy")
|
||||
|
||||
size_mb = output_file.stat().st_size / (1024 * 1024)
|
||||
print(f" ✅ {output_file.name} ({len(df)} rows, {size_mb:.2f} MB)")
|
||||
exported[table] = str(output_file)
|
||||
else:
|
||||
print(f" ⏭️ {table} is empty, skipping")
|
||||
except Exception as e:
|
||||
print(f" ❌ Failed to export {table}: {e}")
|
||||
|
||||
conn.close()
|
||||
return exported
|
||||
|
||||
|
||||
def list_parquet_files(directory="data/parquet"):
|
||||
"""List all Parquet files in directory."""
|
||||
parquet_dir = Path(directory)
|
||||
|
||||
if not parquet_dir.exists():
|
||||
print(f"📁 No Parquet directory found at {directory}")
|
||||
return []
|
||||
|
||||
files = list(parquet_dir.glob("*.parquet"))
|
||||
|
||||
print(f"\n📁 Parquet Files in {directory}:")
|
||||
print("-" * 60)
|
||||
|
||||
for f in files:
|
||||
size_mb = f.stat().st_size / (1024 * 1024)
|
||||
mtime = datetime.fromtimestamp(f.stat().st_mtime).strftime("%Y-%m-%d %H:%M")
|
||||
print(f" {f.name:<40} {size_mb:>6.2f} MB {mtime}")
|
||||
|
||||
print("-" * 60)
|
||||
print(f"Total: {len(files)} files")
|
||||
|
||||
return [str(f) for f in files]
|
||||
|
||||
|
||||
# CLI for testing
|
||||
if __name__ == "__main__":
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description="Parquet Export")
|
||||
parser.add_argument("subreddit", nargs='?', help="Subreddit to export")
|
||||
parser.add_argument("--user", action="store_true", help="Is a user profile")
|
||||
parser.add_argument("--output", type=str, help="Output directory")
|
||||
parser.add_argument("--database", action="store_true", help="Export entire database")
|
||||
parser.add_argument("--list", action="store_true", help="List Parquet files")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.list:
|
||||
list_parquet_files()
|
||||
elif args.database:
|
||||
export_database_to_parquet(args.output)
|
||||
elif args.subreddit:
|
||||
prefix = "u" if args.user else "r"
|
||||
export_to_parquet(args.subreddit, args.output, prefix)
|
||||
else:
|
||||
parser.print_help()
|
||||
339
main.py
339
main.py
|
|
@ -291,15 +291,45 @@ def extract_post_data(post_json):
|
|||
}
|
||||
|
||||
# --- FULL HISTORY SCRAPE ---
|
||||
def run_full_history(target, limit, is_user=False, download_media_flag=True, scrape_comments_flag=True):
|
||||
"""Full scrape with images, videos, and comments."""
|
||||
def run_full_history(target, limit, is_user=False, download_media_flag=True,
|
||||
scrape_comments_flag=True, dry_run=False, use_plugins=False):
|
||||
"""
|
||||
Full scrape with images, videos, and comments.
|
||||
|
||||
Args:
|
||||
target: Subreddit or username
|
||||
limit: Maximum posts to scrape
|
||||
is_user: True if target is a user
|
||||
download_media_flag: Download images/videos
|
||||
scrape_comments_flag: Scrape comments
|
||||
dry_run: Simulate without saving data
|
||||
use_plugins: Run post-processing plugins
|
||||
"""
|
||||
prefix = "u" if is_user else "r"
|
||||
print(f"🚀 Starting FULL HISTORY scrape for {prefix}/{target}")
|
||||
mode = "full" if download_media_flag and scrape_comments_flag else "history"
|
||||
|
||||
# Display mode banner
|
||||
if dry_run:
|
||||
print("=" * 50)
|
||||
print("🧪 DRY RUN MODE - No data will be saved")
|
||||
print("=" * 50)
|
||||
|
||||
print(f"🚀 Starting {'DRY RUN' if dry_run else 'FULL HISTORY'} scrape for {prefix}/{target}")
|
||||
print(f" 📊 Target posts: {limit}")
|
||||
print(f" 🖼️ Download media: {download_media_flag}")
|
||||
print(f" 🖼️ Download media: {download_media_flag and not dry_run}")
|
||||
print(f" 💬 Scrape comments: {scrape_comments_flag}")
|
||||
print(f" 🔌 Plugins enabled: {use_plugins}")
|
||||
print("-" * 50)
|
||||
|
||||
# Start job tracking
|
||||
job_id = None
|
||||
try:
|
||||
from export.database import start_job_record, complete_job_record
|
||||
job_id = start_job_record(target, mode, is_user, dry_run)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Job tracking unavailable: {e}")
|
||||
|
||||
# Setup directories (even for dry run, to check existing data)
|
||||
dirs = setup_directories(target, prefix)
|
||||
load_history(dirs["posts"])
|
||||
|
||||
|
|
@ -307,97 +337,152 @@ def run_full_history(target, limit, is_user=False, download_media_flag=True, scr
|
|||
total_posts = 0
|
||||
total_media = {"images": 0, "videos": 0}
|
||||
total_comments = 0
|
||||
all_scraped_posts = [] # For plugin processing
|
||||
all_scraped_comments = []
|
||||
start_time = time.time()
|
||||
error_msg = None
|
||||
|
||||
while total_posts < limit:
|
||||
random.shuffle(MIRRORS)
|
||||
success = False
|
||||
|
||||
for base_url in MIRRORS:
|
||||
try:
|
||||
if is_user:
|
||||
path = f"/user/{target}/submitted.json"
|
||||
else:
|
||||
path = f"/r/{target}/new.json"
|
||||
|
||||
target_url = f"{base_url}{path}?limit=100&raw_json=1"
|
||||
if after:
|
||||
target_url += f"&after={after}"
|
||||
|
||||
print(f"\n📡 Fetching from: {base_url}")
|
||||
response = SESSION.get(target_url, timeout=15)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
posts = []
|
||||
all_comments = []
|
||||
|
||||
children = data['data']['children']
|
||||
print(f" Found {len(children)} posts in this batch")
|
||||
|
||||
for child in children:
|
||||
p = child['data']
|
||||
post = extract_post_data(p)
|
||||
|
||||
if post['permalink'] in SEEN_URLS:
|
||||
continue
|
||||
|
||||
if download_media_flag:
|
||||
downloaded = download_post_media(p, dirs, post['id'])
|
||||
post['media_downloaded'] = downloaded['images'] > 0 or downloaded['videos'] > 0
|
||||
total_media['images'] += downloaded['images']
|
||||
total_media['videos'] += downloaded['videos']
|
||||
|
||||
posts.append(post)
|
||||
|
||||
if scrape_comments_flag and post['num_comments'] > 0:
|
||||
print(f" 💬 Fetching comments for: {post['title'][:40]}...")
|
||||
comments = scrape_comments(post['permalink'])
|
||||
all_comments.extend(comments)
|
||||
total_comments += len(comments)
|
||||
time.sleep(1)
|
||||
|
||||
saved = save_posts_csv(posts, dirs["posts"])
|
||||
total_posts += saved
|
||||
|
||||
if all_comments:
|
||||
save_comments_csv(all_comments, dirs["comments"])
|
||||
|
||||
print(f"\n📊 Progress: {total_posts}/{limit} posts")
|
||||
print(f" 🖼️ Images: {total_media['images']} | 🎬 Videos: {total_media['videos']}")
|
||||
print(f" 💬 Comments: {total_comments}")
|
||||
|
||||
after = data['data'].get('after')
|
||||
if not after:
|
||||
print("\n🏁 Reached end of available history.")
|
||||
break
|
||||
|
||||
success = True
|
||||
break
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Error with {base_url}: {e}")
|
||||
continue
|
||||
|
||||
if not after:
|
||||
break
|
||||
try:
|
||||
while total_posts < limit:
|
||||
random.shuffle(MIRRORS)
|
||||
success = False
|
||||
|
||||
if not success:
|
||||
print("\n❌ All sources failed. Waiting 30s...")
|
||||
time.sleep(30)
|
||||
else:
|
||||
print(f"\n⏸️ Cooling down (3s)...")
|
||||
time.sleep(3)
|
||||
for base_url in MIRRORS:
|
||||
try:
|
||||
if is_user:
|
||||
path = f"/user/{target}/submitted.json"
|
||||
else:
|
||||
path = f"/r/{target}/new.json"
|
||||
|
||||
target_url = f"{base_url}{path}?limit=100&raw_json=1"
|
||||
if after:
|
||||
target_url += f"&after={after}"
|
||||
|
||||
print(f"\n📡 Fetching from: {base_url}")
|
||||
response = SESSION.get(target_url, timeout=15)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
posts = []
|
||||
batch_comments = []
|
||||
|
||||
children = data['data']['children']
|
||||
print(f" Found {len(children)} posts in this batch")
|
||||
|
||||
for child in children:
|
||||
p = child['data']
|
||||
post = extract_post_data(p)
|
||||
|
||||
if post['permalink'] in SEEN_URLS:
|
||||
continue
|
||||
|
||||
# Download media (skip in dry run)
|
||||
if download_media_flag and not dry_run:
|
||||
downloaded = download_post_media(p, dirs, post['id'])
|
||||
post['media_downloaded'] = downloaded['images'] > 0 or downloaded['videos'] > 0
|
||||
total_media['images'] += downloaded['images']
|
||||
total_media['videos'] += downloaded['videos']
|
||||
|
||||
posts.append(post)
|
||||
|
||||
# Scrape comments
|
||||
if scrape_comments_flag and post['num_comments'] > 0:
|
||||
print(f" 💬 Fetching comments for: {post['title'][:40]}...")
|
||||
comments = scrape_comments(post['permalink'])
|
||||
batch_comments.extend(comments)
|
||||
total_comments += len(comments)
|
||||
time.sleep(1)
|
||||
|
||||
# Collect for plugins
|
||||
all_scraped_posts.extend(posts)
|
||||
all_scraped_comments.extend(batch_comments)
|
||||
|
||||
# Save data (skip in dry run)
|
||||
if not dry_run:
|
||||
saved = save_posts_csv(posts, dirs["posts"])
|
||||
total_posts += saved
|
||||
|
||||
if batch_comments:
|
||||
save_comments_csv(batch_comments, dirs["comments"])
|
||||
else:
|
||||
# In dry run, just count
|
||||
total_posts += len(posts)
|
||||
print(f" 🧪 [DRY RUN] Would save {len(posts)} posts")
|
||||
|
||||
print(f"\n📊 Progress: {total_posts}/{limit} posts")
|
||||
print(f" 🖼️ Images: {total_media['images']} | 🎬 Videos: {total_media['videos']}")
|
||||
print(f" 💬 Comments: {total_comments}")
|
||||
|
||||
after = data['data'].get('after')
|
||||
if not after:
|
||||
print("\n🏁 Reached end of available history.")
|
||||
break
|
||||
|
||||
success = True
|
||||
break
|
||||
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Error with {base_url}: {e}")
|
||||
continue
|
||||
|
||||
if not after:
|
||||
break
|
||||
|
||||
if not success:
|
||||
print("\n❌ All sources failed. Waiting 30s...")
|
||||
time.sleep(30)
|
||||
else:
|
||||
print(f"\n⏸️ Cooling down (3s)...")
|
||||
time.sleep(3)
|
||||
|
||||
# Run plugins on collected data
|
||||
if use_plugins and (all_scraped_posts or all_scraped_comments):
|
||||
print("\n🔌 Running post-processing plugins...")
|
||||
try:
|
||||
from plugins import load_plugins, run_plugins
|
||||
plugins = load_plugins()
|
||||
if plugins:
|
||||
all_scraped_posts, all_scraped_comments = run_plugins(
|
||||
all_scraped_posts, all_scraped_comments, plugins
|
||||
)
|
||||
print(f" ✅ Processed {len(all_scraped_posts)} posts with {len(plugins)} plugins")
|
||||
else:
|
||||
print(" ⚠️ No plugins found")
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Plugin error: {e}")
|
||||
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
print(f"\n❌ Scrape error: {e}")
|
||||
|
||||
duration = time.time() - start_time
|
||||
|
||||
# Complete job tracking
|
||||
if job_id:
|
||||
try:
|
||||
status = 'failed' if error_msg else 'completed'
|
||||
complete_job_record(
|
||||
job_id, status,
|
||||
total_posts, total_comments,
|
||||
total_media['images'] + total_media['videos'],
|
||||
error_msg
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Failed to complete job record: {e}")
|
||||
|
||||
# Summary
|
||||
print("\n" + "=" * 50)
|
||||
print("✅ SCRAPE COMPLETE!")
|
||||
print(f" 📁 Data saved to: {dirs['base']}")
|
||||
print(f" 📊 Total posts: {total_posts}")
|
||||
print(f" 🖼️ Total images: {total_media['images']}")
|
||||
print(f" 🎬 Total videos: {total_media['videos']}")
|
||||
print(f" 💬 Total comments: {total_comments}")
|
||||
if dry_run:
|
||||
print("🧪 DRY RUN COMPLETE!")
|
||||
print(f" 📊 Would scrape: {total_posts} posts")
|
||||
print(f" 💬 Would scrape: {total_comments} comments")
|
||||
else:
|
||||
print("✅ SCRAPE COMPLETE!")
|
||||
print(f" 📁 Data saved to: {dirs['base']}")
|
||||
print(f" 📊 Total posts: {total_posts}")
|
||||
print(f" 🖼️ Total images: {total_media['images']}")
|
||||
print(f" 🎬 Total videos: {total_media['videos']}")
|
||||
print(f" 💬 Total comments: {total_comments}")
|
||||
print(f" ⏱️ Duration: {duration:.1f}s")
|
||||
|
||||
return {
|
||||
|
|
@ -405,7 +490,9 @@ def run_full_history(target, limit, is_user=False, download_media_flag=True, scr
|
|||
'images': total_media['images'],
|
||||
'videos': total_media['videos'],
|
||||
'comments': total_comments,
|
||||
'duration': f"{duration:.1f}s"
|
||||
'duration': f"{duration:.1f}s",
|
||||
'dry_run': dry_run,
|
||||
'job_id': job_id
|
||||
}
|
||||
|
||||
# --- MONITOR MODE ---
|
||||
|
|
@ -470,6 +557,8 @@ Commands:
|
|||
python main.py <target> --mode full --limit 100
|
||||
python main.py <target> --mode history --limit 500
|
||||
python main.py <target> --mode monitor
|
||||
python main.py <target> --dry-run # Test without saving
|
||||
python main.py <target> --plugins # Enable post-processing
|
||||
|
||||
SEARCH:
|
||||
python main.py --search "keyword" --subreddit delhi
|
||||
|
|
@ -484,6 +573,16 @@ Commands:
|
|||
ANALYTICS:
|
||||
python main.py --analyze delhi --sentiment
|
||||
python main.py --analyze delhi --keywords
|
||||
|
||||
MAINTENANCE:
|
||||
python main.py --job-history # View job history
|
||||
python main.py --backup # Backup database
|
||||
python main.py --vacuum # Optimize database
|
||||
python main.py --export-parquet python # Export to Parquet
|
||||
python main.py --list-plugins # List available plugins
|
||||
|
||||
REST API:
|
||||
python main.py --api # Start REST API server
|
||||
"""
|
||||
)
|
||||
|
||||
|
|
@ -519,6 +618,16 @@ Commands:
|
|||
parser.add_argument("--telegram-token", type=str, help="Telegram bot token")
|
||||
parser.add_argument("--telegram-chat", type=str, help="Telegram chat ID")
|
||||
|
||||
# New: Observability & Maintenance
|
||||
parser.add_argument("--dry-run", action="store_true", help="Simulate scrape without saving data")
|
||||
parser.add_argument("--plugins", action="store_true", help="Enable post-processing plugins")
|
||||
parser.add_argument("--list-plugins", action="store_true", help="List available plugins")
|
||||
parser.add_argument("--job-history", action="store_true", help="View job history")
|
||||
parser.add_argument("--backup", action="store_true", help="Backup SQLite database")
|
||||
parser.add_argument("--vacuum", action="store_true", help="Optimize SQLite database")
|
||||
parser.add_argument("--export-parquet", type=str, help="Export subreddit to Parquet format")
|
||||
parser.add_argument("--api", action="store_true", help="Start REST API server (port 8000)")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
print("=" * 50)
|
||||
|
|
@ -532,6 +641,52 @@ Commands:
|
|||
os.system("streamlit run dashboard/app.py")
|
||||
return
|
||||
|
||||
# REST API mode
|
||||
if args.api:
|
||||
print("\n🚀 Starting REST API server...")
|
||||
print(" 📖 Docs: http://localhost:8000/docs")
|
||||
print(" 📊 Connect Metabase/Grafana to http://localhost:8000")
|
||||
try:
|
||||
import uvicorn
|
||||
from api.server import app
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
except ImportError:
|
||||
print("❌ Install dependencies: pip install fastapi uvicorn")
|
||||
return
|
||||
|
||||
# --- NEW: Maintenance & Observability Commands ---
|
||||
|
||||
# Job history
|
||||
if args.job_history:
|
||||
from export.database import print_job_history
|
||||
print_job_history()
|
||||
return
|
||||
|
||||
# Backup database
|
||||
if args.backup:
|
||||
from export.database import backup_database
|
||||
backup_database()
|
||||
return
|
||||
|
||||
# Vacuum/optimize database
|
||||
if args.vacuum:
|
||||
from export.database import vacuum_database
|
||||
vacuum_database()
|
||||
return
|
||||
|
||||
# Export to Parquet
|
||||
if args.export_parquet:
|
||||
from export.parquet import export_to_parquet
|
||||
prefix = "u" if args.user else "r"
|
||||
export_to_parquet(args.export_parquet, prefix=prefix)
|
||||
return
|
||||
|
||||
# List plugins
|
||||
if args.list_plugins:
|
||||
from plugins import list_plugins
|
||||
list_plugins()
|
||||
return
|
||||
|
||||
# Search mode
|
||||
if args.search:
|
||||
print(f"\n🔍 Searching for: {args.search}")
|
||||
|
|
@ -607,11 +762,13 @@ Commands:
|
|||
time.sleep(300)
|
||||
elif args.mode == "history":
|
||||
run_full_history(args.target, args.limit, args.user,
|
||||
download_media_flag=False, scrape_comments_flag=False)
|
||||
download_media_flag=False, scrape_comments_flag=False,
|
||||
dry_run=args.dry_run, use_plugins=args.plugins)
|
||||
else:
|
||||
run_full_history(args.target, args.limit, args.user,
|
||||
download_media_flag=not args.no_media,
|
||||
scrape_comments_flag=not args.no_comments)
|
||||
scrape_comments_flag=not args.no_comments,
|
||||
dry_run=args.dry_run, use_plugins=args.plugins)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
149
plugins/__init__.py
Normal file
149
plugins/__init__.py
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
"""
|
||||
Lightweight Plugin System for Post-Processing
|
||||
Plugins can process posts and comments after scraping.
|
||||
"""
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
import importlib.util
|
||||
import sys
|
||||
|
||||
|
||||
class Plugin(ABC):
|
||||
"""
|
||||
Base class for scraper plugins.
|
||||
|
||||
To create a plugin:
|
||||
1. Create a new .py file in the plugins/ directory
|
||||
2. Create a class that inherits from Plugin
|
||||
3. Implement the process_posts() method
|
||||
4. Optionally implement process_comments()
|
||||
|
||||
Example:
|
||||
class MyPlugin(Plugin):
|
||||
name = "my_plugin"
|
||||
description = "Does something cool"
|
||||
|
||||
def process_posts(self, posts):
|
||||
for post in posts:
|
||||
post['processed'] = True
|
||||
return posts
|
||||
"""
|
||||
name = "base"
|
||||
description = "Base plugin"
|
||||
enabled = True
|
||||
|
||||
@abstractmethod
|
||||
def process_posts(self, posts: list) -> list:
|
||||
"""
|
||||
Process posts after scraping.
|
||||
|
||||
Args:
|
||||
posts: List of post dictionaries
|
||||
|
||||
Returns:
|
||||
Modified list of posts
|
||||
"""
|
||||
pass
|
||||
|
||||
def process_comments(self, comments: list) -> list:
|
||||
"""
|
||||
Process comments after scraping (optional).
|
||||
|
||||
Args:
|
||||
comments: List of comment dictionaries
|
||||
|
||||
Returns:
|
||||
Modified list of comments
|
||||
"""
|
||||
return comments
|
||||
|
||||
def __repr__(self):
|
||||
return f"<Plugin: {self.name}>"
|
||||
|
||||
|
||||
def load_plugins(plugin_dir=None):
|
||||
"""
|
||||
Load all plugins from the plugins directory.
|
||||
|
||||
Args:
|
||||
plugin_dir: Path to plugins directory
|
||||
|
||||
Returns:
|
||||
List of plugin instances
|
||||
"""
|
||||
if plugin_dir is None:
|
||||
plugin_dir = Path(__file__).parent
|
||||
else:
|
||||
plugin_dir = Path(plugin_dir)
|
||||
|
||||
plugins = []
|
||||
|
||||
for file in plugin_dir.glob("*.py"):
|
||||
# Skip __init__.py and base files
|
||||
if file.name.startswith("_"):
|
||||
continue
|
||||
|
||||
try:
|
||||
# Load the module
|
||||
spec = importlib.util.spec_from_file_location(file.stem, file)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[file.stem] = module
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
# Find Plugin subclasses
|
||||
for attr_name in dir(module):
|
||||
attr = getattr(module, attr_name)
|
||||
if (isinstance(attr, type) and
|
||||
issubclass(attr, Plugin) and
|
||||
attr != Plugin and
|
||||
hasattr(attr, 'name')):
|
||||
|
||||
plugin_instance = attr()
|
||||
if plugin_instance.enabled:
|
||||
plugins.append(plugin_instance)
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Failed to load plugin {file.name}: {e}")
|
||||
|
||||
return plugins
|
||||
|
||||
|
||||
def run_plugins(posts, comments, plugins):
|
||||
"""
|
||||
Run all plugins on scraped data.
|
||||
|
||||
Args:
|
||||
posts: List of posts
|
||||
comments: List of comments
|
||||
plugins: List of plugin instances
|
||||
|
||||
Returns:
|
||||
Tuple of (processed_posts, processed_comments)
|
||||
"""
|
||||
for plugin in plugins:
|
||||
try:
|
||||
print(f"🔌 Running plugin: {plugin.name}")
|
||||
posts = plugin.process_posts(posts)
|
||||
comments = plugin.process_comments(comments)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Plugin {plugin.name} failed: {e}")
|
||||
|
||||
return posts, comments
|
||||
|
||||
|
||||
def list_plugins(plugin_dir=None):
|
||||
"""List all available plugins."""
|
||||
plugins = load_plugins(plugin_dir)
|
||||
|
||||
print("\n🔌 Available Plugins:")
|
||||
print("-" * 50)
|
||||
|
||||
if not plugins:
|
||||
print(" No plugins found")
|
||||
else:
|
||||
for plugin in plugins:
|
||||
status = "✅" if plugin.enabled else "❌"
|
||||
print(f" {status} {plugin.name:<20} {plugin.description}")
|
||||
|
||||
print("-" * 50)
|
||||
return plugins
|
||||
45
plugins/deduplicator.py
Normal file
45
plugins/deduplicator.py
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
"""
|
||||
Deduplicator Plugin
|
||||
Removes duplicate posts based on permalink.
|
||||
"""
|
||||
from plugins import Plugin
|
||||
|
||||
|
||||
class Deduplicator(Plugin):
|
||||
"""Remove duplicate posts by permalink."""
|
||||
|
||||
name = "deduplicator"
|
||||
description = "Removes duplicate posts by permalink"
|
||||
enabled = True
|
||||
|
||||
def process_posts(self, posts):
|
||||
"""Remove duplicate posts."""
|
||||
seen = set()
|
||||
unique = []
|
||||
duplicates = 0
|
||||
|
||||
for post in posts:
|
||||
key = post.get('permalink')
|
||||
if key and key not in seen:
|
||||
seen.add(key)
|
||||
unique.append(post)
|
||||
else:
|
||||
duplicates += 1
|
||||
|
||||
if duplicates > 0:
|
||||
print(f" 🔄 Removed {duplicates} duplicate posts")
|
||||
|
||||
return unique
|
||||
|
||||
def process_comments(self, comments):
|
||||
"""Remove duplicate comments."""
|
||||
seen = set()
|
||||
unique = []
|
||||
|
||||
for comment in comments:
|
||||
key = comment.get('comment_id')
|
||||
if key and key not in seen:
|
||||
seen.add(key)
|
||||
unique.append(comment)
|
||||
|
||||
return unique
|
||||
35
plugins/keyword_extractor.py
Normal file
35
plugins/keyword_extractor.py
Normal file
|
|
@ -0,0 +1,35 @@
|
|||
"""
|
||||
Keyword Extractor Plugin
|
||||
Extracts and tags posts with top keywords.
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from plugins import Plugin
|
||||
from analytics.sentiment import extract_keywords
|
||||
|
||||
|
||||
class KeywordExtractor(Plugin):
|
||||
"""Extract and add keywords to posts."""
|
||||
|
||||
name = "keyword_extractor"
|
||||
description = "Adds top keywords to each post"
|
||||
enabled = True
|
||||
top_n = 5 # Number of keywords per post
|
||||
|
||||
def process_posts(self, posts):
|
||||
"""Add keywords to each post."""
|
||||
for post in posts:
|
||||
text = f"{post.get('title', '')} {post.get('selftext', '')}"
|
||||
keywords = extract_keywords([text], top_n=self.top_n)
|
||||
post['keywords'] = ','.join([kw for kw, count in keywords])
|
||||
|
||||
# Also extract global keywords
|
||||
all_texts = [f"{p.get('title', '')} {p.get('selftext', '')}" for p in posts]
|
||||
global_keywords = extract_keywords(all_texts, top_n=10)
|
||||
|
||||
print(f" 🏷️ Top keywords: {', '.join([kw for kw, _ in global_keywords[:5]])}")
|
||||
|
||||
return posts
|
||||
45
plugins/sentiment_tagger.py
Normal file
45
plugins/sentiment_tagger.py
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
"""
|
||||
Sentiment Tagger Plugin
|
||||
Adds sentiment scores and labels to posts and comments.
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add parent to path for imports
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from plugins import Plugin
|
||||
from analytics.sentiment import analyze_sentiment
|
||||
|
||||
|
||||
class SentimentTagger(Plugin):
|
||||
"""Add sentiment analysis to scraped content."""
|
||||
|
||||
name = "sentiment_tagger"
|
||||
description = "Adds sentiment scores and labels to posts"
|
||||
enabled = True
|
||||
|
||||
def process_posts(self, posts):
|
||||
"""Add sentiment to posts."""
|
||||
for post in posts:
|
||||
text = f"{post.get('title', '')} {post.get('selftext', '')}"
|
||||
score, label = analyze_sentiment(text)
|
||||
post['sentiment_score'] = score
|
||||
post['sentiment_label'] = label
|
||||
|
||||
# Count sentiments
|
||||
pos = sum(1 for p in posts if p.get('sentiment_label') == 'positive')
|
||||
neg = sum(1 for p in posts if p.get('sentiment_label') == 'negative')
|
||||
neu = len(posts) - pos - neg
|
||||
|
||||
print(f" 📊 Sentiment: {pos} positive, {neu} neutral, {neg} negative")
|
||||
return posts
|
||||
|
||||
def process_comments(self, comments):
|
||||
"""Add sentiment to comments."""
|
||||
for comment in comments:
|
||||
score, label = analyze_sentiment(comment.get('body', ''))
|
||||
comment['sentiment_score'] = score
|
||||
comment['sentiment_label'] = label
|
||||
|
||||
return comments
|
||||
|
|
@ -11,3 +11,8 @@ streamlit
|
|||
|
||||
# Export
|
||||
openpyxl
|
||||
pyarrow
|
||||
|
||||
# REST API
|
||||
fastapi
|
||||
uvicorn
|
||||
|
|
|
|||
Loading…
Reference in a new issue