Untitled

"""
URL Processing Tool with PHP API Integration
Processes URLs through various PHP payment gateway APIs with multi-threading support.
Features: progress resume, live stats, graceful shutdown, structured logging, auto-retry.
"""

import os
import sys
import subprocess
import logging
import json
import signal
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import List, Optional, Dict
from dataclasses import dataclass, field
from threading import Lock, Event
import time


# ─── ANSI Colors ─────────────────────────────────────────────────────────────
class C:
    """ANSI color codes for terminal output."""
    RESET   = "\033[0m"
    BOLD    = "\033[1m"
    DIM     = "\033[2m"
    RED     = "\033[91m"
    GREEN   = "\033[92m"
    YELLOW  = "\033[93m"
    BLUE    = "\033[94m"
    MAGENTA = "\033[95m"
    CYAN    = "\033[96m"
    WHITE   = "\033[97m"
    BG_RED  = "\033[41m"
    BG_GREEN = "\033[42m"


# ─── Configuration ───────────────────────────────────────────────────────────
@dataclass
class Config:
    """Configuration settings for the processor."""
    PHP_EXECUTABLE: str = r'C:\xampp\php\php.exe'
    INPUT_FILE: str = 'cc.txt'
    PROGRESS_FILE: str = 'progress.json'
    ERROR_FILE: str = 'errors.txt'
    LOG_FILE: str = 'processor.log.json'
    API_FOLDER: str = 'api'

    API_CONFIGS: Dict[str, str] = field(default_factory=lambda: {
        '1': 'vanco',
        '2': 'adyen',
    })

    MAX_RETRIES: int = 2
    RETRY_BACKOFF: float = 0.5        # seconds between retries (doubles each attempt)
    TIMEOUT_SECONDS: int = 45
    PROGRESS_SAVE_INTERVAL: int = 10   # flush progress to disk every N lines

    def __post_init__(self):
        """Validate configuration on initialization."""
        php_path = Path(self.PHP_EXECUTABLE)
        if not php_path.exists():
            print(f"{C.YELLOW}  Warning: PHP executable not found: {self.PHP_EXECUTABLE}{C.RESET}")
            print(f"{C.DIM}  Trying system PATH...{C.RESET}")
            # Try to find php on PATH
            import shutil
            found = shutil.which("php")
            if found:
                self.PHP_EXECUTABLE = found
                print(f"{C.GREEN}  Found PHP at: {found}{C.RESET}")
            else:
                raise FileNotFoundError(f"PHP executable not found: {self.PHP_EXECUTABLE}")

        api_folder = Path(self.API_FOLDER)
        if not api_folder.exists():
            raise FileNotFoundError(f"API folder not found: {self.API_FOLDER}")

        if not self.API_CONFIGS:
            raise ValueError("API_CONFIGS cannot be empty")

        # Validate all API PHP files exist at init
        for key, name in self.API_CONFIGS.items():
            php_file = api_folder / name
            if php_file.suffix != '.php':
                php_file = php_file.with_suffix('.php')
            if not php_file.exists():
                print(f"{C.YELLOW}  Warning: API '{name}' file not found: {php_file}{C.RESET}")


# ─── JSON Log Formatter ─────────────────────────────────────────────────────
class JsonFormatter(logging.Formatter):
    """Outputs each log record as a single JSON line."""
    def format(self, record):
        return json.dumps({
            'time': self.formatTime(record, '%Y-%m-%d %H:%M:%S'),
            'level': record.levelname,
            'message': record.getMessage(),
        })


