Building Your AI Reddit Ghostwriter: Complete Technical Guide with Code Examples

# Core Dependencies
FastAPI          # API framework
PostgreSQL       # Database for posts/responses
Redis            # Caching and job queues
PRAW             # Reddit API wrapper
OpenAI           # AI response generation
Celery           # Background task processing

# Create virtual environment
python -m venv reddit-ai
source reddit-ai/bin/activate  # Windows: reddit-ai\Scripts\activate

# Install packages
pip install fastapi uvicorn praw openai sqlalchemy psycopg2-binary
pip install redis celery python-dotenv pydantic httpx
pip install prometheus-client pytest

# Development tools
pip install black flake8 mypy

# Reddit API (get from reddit.com/prefs/apps)
REDDIT_CLIENT_ID=your_client_id
REDDIT_CLIENT_SECRET=your_client_secret
REDDIT_USER_AGENT=YourBot/1.0
REDDIT_USERNAME=your_username
REDDIT_PASSWORD=your_password

# AI Provider
OPENAI_API_KEY=your_openai_key
OPENAI_MODEL=gpt-4o-mini

# Database
DATABASE_URL=postgresql://user:pass@localhost:5432/reddit_ghostwriter
REDIS_URL=redis://localhost:6379/0

# Settings
MAX_RESPONSES_PER_HOUR=20
MIN_COMMENT_KARMA=50
DEBUG=True

reddit-ghostwriter/
├── app/
│   ├── __init__.py
│   ├── config.py          # Configuration management
│   ├── database.py        # Database models
│   ├── reddit_monitor.py  # Reddit API integration
│   ├── ai_generator.py    # AI response creation
│   ├── quality_control.py # Response filtering
│   └── main.py           # FastAPI app
├── tests/                # Unit tests
├── requirements.txt      # Dependencies
├── docker-compose.yml    # Local development
└── .env                 # Environment variables

# app/reddit_monitor.py
import praw
import logging
from datetime import datetime, timezone
from typing import List, Dict, Optional
from app.config import settings
from app.database import Post, Session

logger = logging.getLogger(__name__)

