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, reasoning_level: int = 2):
        """Initialize the test moderator."""
        super().__init__(openai_api_key)
        self.reasoning_level = reasoning_level
    
    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()
            
            # Parse file content to extract metadata
            lines = content.split('\n')
            title = file_path.stem.replace('_', ' ').title()
            body = content
            is_video = False
            is_gallery = False
            
            # Check if file contains metadata markers
            if "Title:" in content and "Body:" in content:
                # Parse structured content
                title_line = None
                body_start = None
                for i, line in enumerate(lines):
                    if line.startswith("Title:"):
                        title_line = line
                    elif line.startswith("Body:"):
                        body_start = i
                        break
                
                if title_line:
                    title = title_line.replace("Title:", "").strip()
                
                if body_start is not None:
                    body_lines = lines[body_start + 1:]
                    # Remove metadata lines
                    body_lines = [line for line in body_lines if not line.startswith(("Has Media:", "Media Type:"))]
                    body = '\n'.join(body_lines).strip()
            
            # Check for video indicators in content
            content_lower = content.lower()
            if "video" in content_lower or "mp4" in content_lower or "webm" in content_lower:
                is_video = True
            
            # Create mock submission
            submission = MockSubmission(
                id=f"test_{file_path.stem}",
                title=title,
                selftext=body,
                url="",
                is_video=is_video,
                is_self=True,
                is_gallery=is_gallery,
                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
    
    def _format_reasoning(self, evaluation: dict, is_art_flair_change: bool = False) -> str:
        """Format reasoning based on the configured reasoning level."""
        # Never include reasoning for art flair changes
        if is_art_flair_change:
            return ""
            
        original_reasoning = evaluation.get('reasoning', '')
        
        if self.reasoning_level == 0:
            # No reasoning included
            return ""
        elif self.reasoning_level == 1:
            # Brief reasoning - extract key points
            return self._extract_brief_reasoning(original_reasoning)
        else:
            # Full reasoning (default)
            return original_reasoning
    
    def _extract_brief_reasoning(self, reasoning: str) -> str:
        """Extract brief key points from the full reasoning."""
        # Convert to lowercase for easier processing
        reasoning_lower = reasoning.lower()
        
        # Extract key issues
        issues = []
        
        # Check for common low-effort indicators
        if any(word in reasoning_lower for word in ['short', 'brief', 'minimal', 'little']):
            issues.append('short')
        if any(word in reasoning_lower for word in ['word count', 'words', 'length']):
            issues.append('low word count')
        if any(word in reasoning_lower for word in ['plot', 'story', 'narrative']):
            if 'no plot' in reasoning_lower or 'lack' in reasoning_lower:
                issues.append('no plot development')
        if any(word in reasoning_lower for word in ['effort', 'substance', 'content']):
            if 'low' in reasoning_lower or 'lack' in reasoning_lower:
                issues.append('low-effort')
        if any(word in reasoning_lower for word in ['creativity', 'originality']):
            if 'lack' in reasoning_lower or 'no' in reasoning_lower:
                issues.append('lacks creativity')
        if any(word in reasoning_lower for word in ['artwork', 'art', 'visual', 'image']):
            if 'primarily' in reasoning_lower or 'showcasing' in reasoning_lower:
                issues.append('artwork showcase')
        
        # If no specific issues found, return a generic brief message
        if not issues:
            return "Content quality below standards"
        
        # Return comma-separated list of issues
        return ", ".join(issues)
    
    # 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]], pause: bool = False, moderator: TestIntelligentModerator = None) -> 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']}")
            if pause and i < len(results):
                input("\n⏸️  Press Enter to continue to next result...")
            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 in-character post has been automatically re-flaired as 'Art' because it appears to be primarily showcasing artwork or visual content rather than in-character content.")
            
            # Format reasoning based on moderator's reasoning level (never include for art flair changes)
            if moderator:
                formatted_reasoning = moderator._format_reasoning(evaluation, is_art_flair_change=True)
                if formatted_reasoning:
                    print(f"  Reasoning: {formatted_reasoning}")
            else:
                # For art flair changes, never show reasoning even without moderator
                pass
        
        if "REMOVE_POST" in actions:
            print(f"\n📧 Mod Mail (Low Effort Removal):")
            print(f"  Subject: Your in-character post has been removed for low-effort")
            print(f"  Message: Your in-character post has been removed because it was determined to be low-effort content.")
            
            # Format reasoning based on moderator's reasoning level
            if moderator:
                formatted_reasoning = moderator._format_reasoning(evaluation)
                if formatted_reasoning:
                    print(f"  Reasoning: {formatted_reasoning}")
            else:
                print(f"  Reasoning: {evaluation['reasoning']}")
            
            print(f"  Discord Recommendation: If you enjoy active roleplay, join the official Discord server! It features a comprehensive layout of channels, forums, roles, bots, events, and more!")
        
        # Add pause between results if requested
        if pause and i < len(results):
            input("\n⏸️  Press Enter to continue to next result...")


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
  
  # Test with pause between results
  python test_moderator.py --directory real_test_posts/ --random 3 --pause
  
  # Test with brief reasoning in mod mail
  python test_moderator.py --directory real_test_posts/ --random 2 --reasoning-level 1
  
  # Test with no reasoning in mod mail
  python test_moderator.py --file sample_post.txt --reasoning-level 0
  
  # 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(
        "--pause", "-p",
        action="store_true",
        help="Pause between each evaluation result (press Enter to continue)"
    )
    
    parser.add_argument(
        "--reasoning-level", "-rl",
        type=int,
        choices=[0, 1, 2],
        default=2,
        help="Reasoning level for mod mail (0=none, 1=brief, 2=full, default: 2)"
    )
    
    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, reasoning_level=args.reasoning_level)
    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, pause=args.pause, moderator=moderator)
    
    # 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()