csv_launch_detection_workflow.py

#!/usr/bin/env python3
"""
CSV launch detection workflow using domain architecture
Processes historical token data from CSV file with same pipeline as live detection
"""

import asyncio
import logging
import signal
import sys
import platform
from pathlib import Path

from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Add project root to Python path and import constants
from src.constants import PROJECT_ROOT

sys.path.insert(0, str(PROJECT_ROOT))

from src.domain.workflow.default_workflow import DefaultWorkflow
from src.domain.input_source.csv_token_input_source import CSVTokenInputSource
from src.domain.processor.simple_launch_detection_processor import (
    SimpleLaunchDetectionProcessor,
)
from src.domain.executor.simple_file_executor import SimpleFileExecutor

# Configure logging with detailed error tracking
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"
)

# Set INFO level logging for key workflow components
logging.getLogger("src.domain.processor.simple_launch_detection_processor").setLevel(
    logging.INFO
)
logging.getLogger("src.domain.processor.stages.token_metadata_stage").setLevel(
    logging.INFO
)
logging.getLogger("src.domain.processor.stages.simple_file_storage_stage").setLevel(
    logging.INFO
)
logging.getLogger("src.domain.input_source.csv_token_input_source").setLevel(
    logging.INFO
)
logging.getLogger("src.domain.executor.simple_file_executor").setLevel(logging.INFO)
logger = logging.getLogger(__name__)


class CSVLaunchDetectionWorkflow:
    """CSV launch detection workflow orchestrator - processes historical token data"""

    def __init__(self, csv_file: str, batch_size: int = 100):
        """
        Initialize CSV launch detection workflow

        Args:
            csv_file: Path to CSV file containing token data
            batch_size: Number of rows to read and process at a time (default: 100)
        """
        self.csv_file = csv_file
        self.batch_size = batch_size
        self.workflow = None
        self.input_source = None
        self.running = False

    async def initialize(self):
        """Initialize all workflow components"""
        logger.info("🔧 Initializing CSV Launch Detection Workflow...")

        # Create CSV input source
        self.input_source = CSVTokenInputSource(
            csv_file=self.csv_file, batch_size=self.batch_size
        )

        # Create simplified processor (no AI services needed)
        processor = SimpleLaunchDetectionProcessor()

        # Create simple file executor (no database operations)
        executor = SimpleFileExecutor()
        await executor.initialize()

        # Create workflow
        self.workflow = DefaultWorkflow(
            input_source=self.input_source, processor=processor, executor=executor
        )

        # Initialization will happen in DefaultWorkflow
        logger.info(
            "✅ CSV Launch Detection Workflow initialized with 3-stage pipeline: "
            "metadata extraction + image processing + file storage (no AI processing)"
        )

    async def start(self):
        """Start the workflow"""
        if not self.workflow:
            await self.initialize()

        logger.info("🚀 Starting CSV Launch Detection Workflow...")
        print("🚀 CSV Token Launch Detection Workflow Starting...")
        print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
        print(f"📁 CSV File: {self.csv_file}")
        print(f"📦 Batch size: {self.batch_size} rows")
        print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
        print("🎯 Processing historical token launches from CSV...")
        print("📁 File storage enabled - memecoins will be saved as JSON + JPG pairs")
        print("🖼️ Images processed to 512x512px JPEG with 90% quality")
        print("💾 Target directory: res/memecoins/pending/")
        print("🗑️ Processed rows will be deleted from CSV file")
        print("❌ AI processing disabled - no tags, captions, or embeddings")
        print("❌ Database storage disabled - only file operations")
        print("📊 Complete token data will be stored as files for later processing")
        print("🛑 Press Ctrl+C to stop")
        print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n")

        self.running = True

        # Use the DefaultWorkflow's built-in initialization and run
        await self.workflow.run_indefinitely()

    async def stop(self):
        """Stop the workflow"""
        logger.info("🛑 Stopping CSV Launch Detection Workflow...")
        self.running = False

        if self.workflow:
            await self.workflow.stop()

        logger.info("✅ CSV Launch Detection Workflow stopped")


async def main(csv_file: str = "res/token_dataset.csv", batch_size: int = 100):
    """Main function

    Args:
        csv_file: Path to CSV file containing token data (default: res/token_dataset.csv)
        batch_size: Number of rows to read and process at a time (default: 100)
    """

    # Verify CSV file exists
    csv_path = Path(csv_file)
    if not csv_path.is_absolute():
        csv_path = PROJECT_ROOT / csv_file

    if not csv_path.exists():
        logger.error(f"❌ CSV file not found: {csv_path}")
        logger.info(f"💡 Please ensure the CSV file exists at: {csv_path}")
        sys.exit(1)

    # Create and run workflow
    workflow = CSVLaunchDetectionWorkflow(csv_file=csv_file, batch_size=batch_size)

    # Handle graceful shutdown with cross-platform signal handling
    def signal_handler():
        logger.info("🛑 Shutdown signal received")
        asyncio.create_task(workflow.stop())

    # Set up signal handlers (Unix/Linux/macOS only - Windows doesn't support loop.add_signal_handler)
    if platform.system() != "Windows":
        loop = asyncio.get_event_loop()
        for sig in [signal.SIGTERM, signal.SIGINT]:
            loop.add_signal_handler(sig, signal_handler)
    else:
        # Windows: signal handling is done through KeyboardInterrupt exception
        logger.info(
            "🪟 Running on Windows - signal handling via KeyboardInterrupt only"
        )

    try:
        await workflow.start()
    except KeyboardInterrupt:
        await workflow.stop()
    except Exception as e:
        logger.error(f"❌ Unexpected error: {e}")
        await workflow.stop()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n👋 Goodbye!")
        sys.exit(0)