diff --git a/Dockerfile b/Dockerfile index 39a0129..7442298 100644 --- a/Dockerfile +++ b/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"] diff --git a/README.md b/README.md index ed6b126..45f9a51 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ [![Docker Build & Publish](https://github.com/ksanjeev284/reddit-universal-scraper/actions/workflows/docker-publish.yml/badge.svg)](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!** image @@ -11,14 +11,18 @@ A **full-featured** Reddit scraper suite with analytics dashboard, sentiment ana | 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 @@ -26,22 +30,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) @@ -51,46 +58,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 @@ -99,51 +155,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 @@ -156,9 +286,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 | @@ -168,31 +296,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. diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..fe99578 --- /dev/null +++ b/api/__init__.py @@ -0,0 +1,2 @@ +"""Reddit Scraper REST API""" +from .server import app diff --git a/api/server.py b/api/server.py new file mode 100644 index 0000000..de70f16 --- /dev/null +++ b/api/server.py @@ -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) diff --git a/dashboard/app.py b/dashboard/app.py index 8e16830..41304fc 100644 --- a/dashboard/app.py +++ b/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 --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 + +# Backup database +python main.py --backup + +# Scrape with plugins +python main.py --plugins + +# Dry run (test without saving) +python main.py --dry-run + """, language="bash") if __name__ == "__main__": main() diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..0dfcc93 --- /dev/null +++ b/docker-compose.yml @@ -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 +# 3. docker-compose up -d +# 4. Open firewall: ports 8000, 8501 +# +# Access: +# http://:8501 (Dashboard) +# http://:8000 (API) +# =========================================== diff --git a/docs/INTEGRATION.md b/docs/INTEGRATION.md new file mode 100644 index 0000000..09fb588 --- /dev/null +++ b/docs/INTEGRATION.md @@ -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" +``` diff --git a/export/database.py b/export/database.py index 4a1f504..c216716 100644 --- a/export/database.py +++ b/export/database.py @@ -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() + diff --git a/export/parquet.py b/export/parquet.py new file mode 100644 index 0000000..628b1de --- /dev/null +++ b/export/parquet.py @@ -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() diff --git a/main.py b/main.py index a02f964..a2eb991 100644 --- a/main.py +++ b/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 --mode full --limit 100 python main.py --mode history --limit 500 python main.py --mode monitor + python main.py --dry-run # Test without saving + python main.py --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() diff --git a/plugins/__init__.py b/plugins/__init__.py new file mode 100644 index 0000000..3552839 --- /dev/null +++ b/plugins/__init__.py @@ -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"" + + +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 diff --git a/plugins/deduplicator.py b/plugins/deduplicator.py new file mode 100644 index 0000000..9cfa4b4 --- /dev/null +++ b/plugins/deduplicator.py @@ -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 diff --git a/plugins/keyword_extractor.py b/plugins/keyword_extractor.py new file mode 100644 index 0000000..5288b47 --- /dev/null +++ b/plugins/keyword_extractor.py @@ -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 diff --git a/plugins/sentiment_tagger.py b/plugins/sentiment_tagger.py new file mode 100644 index 0000000..061bc99 --- /dev/null +++ b/plugins/sentiment_tagger.py @@ -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 diff --git a/requirements.txt b/requirements.txt index b3ed494..99a4b61 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,3 +11,8 @@ streamlit # Export openpyxl +pyarrow + +# REST API +fastapi +uvicorn