class RedditMonitor:
    def __init__(self):
        """Initialize Reddit API connection"""
        self.reddit = praw.Reddit(
            client_id=settings.reddit_client_id,
            client_secret=settings.reddit_client_secret,
            user_agent=settings.reddit_user_agent,
            username=settings.reddit_username,
            password=settings.reddit_password
        )

    def monitor_keywords(self, keywords: List[str], limit: int = 25) -> List[Dict]:
        """Search Reddit for specific keywords"""
        posts = []

        for keyword in keywords:
            try:
                # Search across all subreddits
                for submission in self.reddit.subreddit("all").search(
                    keyword,
                    sort="new",
                    time_filter="day",
                    limit=limit
                ):
                    post_data = self._extract_post_data(submission)
                    if self._is_worth_processing(post_data):
                        posts.append(post_data)

            except Exception as e:
                logger.error(f"Error searching keyword '{keyword}': {e}")
                continue

        return posts

    def monitor_subreddits(self, subreddit_names: List[str], limit: int = 50) -> List[Dict]:
        """Monitor specific subreddits for new posts"""
        posts = []

        for subreddit_name in subreddit_names:
            try:
                subreddit = self.reddit.subreddit(subreddit_name)

                # Get new posts
                for submission in subreddit.new(limit=limit//2):
                    post_data = self._extract_post_data(submission)
                    if self._is_worth_processing(post_data):
                        posts.append(post_data)

                # Get hot posts (high engagement)
                for submission in subreddit.hot(limit=limit//2):
                    post_data = self._extract_post_data(submission)
                    if self._is_worth_processing(post_data):
                        posts.append(post_data)

            except Exception as e:
                logger.error(f"Error monitoring r/{subreddit_name}: {e}")
                continue

        return posts

    def _extract_post_data(self, submission) -> Dict:
        """Extract relevant data from Reddit submission"""
        return {
            'reddit_id': submission.id,
            'subreddit': submission.subreddit.display_name,
            'title': submission.title,
            'content': submission.selftext or "",
            'author': submission.author.name if submission.author else "[deleted]",
            'created_utc': datetime.fromtimestamp(submission.created_utc, tz=timezone.utc),
            'score': submission.score,
            'num_comments': submission.num_comments,
            'url': submission.url
        }

    def _is_worth_processing(self, post_data: Dict) -> bool:
        """Basic filtering for post quality"""
        # Skip deleted authors
        if post_data['author'] == '[deleted]':
            return False

        # Skip our own posts
        if post_data['author'] == settings.reddit_username:
            return False

        # Skip posts that are too old
        post_age = (datetime.now(timezone.utc) - post_data['created_utc']).hours
        if post_age > 24:  # Only process posts from last 24 hours
            return False

        return True

    def get_post_comments(self, reddit_id: str, limit: int = 10) -> List[Dict]:
        """Get comments for a post (for context)"""
        try:
            submission = self.reddit.submission(id=reddit_id)
            submission.comments.replace_more(limit=0)

            comments = []
            for comment in submission.comments[:limit]:
                if comment.author:
                    comments.append({
                        'author': comment.author.name,
                        'body': comment.body,
                        'score': comment.score,
                        'created_utc': comment.created_utc
                    })
            return comments

        except Exception as e:
            logger.error(f"Error getting comments for {reddit_id}: {e}")
            return []

# app/database.py
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, Boolean, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import func
from app.config import settings

engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    reddit_id = Column(String, unique=True, index=True)
    subreddit = Column(String, index=True)
    title = Column(String)
    content = Column(Text)
    author = Column(String)
    created_utc = Column(DateTime)
    score = Column(Integer)
    num_comments = Column(Integer)
    url = Column(String)

    # Analysis fields
    relevance_score = Column(Float)
    sentiment = Column(String)
    processed = Column(Boolean, default=False)
    responded = Column(Boolean, default=False)

    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

class Response(Base):
    __tablename__ = "responses"

    id = Column(Integer, primary_key=True, index=True)
    post_id = Column(Integer, index=True)
    response_text = Column(Text)
    confidence_score = Column(Float)
    strategy_type = Column(String)  # helper, storyteller, problem_solver

    status = Column(String, default="pending")  # pending, approved, posted, rejected
    posted_at = Column(DateTime, nullable=True)
    reddit_comment_id = Column(String, nullable=True)

    created_at = Column(DateTime, server_default=func.now())

# Create tables
Base.metadata.create_all(bind=engine)

# app/content_analyzer.py
import openai
from typing import List, Dict
from app.config import settings

openai.api_key = settings.openai_api_key

class ContentAnalyzer:
    def __init__(self):
        self.target_keywords = [
            "project management",
            "CRM software",
            "email marketing",
            "team collaboration"
        ]

    def calculate_relevance_score(self, post_data: Dict) -> float:
        """Calculate how relevant a post is to our target keywords"""
        text = f"{post_data['title']} {post_data['content']}".lower()

        # Keyword matching
        keyword_score = 0
        for keyword in self.target_keywords:
            if keyword.lower() in text:
                keyword_score += 1

        keyword_score = keyword_score / len(self.target_keywords)

        # Engagement score (higher engagement = more visibility)
        engagement_score = min((post_data['score'] + post_data['num_comments']) / 100, 1.0)

        # Combined relevance (60% keywords, 40% engagement)
        return (keyword_score * 0.6) + (engagement_score * 0.4)

    def analyze_sentiment(self, post_data: Dict) -> str:
        """Analyze post sentiment using AI"""
        try:
            prompt = f"""
            Analyze the sentiment of this Reddit post. Respond with only one word: positive, negative, or neutral.

            Title: {post_data['title']}
            Content: {post_data['content']}

            Consider: Is the user happy, frustrated, or neutral? Are they seeking help?

            Response:"""

            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=10,
                temperature=0.1
            )

            sentiment = response.choices[0].message.content.strip().lower()
            return sentiment if sentiment in ['positive', 'negative', 'neutral'] else 'neutral'

        except Exception as e:
            logger.error(f"Error analyzing sentiment: {e}")
            return 'neutral'

    def should_respond(self, post_data: Dict, relevance_score: float) -> bool:
        """Determine if we should respond to this post"""
        # Skip if relevance is too low
        if relevance_score < 0.3:
            return False

        # Skip certain subreddits with strict moderation
        strict_subreddits = ['askreddit', 'news', 'todayilearned']
        if post_data['subreddit'].lower() in strict_subreddits:
            return False

        # Skip if post is too controversial (high comment to upvote ratio)
        if post_data['num_comments'] > post_data['score'] * 2:
            return False

        return True

# app/ai_generator.py
import openai
import random
from typing import Dict, List, Optional
from app.config import settings

class AIResponseGenerator:
    def __init__(self):
        openai.api_key = settings.openai_api_key

    def generate_response(self, post_data: Dict, comments: List[Dict] = None) -> Dict:
        """Generate an appropriate response for a Reddit post"""

        # Choose response strategy
        strategy = self._select_strategy(post_data)

        # Build the prompt
        prompt = self._build_prompt(post_data, strategy, comments)

        try:
            # Generate response
            response = openai.ChatCompletion.create(
                model=settings.openai_model,
                messages=[
                    {"role": "system", "content": self._get_system_prompt()},
                    {"role": "user", "content": prompt}
                ],
                max_tokens=400,
                temperature=0.7
            )

            response_text = response.choices[0].message.content.strip()
            confidence = self._calculate_confidence(response_text, post_data)

            return {
                'response_text': response_text,
                'confidence_score': confidence,
                'strategy_type': strategy,
                'status': 'pending'
            }

        except Exception as e:
            logger.error(f"Error generating response: {e}")
            return None

    def _select_strategy(self, post_data: Dict) -> str:
        """Choose appropriate response strategy based on post content"""
        content = f"{post_data['title']} {post_data['content']}".lower()

        # Help-seeking posts
        help_keywords = ['help', 'need', 'struggling', 'problem', 'issue', 'how to']
        if any(word in content for word in help_keywords):
            return 'helper'

        # Experience requests
        experience_keywords = ['experience', 'thoughts', 'review', 'opinion']
        if any(word in content for word in experience_keywords):
            return 'storyteller'

        # Technical questions
        tech_keywords = ['how', 'why', 'what', 'which', 'best way']
        if any(word in content for word in tech_keywords):
            return 'problem_solver'

        return 'helper'  # Default

    def _get_system_prompt(self) -> str:
        """System prompt that defines AI behavior"""
        return """You are a helpful Reddit community member who provides genuine value.

Guidelines:
1. Always be helpful and provide real value first
2. Match the tone of the subreddit (professional for business subs, casual for others)
3. Be conversational and authentic, never corporate or salesy
4. Only mention products/services when genuinely relevant (less than 20% of responses)
5. Include specific, actionable advice
6. Use Reddit formatting when appropriate
7. Keep responses under 300 words

Avoid:
- Obvious sales pitches
- Generic templated responses
- Corporate jargon or marketing speak
- Overpromising or making unrealistic claims"""

    def _build_prompt(self, post_data: Dict, strategy: str, comments: List[Dict]) -> str:
        """Build specific prompt for this post"""

        base_prompt = f"""
        Subreddit: r/{post_data['subreddit']}
        Post Title: {post_data['title']}
        Post Content: {post_data['content']}
        Upvotes: {post_data['score']} | Comments: {post_data['num_comments']}
        """

        if comments:
            base_prompt += "\nRecent Comments:\n"
            for comment in comments[:3]:
                base_prompt += f"- {comment['body'][:100]}...\n"

        strategy_prompts = {
            'helper': """
            Strategy: Be a helpful community member providing valuable advice.

            Approach:
            1. Acknowledge their specific situation
            2. Provide 2-3 concrete, actionable suggestions
            3. If genuinely relevant, naturally mention a helpful tool/resource
            4. End encouragingly or offer further help

            Write a helpful Reddit comment:
            """,

            'storyteller': """
            Strategy: Share a relevant personal experience.

            Approach:
            1. Start with "I had a similar situation..." or similar
            2. Share a brief, relevant story
            3. Explain what worked or didn't work
            4. Mention specific tools/approaches you used (if relevant)
            5. Relate back to their situation

            Write an engaging Reddit comment:
            """,

            'problem_solver': """
            Strategy: Directly solve their technical question.

            Approach:
            1. Directly answer their question
            2. Provide step-by-step guidance if appropriate
            3. Mention alternative approaches
            4. Suggest relevant tools/resources for their specific issue
            5. Offer to clarify if they have follow-ups

            Write a solution-focused Reddit comment:
            """
        }

        return base_prompt + strategy_prompts.get(strategy, strategy_prompts['helper'])

    def _calculate_confidence(self, response_text: str, post_data: Dict) -> float:
        """Calculate confidence score for the response"""
        score = 0.5  # Base confidence

        # Length check
        if 50 <= len(response_text) <= 300:
            score += 0.2

        # Keyword relevance
        post_words = set(f"{post_data['title']} {post_data['content']}".lower().split())
        response_words = set(response_text.lower().split())
        overlap = len(post_words.intersection(response_words))

        if overlap > 2:
            score += 0.2

        # Avoid overly promotional language
        promo_words = ['amazing', 'revolutionary', 'game-changer', 'incredible']
        if not any(word in response_text.lower() for word in promo_words):
            score += 0.1

        return min(score, 1.0)

# app/tasks.py
from celery import Celery
from app.reddit_monitor import RedditMonitor
from app.content_analyzer import ContentAnalyzer
from app.ai_generator import AIResponseGenerator
from app.quality_control import QualityController
from app.engagement_manager import EngagementManager
from app.database import SessionLocal, Post, Response
from app.config import settings

# Initialize Celery
celery = Celery('reddit-ghostwriter')
celery.conf.broker_url = settings.redis_url
celery.conf.result_backend = settings.redis_url

@celery.task
def monitor_and_analyze():
    """Main monitoring task"""
    monitor = RedditMonitor()
    analyzer = ContentAnalyzer()

    # Monitor keywords and subreddits
    target_keywords = ["project management", "CRM software", "team collaboration"]
    target_subreddits = ["entrepreneur", "smallbusiness", "productivity"]

    posts = []
    posts.extend(monitor.monitor_keywords(target_keywords))
    posts.extend(monitor.monitor_subreddits(target_subreddits))

    db = SessionLocal()

    for post_data in posts:
        # Skip if already processed
        if db.query(Post).filter(Post.reddit_id == post_data['reddit_id']).first():
            continue

        # Calculate relevance
        relevance_score = analyzer.calculate_relevance_score(post_data)
        sentiment = analyzer.analyze_sentiment(post_data)

        # Create post record
        post = Post(
            reddit_id=post_data['reddit_id'],
            subreddit=post_data['subreddit'],
            title=post_data['title'],
            content=post_data['content'],
            author=post_data['author'],
            created_utc=post_data['created_utc'],
            score=post_data['score'],
            num_comments=post_data['num_comments'],
            url=post_data['url'],
            relevance_score=relevance_score,
            sentiment=sentiment,
            processed=True
        )

        db.add(post)
        db.commit()

        # Queue for response if relevant
        if analyzer.should_respond(post_data, relevance_score):
            generate_response.delay(post.id)

    db.close()

@celery.task
def generate_response(post_id: int):
    """Generate AI response for a post"""
    db = SessionLocal()
    generator = AIResponseGenerator()
    quality_control = QualityController()

    post = db.query(Post).filter(Post.id == post_id).first()
    if not post:
        return

    # Get additional context
    monitor = RedditMonitor()
    comments = monitor.get_post_comments(post.reddit_id)

    # Generate response
    post_data = {
        'subreddit': post.subreddit,
        'title': post.title,
        'content': post.content,
        'score': post.score,
        'num_comments': post.num_comments
    }

    response_data = generator.generate_response(post_data, comments)
    if not response_data:
        return

    # Create response record
    response = Response(
        post_id=post_id,
        response_text=response_data['response_text'],
        confidence_score=response_data['confidence_score'],
        strategy_type=response_data['strategy_type'],
        status='pending'
    )

    db.add(response)
    db.commit()

    # Quality check
    evaluation = quality_control.evaluate_response(response_data, post_data)

    if evaluation['approved']:
        # Queue for posting (with small delay to seem natural)
        post_response.apply_async(args=[response.id], countdown=300)  # 5 min delay
    else:
        response.status = 'rejected'
        db.commit()

    db.close()

# Schedule periodic tasks
from celery.schedules import crontab

celery.conf.beat_schedule = {
    'monitor-reddit': {
        'task': 'app.tasks.monitor_and_analyze',
        'schedule': crontab(minute='*/15'),  # Every 15 minutes
    },
}

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/reddit_ghostwriter
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis

  worker:
    build: .
    command: celery -A app.tasks worker --loglevel=info
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/reddit_ghostwriter
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis

  scheduler:
    build: .
    command: celery -A app.tasks beat --loglevel=info
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/reddit_ghostwriter
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis

  db:
    image: postgres:13
    environment:
      - POSTGRES_DB=reddit_ghostwriter
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:6-alpine

volumes:
  postgres_data:

# Start services
docker-compose up -d db redis

# Run migrations
python -c "from app.database import Base, engine; Base.metadata.create_all(engine)"

# Start API server
uvicorn app.main:app --reload

# Start worker (separate terminal)
celery -A app.tasks worker --loglevel=info

# Start scheduler (separate terminal)
celery -A app.tasks beat --loglevel=info

# Start all services
docker-compose up -d

# Check logs
docker-compose logs -f app
docker-compose logs -f worker

# app/main.py
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from app.database import SessionLocal, Post, Response

app = FastAPI(title="Reddit AI Ghostwriter")

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/stats")
def get_stats(db: Session = Depends(get_db)):
    """Get system statistics"""
    return {
        "total_posts": db.query(Post).count(),
        "processed_posts": db.query(Post).filter(Post.processed == True).count(),
        "generated_responses": db.query(Response).count(),
        "posted_responses": db.query(Response).filter(Response.status == "posted").count(),
    }

@app.get("/posts")
def get_recent_posts(limit: int = 20, db: Session = Depends(get_db)):
    """Get recent posts"""
    posts = db.query(Post).order_by(Post.created_at.desc()).limit(limit).all()
    return [{"id": p.id, "title": p.title, "subreddit": p.subreddit,
             "relevance_score": p.relevance_score, "responded": p.responded}
            for p in posts]

# app/rate_limiter.py
from datetime import datetime, timedelta
from typing import Dict

class RateLimiter:
    def __init__(self):
        self.requests = {}

    def can_make_request(self, key: str, max_requests: int = 60, window_minutes: int = 1) -> bool:
        """Check if we can make a request within rate limits"""
        now = datetime.now()
        window_start = now - timedelta(minutes=window_minutes)

        # Clean old requests
        if key in self.requests:
            self.requests[key] = [
                req_time for req_time in self.requests[key]
                if req_time > window_start
            ]
        else:
            self.requests[key] = []

        # Check if we can make request
        if len(self.requests[key]) < max_requests:
            self.requests[key].append(now)
            return True

        return False

# app/error_handler.py
import logging
import traceback
from functools import wraps

def handle_errors(logger):
    """Decorator for handling and logging errors"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.error(f"Error in {func.__name__}: {str(e)}")
                logger.error(f"Traceback: {traceback.format_exc()}")
                return None
        return wrapper
    return decorator