path: root/test_moderator.py

#!/usr/bin/env python3
"""Test CLI for the intelligent roleplay moderator."""

import argparse
import json
import os
import random
import sys
from pathlib import Path
from typing import Dict, Any, List
from dataclasses import dataclass
from openai import OpenAI
from dotenv import load_dotenv
import praw

# Import the base class
sys.path.append('src')
from umabot.rules.intelligent_moderator_base import IntelligentModeratorBase


@dataclass
class MockSubmission:
    """Mock Reddit submission for testing."""
    id: str
    title: str
    selftext: str
    url: str
    is_video: bool
    is_self: bool
    is_gallery: bool
    link_flair_text: str
    author: 'MockAuthor'
    
    @property
    def permalink(self) -> str:
        return f"/r/test/comments/{self.id}/"


@dataclass
class MockAuthor:
    """Mock Reddit author for testing."""
    name: str
    
    def message(self, subject: str, message: str) -> None:
        print(f"\n📧 MOD MAIL TO u/{self.name}")
        print(f"Subject: {subject}")
        print(f"Message:\n{message}")


class RedditDownloader:
    """Downloads posts from Reddit for testing."""
    
    def __init__(self, config):
        """Initialize Reddit client."""
        self.reddit = praw.Reddit(
            client_id=config.client_id,
            client_secret=config.client_secret,
            username=config.username,
            password=config.password,
            user_agent=config.user_agent
        )
        self.subreddit_name = config.subreddit_name
    
    def download_roleplay_posts(self, limit: int = 50, output_dir: Path = None) -> List[Path]:
        """Download roleplay posts from the subreddit."""
        if output_dir is None:
            output_dir = Path("real_test_posts")
        
        output_dir.mkdir(exist_ok=True)
        
        subreddit = self.reddit.subreddit(self.subreddit_name)
        downloaded_files = []
        
        print(f"Downloading roleplay posts from r/{self.subreddit_name}...")
        
        count = 0
        for submission in subreddit.new(limit=limit * 3):  # Get more to filter
            if count >= limit:
                break
                
            # Skip removed posts
            if submission.removed_by_category or submission.selftext == "[removed]" or submission.selftext == "[deleted]":
                continue
                
            # Check if it's a roleplay post by flair template ID
            is_roleplay = False
            if hasattr(submission, 'link_flair_template_id'):
                is_roleplay = submission.link_flair_template_id == "311f0024-8302-11f0-9b41-46c005ad843c"
            elif hasattr(submission, 'link_flair_text'):
                is_roleplay = submission.link_flair_text and submission.link_flair_text.lower() == "roleplay"
                
            if is_roleplay:
                # Create filename
                safe_title = "".join(c for c in submission.title if c.isalnum() or c in (' ', '-', '_')).rstrip()
                safe_title = safe_title[:50]  # Limit length
                filename = f"{submission.id}_{safe_title}.txt"
                filepath = output_dir / filename
                
                # Write post content
                content = f"Title: {submission.title}\n\n"
                if submission.selftext:
                    content += submission.selftext
                else:
                    content += f"[Link Post: {submission.url}]"
                
                # Add flair info for debugging
                if hasattr(submission, 'link_flair_text') and submission.link_flair_text:
                    content += f"\n\nFlair: {submission.link_flair_text}"
                if hasattr(submission, 'link_flair_template_id') and submission.link_flair_template_id:
                    content += f"\nFlair Template ID: {submission.link_flair_template_id}"
                
                with open(filepath, 'w', encoding='utf-8') as f:
                    f.write(content)
                
                downloaded_files.append(filepath)
                count += 1
                print(f"Downloaded: {filename}")
        
        print(f"Downloaded {len(downloaded_files)} roleplay posts")
        return downloaded_files
    
    def download_random_posts(self, limit: int = 20, output_dir: Path = None) -> List[Path]:
        """Download random posts from the subreddit."""
        if output_dir is None:
            output_dir = Path("real_test_posts")
        
        output_dir.mkdir(exist_ok=True)
        
        subreddit = self.reddit.subreddit(self.subreddit_name)
        downloaded_files = []
        
        print(f"Downloading random posts from r/{self.subreddit_name}...")
        
        count = 0
        for submission in subreddit.hot(limit=limit * 2):  # Get more to filter
            if count >= limit:
                break
            
            # Skip stickied posts
            if submission.stickied:
                continue
                
            # Skip removed posts
            if submission.removed_by_category or submission.selftext == "[removed]" or submission.selftext == "[deleted]":
                continue
                
            # Create filename
            safe_title = "".join(c for c in submission.title if c.isalnum() or c in (' ', '-', '_')).rstrip()
            safe_title = safe_title[:50]  # Limit length
            filename = f"{submission.id}_{safe_title}.txt"
            filepath = output_dir / filename
            
            # Write post content
            content = f"Title: {submission.title}\n\n"
            if submission.selftext:
                content += submission.selftext
            else:
                content += f"[Link Post: {submission.url}]"
            
            # Add flair info
            if hasattr(submission, 'link_flair_text') and submission.link_flair_text:
                content += f"\n\nFlair: {submission.link_flair_text}"
            
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(content)
            
            downloaded_files.append(filepath)
            count += 1
            print(f"Downloaded: {filename}")
        
        print(f"Downloaded {len(downloaded_files)} random posts")
        return downloaded_files