# ─── Main Processor ─────────────────────────────────────────────────────────
class URLProcessor:
    """Processes URLs through PHP APIs with threading, progress tracking, and live stats."""

    def __init__(self, config: Config):
        self.config = config
        self._lock_progress = Lock()
        self._lock_error = Lock()
        self._lock_stats = Lock()
        self._lock_print = Lock()
        self._shutdown = Event()
        self.stats = {'processed': 0, 'successful': 0, 'failed': 0, 'errors': 0}
        self.total_urls = 0
        self.api_file_cache: Dict[str, Optional[Path]] = {}
        self.progress_counter = 0
        self._setup_logging()
        self._setup_signal_handlers()

    # ── Logging ──────────────────────────────────────────────────────────────

    def _setup_logging(self) -> None:
        """Configure JSON file logging + console logging (fresh each run)."""
        root = logging.getLogger()
        root.setLevel(logging.INFO)
        root.handlers.clear()

        # JSON file — overwrite each run
        fh = logging.FileHandler(self.config.LOG_FILE, mode='w', encoding='utf-8')
        fh.setFormatter(JsonFormatter())
        root.addHandler(fh)

        # Console — only warnings+ to avoid noise during threaded output
        ch = logging.StreamHandler(sys.stderr)
        ch.setLevel(logging.WARNING)
        ch.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
        root.addHandler(ch)

        self.logger = logging.getLogger(__name__)

    # ── Graceful Shutdown ────────────────────────────────────────────────────

    def _setup_signal_handlers(self) -> None:
        """Register Ctrl+C handler for graceful shutdown."""
        def handler(signum, frame):
            if not self._shutdown.is_set():
                self._safe_print(f"\n{C.YELLOW}{C.BOLD}  Shutting down gracefully... "
                                 f"(waiting for active tasks){C.RESET}")
                self._shutdown.set()
        signal.signal(signal.SIGINT, handler)

    # ── Thread-safe Printing ─────────────────────────────────────────────────

    def _safe_print(self, msg: str) -> None:
        """Print without garbled output in multi-threaded context."""
        with self._lock_print:
            print(msg, flush=True)

    # ── Screen ───────────────────────────────────────────────────────────────

    @staticmethod
    def clear_screen() -> None:
        os.system('cls' if os.name == 'nt' else 'clear')

    @staticmethod
    def banner() -> None:
        print(f"""
{C.CYAN}{C.BOLD}  ╔══════════════════════════════════════╗
  ║        EONARA - URL Processor        ║
  ╚══════════════════════════════════════╝{C.RESET}
""")

    # ── API File Resolution ──────────────────────────────────────────────────

    def _get_api_file_path(self, api_name: str) -> Optional[Path]:
        """Resolve and cache the full path to a PHP API script."""
        if api_name in self.api_file_cache:
            return self.api_file_cache[api_name]

        full_path = Path(self.config.API_FOLDER) / api_name
        if full_path.suffix != '.php':
            full_path = full_path.with_suffix('.php')

        if full_path.exists():
            self.api_file_cache[api_name] = full_path
            return full_path

        self.logger.error(f"API file not found: {full_path}")
        self.api_file_cache[api_name] = None
        return None

    # ── PHP Execution ────────────────────────────────────────────────────────

    def call_php_api(self, url: str, api_name: str) -> Optional[str]:
        """
        Execute a PHP script with the given URL piped to stdin.
        Retries with exponential backoff on timeout/failure.
        Returns stripped stdout or None.
        """
        file_path = self._get_api_file_path(api_name)
        if not file_path:
            return None

        backoff = self.config.RETRY_BACKOFF

        for attempt in range(1, self.config.MAX_RETRIES + 1):
            if self._shutdown.is_set():
                return None

            process = None
            try:
                process = subprocess.Popen(
                    [self.config.PHP_EXECUTABLE, str(file_path)],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    encoding='utf-8',
                    errors='replace',
                )
                output, error = process.communicate(
                    input=url,
                    timeout=self.config.TIMEOUT_SECONDS,
                )

                if error and error.strip():
                    self.logger.debug(f"PHP stderr (attempt {attempt}): {error.strip()}")

                if output and output.strip():
                    return output.strip()

                # Empty output — treat as failure, retry
                self.logger.debug(f"Empty output on attempt {attempt} for {api_name}")

            except subprocess.TimeoutExpired:
                self.logger.info(f"Timeout attempt {attempt}/{self.config.MAX_RETRIES} [{api_name}]")
                if process:
                    try:
                        process.kill()
                        process.communicate(timeout=5)
                    except Exception:
                        pass
            except Exception as e:
                self.logger.error(f"Error attempt {attempt}/{self.config.MAX_RETRIES} [{api_name}]: {e}")

            # Backoff before retry
            if attempt < self.config.MAX_RETRIES:
                time.sleep(backoff)
                backoff *= 2

        return None

    # ── File I/O ─────────────────────────────────────────────────────────────

    def read_urls_from_file(self, file_path: str) -> List[str]:
        """Read non-empty lines from input file."""
        try:
            with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
                urls = [line.strip() for line in f if line.strip()]
            self.logger.info(f"Loaded {len(urls)} lines from {file_path}")
            return urls
        except FileNotFoundError:
            self.logger.error(f"Input file not found: {file_path}")
            return []
        except Exception as e:
            self.logger.error(f"Error reading {file_path}: {e}")
            return []

    def save_progress(self, force: bool = False) -> None:
        """Increment and periodically flush progress to disk (thread-safe)."""
        with self._lock_progress:
            self.progress_counter += 1
            if force or self.progress_counter % self.config.PROGRESS_SAVE_INTERVAL == 0:
                try:
                    with open(self.config.PROGRESS_FILE, 'w') as f:
                        json.dump({'lines_processed': self.progress_counter}, f)
                except Exception as e:
                    self.logger.error(f"Error saving progress: {e}")

    def save_error_url(self, url: str) -> None:
        """Append a failed URL to the error file (thread-safe)."""
        with self._lock_error:
            try:
                with open(self.config.ERROR_FILE, 'a', encoding='utf-8') as f:
                    f.write(url + '\n')
            except Exception as e:
                self.logger.error(f"Error writing to error file: {e}")

    # ── Progress Resume ──────────────────────────────────────────────────────

    def get_start_line(self) -> int:
        """Check for saved progress and ask user whether to resume."""
        prog_path = Path(self.config.PROGRESS_FILE)
        if not prog_path.exists():
            return 1

        try:
            with open(prog_path, 'r') as f:
                data = json.load(f)
            lines_done = data.get('lines_processed', 0)

            if lines_done > 0:
                print(f"  {C.GREEN}Found saved progress: {lines_done} lines processed{C.RESET}")
                choice = input("  Continue from where you left off? (y/n): ").strip().lower()
                if choice == 'y':
                    self.progress_counter = lines_done
                    return lines_done + 1
                else:
                    self._reset_progress()
                    return 1
        except Exception as e:
            self.logger.warning(f"Could not read progress file: {e}")

        return 1

    def _reset_progress(self) -> None:
        """Reset progress file to zero."""
        try:
            with open(self.config.PROGRESS_FILE, 'w') as f:
                json.dump({'lines_processed': 0}, f)
        except Exception as e:
            self.logger.error(f"Error resetting progress: {e}")

    # ── Stats ────────────────────────────────────────────────────────────────

    def _update_stats(self, key: str) -> None:
        with self._lock_stats:
            self.stats[key] += 1

    def _get_stats_snapshot(self) -> dict:
        with self._lock_stats:
            return dict(self.stats)

    def _progress_line(self) -> str:
        """Build a compact one-line progress summary."""
        s = self._get_stats_snapshot()
        done = s['processed']
        pct = (done / self.total_urls * 100) if self.total_urls else 0
        return (f"  {C.DIM}[{done}/{self.total_urls}  {pct:5.1f}%]  "
                f"{C.GREEN}OK:{s['successful']}  {C.RED}FAIL:{s['failed']}  "
                f"{C.YELLOW}ERR:{s['errors']}{C.RESET}")

    def display_stats(self, elapsed: float) -> None:
        """Display final processing statistics."""
        s = self.stats
        speed = s['processed'] / max(elapsed, 0.01)
        print(f"""
{C.BOLD}  {'─' * 46}
  RESULTS
  {'─' * 46}{C.RESET}
  {C.GREEN}Successful : {s['successful']}{C.RESET}
  {C.RED}Failed     : {s['failed']}{C.RESET}
  {C.YELLOW}Errors     : {s['errors']}{C.RESET}
  {C.CYAN}Total      : {s['processed']}{C.RESET}
  {C.DIM}Time       : {elapsed:.1f}s  ({speed:.1f} URLs/sec){C.RESET}
  {'─' * 46}
""")

    # ── Single URL Processing ────────────────────────────────────────────────

    def process_single_url(self, url: str, line_number: int, api_name: str) -> None:
        """Process one URL and update stats/progress."""
        if self._shutdown.is_set():
            return

        if not url:
            self._update_stats('errors')
            return

        output = self.call_php_api(url, api_name)

        if self._shutdown.is_set():
            return

        if output:
            self._safe_print(f"  {C.BLUE}[{line_number}]{C.RESET} {output}")
            self.save_progress()
            self._update_stats('successful')
        else:
            self.save_error_url(url)
            self._safe_print(f"  {C.BLUE}[{line_number}]{C.RESET} {C.RED}TIMEOUT / NO RESPONSE{C.RESET}")
            self._update_stats('failed')

        self._update_stats('processed')

        # Print live progress every 25 items
        with self._lock_stats:
            if self.stats['processed'] % 25 == 0:
                pass  # printed below outside lock
        if self.stats['processed'] % 25 == 0:
            self._safe_print(self._progress_line())

    # ── User Input Helpers ───────────────────────────────────────────────────

    @staticmethod
    def get_int_input(prompt: str, lo: int, hi: int) -> int:
        """Prompt until a valid integer in [lo, hi] is entered."""
        while True:
            try:
                val = int(input(prompt).strip())
                if lo <= val <= hi:
                    return val
                print(f"  Enter a number between {lo} and {hi}.")
            except ValueError:
                print("  Invalid number.")

    @staticmethod
    def get_choice(prompt: str, options: List[str]) -> Optional[str]:
        """Prompt for a choice from a list."""
        try:
            choice = input(prompt).strip()
            return choice if choice in options else None
        except (EOFError, KeyboardInterrupt):
            return None

    # ── Main Loop ────────────────────────────────────────────────────────────

    def run(self) -> None:
        """Main execution entrypoint."""
        self.clear_screen()
        self.banner()

        # ── Workers
        num_workers = self.get_int_input(
            f"  {C.BOLD}Workers{C.RESET} (1-50): ", 1, 50
        )

        # ── Mode
        print(f"\n  {C.BOLD}Mode:{C.RESET}")
        print(f"  1. Debug  {C.DIM}(sequential, verbose){C.RESET}")
        print(f"  2. Normal {C.DIM}(multi-threaded){C.RESET}")
        mode = self.get_choice("  Choice (1/2): ", ['1', '2'])
        if not mode:
            print("  Invalid. Exiting.")
            return
        debug_mode = mode == '1'

        # ── API selection
        print(f"\n  {C.BOLD}APIs:{C.RESET}")
        for k, v in self.config.API_CONFIGS.items():
            print(f"  {k}. {v}")

        api_choice = self.get_choice(
            f"\n  {C.BOLD}Pick API:{C.RESET} ",
            list(self.config.API_CONFIGS.keys()),
        )
        if not api_choice:
            print("  Invalid API. Exiting.")
            return

        api_name = self.config.API_CONFIGS[api_choice]
        print(f"  {C.GREEN}API: {api_name}{C.RESET}")

        # ── Validate API file exists
        if not self._get_api_file_path(api_name):
            print(f"\n  {C.RED}PHP script not found for '{api_name}'. Exiting.{C.RESET}")
            return

        # ── Load URLs
        start_line = self.get_start_line()
        all_urls = self.read_urls_from_file(self.config.INPUT_FILE)
        if not all_urls:
            print(f"  {C.RED}No URLs to process. Add lines to {self.config.INPUT_FILE}{C.RESET}")
            return

        urls = all_urls[start_line - 1:]
        self.total_urls = len(urls)

        print(f"\n  {C.CYAN}Starting from line {start_line}, "
              f"processing {self.total_urls} URLs with {num_workers} workers...{C.RESET}\n")

        # ── Reset stats
        self.stats = {'processed': 0, 'successful': 0, 'failed': 0, 'errors': 0}
        start_time = time.time()

        # ── Process
        try:
            if debug_mode:
                for idx, url in enumerate(urls, start=start_line):
                    if self._shutdown.is_set():
                        break
                    self.process_single_url(url, idx, api_name)
            else:
                with ThreadPoolExecutor(max_workers=num_workers) as executor:
                    futures = {
                        executor.submit(self.process_single_url, url, idx, api_name): idx
                        for idx, url in enumerate(urls, start=start_line)
                    }
                    for future in as_completed(futures):
                        if self._shutdown.is_set():
                            # Cancel pending futures
                            for f in futures:
                                f.cancel()
                            break
                        try:
                            future.result()
                        except Exception as e:
                            self.logger.error(f"Future error: {e}")
        except Exception as e:
            self.logger.error(f"Processing error: {e}")

        # ── Finalize
        elapsed = time.time() - start_time
        self.save_progress(force=True)
        self.display_stats(elapsed)

        if self._shutdown.is_set():
            print(f"  {C.YELLOW}Interrupted — progress saved. Resume next run.{C.RESET}\n")


# ─── Entry Point ─────────────────────────────────────────────────────────────

def main():
    try:
        config = Config()
        processor = URLProcessor(config)
        processor.run()
        print(f"  {C.GREEN}{C.BOLD}All Done!{C.RESET}\n")
    except FileNotFoundError as e:
        print(f"\n  {C.RED}Config Error: {e}{C.RESET}")
        sys.exit(1)
    except KeyboardInterrupt:
        print(f"\n  {C.YELLOW}Interrupted.{C.RESET}")
    except Exception as e:
        logging.error(f"Unexpected error: {e}", exc_info=True)
        print(f"\n  {C.RED}Error: {e}{C.RESET}")
        sys.exit(1)


if __name__ == "__main__":
    main()