utils.py

"""Utility functions for the bot"""
import re
from typing import Optional
from urllib.parse import urlparse


def is_valid_url(url: str) -> bool:
    """
    Validate if string is a valid URL - supports multiple protocols and domain types
    
    Supported protocols:
    - HTTP/HTTPS (web)
    - FTP/FTPS (file transfer)
    - WS/WSS (websockets)
    - RTSP/RTMP (streaming)
    - SSH/SFTP (secure protocols)
    
    Supported domains:
    - gTLDs: .com, .net, .org, .info, .biz, etc.
    - New gTLDs: .io, .ai, .dev, .app, .tech, .cloud, .store, .buy, .mov, etc.
    - Country codes: .uk, .de, .jp, .cn, .in, .ru, etc.
    - Subdomains: api.example.com, www.site.co.uk
    - Localhost and IP addresses
    - URLs with file paths: /index.html, /api/endpoint.php
    
    Args:
        url: URL string to validate
        
    Returns:
        bool: True if valid URL
    """
    try:
        # Supported protocols
        supported_protocols = (
            'http://', 'https://',  # Web
            'ftp://', 'ftps://',    # File transfer
            'ws://', 'wss://',      # WebSocket
            'rtsp://', 'rtmp://',   # Streaming
            'ssh://', 'sftp://',    # Secure protocols
            'file://',              # Local files
            'data://',              # Data URIs
        )
        
        # Add scheme if not present (default to https)
        if not any(url.startswith(proto) for proto in supported_protocols):
            url = 'https://' + url
        
        result = urlparse(url)
        
        # Check if we have netloc (domain) and scheme
        if not all([result.scheme, result.netloc]):
            return False
        
        # Validate domain name or IP
        netloc = result.netloc.split(':')[0]  # Remove port if present
        
        # Allow localhost and IP addresses
        if netloc in ['localhost', '127.0.0.1', '0.0.0.0']:
            return True
        
        # Check for valid IP address (IPv4)
        if re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', netloc):
            return True
        
        # Check for valid IPv6 address
        if ':' in netloc and netloc.startswith('[') and netloc.endswith(']'):
            return True
        
        # Check for valid domain name (including new TLDs and country codes)
        # Pattern: subdomain.domain.tld or domain.tld or domain.co.uk
        # More permissive pattern to support all TLD lengths (2-63 chars per RFC)
        domain_pattern = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)*[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.([a-zA-Z]{2,63}|[a-zA-Z]{2,63}\.[a-zA-Z]{2,63})$'
        if re.match(domain_pattern, netloc):
            return True
        
        # Check for domain without explicit TLD (like 'localhost' variations)
        if re.match(r'^[a-zA-Z0-9][a-zA-Z0-9\-]{0,61}[a-zA-Z0-9]$', netloc):
            return True
        
        return False
    except Exception:
        return False


def normalize_url(url: str) -> str:
    """
    Normalize URL by adding scheme if missing - supports multiple protocols
    
    Args:
        url: URL to normalize
        
    Returns:
        str: Normalized URL
    """
    url = url.strip()
    
    # Supported protocols
    protocols = (
        'http://', 'https://',
        'ftp://', 'ftps://',
        'ws://', 'wss://',
        'rtsp://', 'rtmp://',
        'ssh://', 'sftp://',
        'file://', 'data://'
    )
    
    # Remove trailing slashes (except for protocol-only URLs)
    if not url.endswith('://') and not any(url == proto.rstrip('/') for proto in protocols):
        url = url.rstrip('/')
    
    # Add https if no scheme
    if not any(url.startswith(proto) for proto in protocols):
        url = 'https://' + url
    
    return url


def extract_domain(url: str) -> Optional[str]:
    """
    Extract domain from URL - supports multiple protocols
    
    Args:
        url: URL to extract domain from
        
    Returns:
        str: Domain name or None
    """
    try:
        protocols = (
            'http://', 'https://',
            'ftp://', 'ftps://',
            'ws://', 'wss://',
            'rtsp://', 'rtmp://',
            'ssh://', 'sftp://',
            'file://', 'data://'
        )
        
        if not any(url.startswith(proto) for proto in protocols):
            url = 'https://' + url
        
        parsed = urlparse(url)
        return parsed.netloc
    except Exception:
        return None


def truncate_text(text: str, max_length: int = 100, suffix: str = '...') -> str:
    """
    Truncate text to maximum length
    
    Args:
        text: Text to truncate
        max_length: Maximum length
        suffix: Suffix to add if truncated
        
    Returns:
        str: Truncated text
    """
    if len(text) <= max_length:
        return text
    return text[:max_length - len(suffix)] + suffix


def format_size(bytes_size: int) -> str:
    """
    Format bytes size to human readable format
    
    Args:
        bytes_size: Size in bytes
        
    Returns:
        str: Formatted size (e.g., "1.5 MB")
    """
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if bytes_size < 1024.0:
            return f"{bytes_size:.2f} {unit}"
        bytes_size /= 1024.0
    return f"{bytes_size:.2f} PB"


def escape_markdown(text: str) -> str:
    """
    Escape markdown special characters for Telegram
    
    Args:
        text: Text to escape
        
    Returns:
        str: Escaped text
    """
    special_chars = ['_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!']
    for char in special_chars:
        text = text.replace(char, f'\\{char}')
    return text


def format_list(items: list, bullet: str = '•', max_items: int = 5) -> str:
    """
    Format list of items with bullets
    
    Args:
        items: List of items
        bullet: Bullet character
        max_items: Maximum items to show
        
    Returns:
        str: Formatted list
    """
    if not items:
        return 'None'
    
    display_items = items[:max_items]
    result = '\n'.join([f"{bullet} {item}" for item in display_items])
    
    if len(items) > max_items:
        result += f"\n{bullet} ... and {len(items) - max_items} more"
    
    return result


def get_status_emoji(status: str) -> str:
    """
    Get emoji for status
    
    Args:
        status: Status string
        
    Returns:
        str: Emoji
    """
    status_map = {
        'AVAILABLE': '🟢',
        'UNAVAILABLE': '🔴',
        'AUTH_REQUIRED': '🔐',
        'REGION_BLOCKED': '🌍',
        'IP_BLOCKED': '🚫',
        'TIMEOUT': '⏱️',
        'ERROR': '❌',
        'SECURE': '✅',
        'LEAKED': '⚠️',
        'OUTDATED': '⚠️',
        'NO_INFO': 'ℹ️'
    }
    return status_map.get(status.upper(), '❓')


def validate_user_input(text: str, input_type: str = 'url') -> tuple[bool, str]:
    """
    Validate user input
    
    Args:
        text: Input text
        input_type: Type of input ('url', 'email', etc.)
        
    Returns:
        tuple: (is_valid, error_message)
    """
    if not text or not text.strip():
        return False, "Input cannot be empty"
    
    text = text.strip()
    
    if input_type == 'url':
        if len(text) > 2048:
            return False, "URL too long (max 2048 characters)"
        
        if not is_valid_url(text):
            return False, "Invalid URL format"
        
        return True, ""
    
    return True, ""


def get_supported_protocols() -> dict:
    """
    Get list of supported protocols and their descriptions
    
    Returns:
        dict: Protocol information
    """
    return {
        'web': {
            'protocols': ['http', 'https'],
            'description': 'Web protocols for websites and APIs',
            'examples': ['https://example.com', 'http://api.example.org']
        },
        'file_transfer': {
            'protocols': ['ftp', 'ftps', 'sftp'],
            'description': 'File transfer protocols',
            'examples': ['ftp://files.example.com', 'sftp://secure.example.net']
        },
        'websocket': {
            'protocols': ['ws', 'wss'],
            'description': 'WebSocket protocols for real-time communication',
            'examples': ['wss://api.example.com/socket', 'ws://localhost:8080']
        },
        'streaming': {
            'protocols': ['rtsp', 'rtmp'],
            'description': 'Streaming protocols for audio/video',
            'examples': ['rtsp://stream.example.com:554/live', 'rtmp://live.example.tv']
        },
        'secure': {
            'protocols': ['ssh'],
            'description': 'Secure shell and remote access',
            'examples': ['ssh://server.example.com:22']
        }
    }


def get_supported_domains() -> dict:
    """
    Get information about supported domain types
    
    Returns:
        dict: Domain type information
    """
    return {
        'generic_tlds': {
            'examples': ['.com', '.net', '.org', '.info', '.biz'],
            'description': 'Generic top-level domains'
        },
        'new_tlds': {
            'examples': ['.io', '.ai', '.dev', '.app', '.tech', '.cloud', '.online', '.store', '.buy', '.mov', '.shop', '.xyz'],
            'description': 'New generic top-level domains'
        },
        'country_codes': {
            'examples': ['.uk', '.de', '.jp', '.cn', '.in', '.au', '.ca', '.fr', '.it', '.br', '.ru', '.es', '.nl', '.mx'],
            'description': 'Country code top-level domains'
        },
        'second_level': {
            'examples': ['.co.uk', '.com.au', '.gov.uk', '.ac.uk', '.org.in'],
            'description': 'Second-level domains (country-specific)'
        },
        'subdomains': {
            'examples': ['api.example.com', 'www.site.io', 'dev.test.example.org'],
            'description': 'Subdomains and multi-level domains'
        },
        'special': {
            'examples': ['localhost', '127.0.0.1', '192.168.1.1', '[::1]'],
            'description': 'Localhost and IP addresses (IPv4 and IPv6)'
        }
    }


def format_protocol_info(protocol: str) -> str:
    """
    Format protocol information for display
    
    Args:
        protocol: Protocol name
        
    Returns:
        str: Formatted information
    """
    protocols_info = get_supported_protocols()
    
    for category, info in protocols_info.items():
        if protocol.lower() in info['protocols']:
            return f"{protocol.upper()} - {info['description']}"
    
    return f"{protocol.upper()} - Unknown protocol"


def create_progress_bar(percentage: float, length: int = 10) -> str:
    """
    Create a text-based progress bar
    
    Args:
        percentage: Percentage (0-100)
        length: Length of bar
        
    Returns:
        str: Progress bar
    """
    filled = int(length * percentage / 100)
    bar = '█' * filled + '░' * (length - filled)
    return f"{bar} {percentage:.0f}%"


def sanitize_filename(filename: str) -> str:
    """
    Sanitize filename by removing invalid characters
    
    Args:
        filename: Filename to sanitize
        
    Returns:
        str: Sanitized filename
    """
    # Remove invalid characters
    filename = re.sub(r'[<>:"/\\|?*]', '_', filename)
    # Remove leading/trailing spaces and dots
    filename = filename.strip('. ')
    # Limit length
    if len(filename) > 255:
        filename = filename[:255]
    return filename


def setup_logging(log_level: str = 'INFO', log_file: Optional[str] = None):
    """
    Setup logging configuration
    
    Args:
        log_level: Logging level
        log_file: Optional log file path
    """
    import sys

    from loguru import logger
    
    # Remove default handler
    logger.remove()
    
    # Add console handler with colors
    logger.add(
        sys.stdout,
        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan> - <level>{message}</level>",
        level=log_level,
        colorize=True
    )
    
    # Add file handler if specified
    if log_file:
        logger.add(
            log_file,
            format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function} - {message}",
            level=log_level,
            rotation="10 MB",
            retention="7 days",
            compression="zip"
        )
    
    logger.info("Logging configured successfully")