class TestIntelligentModerator(IntelligentModeratorBase):
    """Test version of the intelligent roleplay moderator."""
    
    def __init__(self, openai_api_key: str):
        """Initialize the test moderator."""
        super().__init__(openai_api_key)
    
    def test_file(self, file_path: Path, author_name: str = "testuser") -> Dict[str, Any]:
        """Test a single file against the moderator."""
        try:
            # Read file content
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read().strip()
            
            # Create mock submission
            submission = MockSubmission(
                id=f"test_{file_path.stem}",
                title=file_path.stem.replace('_', ' ').title(),
                selftext=content,
                url="",
                is_video=False,
                is_self=True,
                is_gallery=False,
                link_flair_text="Roleplay",
                author=MockAuthor(author_name)
            )
            
            # Evaluate the post
            evaluation = self.evaluate_post(submission)
            
            # Determine actions (same logic as production moderator)
            actions = self.determine_actions(evaluation)
            
            return {
                "file": str(file_path),
                "title": submission.title,
                "content_preview": content[:200] + "..." if len(content) > 200 else content,
                "word_count": len(content.split()),
                "has_media": self._has_media(submission),
                "media_type": self._get_media_type(submission),
                "evaluation": evaluation,
                "actions": actions,
                "success": True
            }
            
        except Exception as e:
            return {
                "file": str(file_path),
                "error": str(e),
                "success": False
            }
    
    def test_directory(self, directory_path: Path, author_name: str = "testuser", random_count: int = None) -> List[Dict[str, Any]]:
        """Test all text files in a directory."""
        results = []
        
        # Find all text files
        text_files = list(directory_path.glob("*.txt"))
        
        if not text_files:
            print(f"No .txt files found in {directory_path}")
            return results
        
        # Random selection if requested
        if random_count and random_count < len(text_files):
            text_files = random.sample(text_files, random_count)
            print(f"Randomly selected {random_count} files from {len(list(directory_path.glob('*.txt')))} available files")
        
        print(f"Testing {len(text_files)} text files...")
        
        for file_path in text_files:
            print(f"Testing {file_path.name}...")
            result = self.test_file(file_path, author_name)
            results.append(result)
        
        return results
    
    # Abstract method implementations for IntelligentModeratorBase
    def _get_submission_id(self, submission: MockSubmission) -> str:
        """Get the submission ID."""
        return submission.id
    
    def _get_title(self, submission: MockSubmission) -> str:
        """Get the submission title."""
        return submission.title or ""
    
    def _get_content(self, submission: MockSubmission) -> str:
        """Get the submission content."""
        return submission.selftext or ""
    
    def _get_url(self, submission: MockSubmission) -> str:
        """Get the submission URL."""
        return submission.url or ""
    
    def _is_video(self, submission: MockSubmission) -> bool:
        """Check if submission is a video."""
        return submission.is_video
    
    def _is_self(self, submission: MockSubmission) -> bool:
        """Check if submission is a self post."""
        return submission.is_self
    
    def _is_gallery(self, submission: MockSubmission) -> bool:
        """Check if submission is a gallery."""
        return submission.is_gallery
    
    def _log_info(self, message: str) -> None:
        """Log info message."""
        print(f"INFO: {message}")
    
    def _log_warning(self, message: str) -> None:
        """Log warning message."""
        print(f"WARNING: {message}")
    
    def _log_error(self, message: str) -> None:
        """Log error message."""
        print(f"ERROR: {message}")


def print_results(results: List[Dict[str, Any]]) -> None:
    """Print test results in a formatted way."""
    print("\n" + "="*80)
    print("TEST RESULTS")
    print("="*80)
    
    for i, result in enumerate(results, 1):
        print(f"\n{i}. {result['file']}")
        print("-" * 60)
        
        if not result["success"]:
            print(f"❌ ERROR: {result['error']}")
            continue
        
        print(f"Title: {result['title']}")
        print(f"Word Count: {result['word_count']}")
        print(f"Has Media: {result['has_media']}")
        print(f"Media Type: {result['media_type']}")
        print(f"Content Preview: {result['content_preview']}")
        
        evaluation = result["evaluation"]
        print(f"\n🤖 AI Evaluation:")
        print(f"  Should be Art: {evaluation['should_be_art']}")
        print(f"  Is Low Effort: {evaluation['is_low_effort']}")
        print(f"  Confidence: {evaluation['confidence']:.2f}")
        print(f"  Reasoning: {evaluation['reasoning']}")
        
        actions = result["actions"]
        print(f"\n📋 Actions:")
        for action in actions:
            if action == "CHANGE_FLAIR_TO_ART":
                print("  🎨 Change flair to Art")
            elif action == "REMOVE_POST":
                print("  🗑️  Remove post (low effort)")
            elif action == "ALLOW_POST":
                print("  ✅ Allow post")
        
        # Show what mod mail would be sent
        if "CHANGE_FLAIR_TO_ART" in actions:
            print(f"\n📧 Mod Mail (Art Flair Change):")
            print(f"  Subject: Your post flair has been changed to Art")
            print(f"  Message: Your roleplay post has been automatically re-flaired as 'Art' because it appears to be primarily showcasing artwork or visual content rather than roleplay.")
            print(f"  Reasoning: {evaluation['reasoning']}")
        
        if "REMOVE_POST" in actions:
            print(f"\n📧 Mod Mail (Low Effort Removal):")
            print(f"  Subject: Your roleplay post has been removed for low effort")
            print(f"  Message: Your roleplay post has been removed because it was determined to be low effort content.")
            print(f"  Reasoning: {evaluation['reasoning']}")


def main():
    """Main CLI function."""
    # Load environment variables from .env file
    load_dotenv()
    
    parser = argparse.ArgumentParser(
        description="Test the intelligent roleplay moderator against text files",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Test a single file
  python test_moderator.py --file sample_post.txt
  
  # Test all files in a directory
  python test_moderator.py --directory test_posts/
  
  # Test 5 random files from a directory
  python test_moderator.py --directory real_test_posts/ --random 5
  
  # Download 20 roleplay posts from Reddit
  python test_moderator.py --download-roleplay 20
  
  # Download 10 random posts from Reddit
  python test_moderator.py --download-random 10
  
  # Download posts then test them
  python test_moderator.py --download-roleplay 15
  python test_moderator.py --directory real_test_posts/ --random 5
        """
    )
    
    parser.add_argument(
        "--file", "-f",
        type=Path,
        help="Test a single text file"
    )
    
    parser.add_argument(
        "--directory", "-d",
        type=Path,
        help="Test all .txt files in a directory"
    )
    
    parser.add_argument(
        "--author", "-a",
        default="testuser",
        help="Author name for mock submissions (default: testuser)"
    )
    
    parser.add_argument(
        "--api-key", "-k",
        help="OpenAI API key (or set OPENAI_API_KEY environment variable)"
    )
    
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Show detailed output"
    )
    
    parser.add_argument(
        "--random", "-r",
        type=int,
        help="Randomly select N files from directory for testing"
    )
    
    parser.add_argument(
        "--download-roleplay",
        type=int,
        metavar="COUNT",
        help="Download N roleplay posts from Reddit for testing"
    )
    
    parser.add_argument(
        "--download-random",
        type=int,
        metavar="COUNT", 
        help="Download N random posts from Reddit for testing"
    )
    
    args = parser.parse_args()
    
    # Get API key (from .env file, environment variable, or command line)
    api_key = args.api_key or os.getenv("OPENAI_API_KEY")
    if not api_key:
        print("❌ Error: OpenAI API key required")
        print("Set OPENAI_API_KEY in .env file, environment variable, or use --api-key")
        sys.exit(1)
    
    # Handle download commands first
    if args.download_roleplay or args.download_random:
        # Load Reddit config
        try:
            from src.umabot.config import Config
            config = Config.from_env()
            config.validate()
        except Exception as e:
            print(f"❌ Error loading Reddit config: {e}")
            print("Make sure your .env file has all required Reddit credentials")
            sys.exit(1)
        
        downloader = RedditDownloader(config)
        
        if args.download_roleplay:
            downloader.download_roleplay_posts(args.download_roleplay)
        
        if args.download_random:
            downloader.download_random_posts(args.download_random)
        
        print("✅ Download complete!")
        return
    
    # Validate inputs for testing
    if not args.file and not args.directory:
        print("❌ Error: Must specify either --file or --directory")
        parser.print_help()
        sys.exit(1)
    
    if args.file and not args.file.exists():
        print(f"❌ Error: File {args.file} does not exist")
        sys.exit(1)
    
    if args.directory and not args.directory.exists():
        print(f"❌ Error: Directory {args.directory} does not exist")
        sys.exit(1)
    
    # Initialize moderator
    try:
        moderator = TestIntelligentModerator(api_key)
    except Exception as e:
        print(f"❌ Error initializing moderator: {e}")
        sys.exit(1)
    
    # Run tests
    results = []
    
    if args.file:
        print(f"Testing single file: {args.file}")
        result = moderator.test_file(args.file, args.author)
        results.append(result)
    
    if args.directory:
        print(f"Testing directory: {args.directory}")
        dir_results = moderator.test_directory(args.directory, args.author, args.random)
        results.extend(dir_results)
    
    # Print results
    print_results(results)
    
    # Summary
    total = len(results)
    successful = sum(1 for r in results if r["success"])
    errors = total - successful
    
    print(f"\n" + "="*80)
    print(f"SUMMARY: {successful}/{total} tests completed successfully")
    if errors > 0:
        print(f"Errors: {errors}")
    print("="*80)


if __name__ == "__main__":
    main()