app.py

import re
import string
import time
import sys
import os
import requests
import gradio as gr
from threading import Lock
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
import asyncio
import aiohttp
import multiprocessing as mp

# ═══════════════════════════════════════════════════
# ⚙️ ULTRA-FAST CONFIGURATION
# ═══════════════════════════════════════════════════
TIMEOUT = 1.5
TARGET_TIME = 30
MIN_WORKERS = 500
MAX_WORKERS = 5000
BATCH_SIZE = 500

checked_lock = Lock()

# ═══════════════════════════════════════════════════
# 🎨 UTILITY FUNCTIONS
# ═══════════════════════════════════════════════════

def format_size(size_str):
    try:
        size = int(size_str)
        if size >= 1_000_000_000:
            return f"{size/1_000_000_000:.2f} GB"
        elif size >= 1_000_000:
            return f"{size/1_000_000:.2f} MB"
        elif size >= 1_000:
            return f"{size/1_000:.2f} KB"
        else:
            return f"{size} B"
    except:
        return size_str

def format_time(seconds):
    if seconds < 60:
        return f"{seconds:.1f}s"
    elif seconds < 3600:
        mins = int(seconds // 60)
        secs = seconds % 60
        return f"{mins}m {secs:.0f}s"
    else:
        hours = int(seconds // 3600)
        mins = int((seconds % 3600) // 60)
        return f"{hours}h {mins}m"

def calculate_optimal_workers(total_requests):
    requests_per_second_per_worker = 100
    total_needed_per_second = total_requests / TARGET_TIME
    optimal_workers = total_needed_per_second / requests_per_second_per_worker
    optimal_workers *= 1.2
    optimal_workers = max(MIN_WORKERS, min(optimal_workers, MAX_WORKERS))
    return int(optimal_workers)

def parse_firmware_url(url):
    pattern = re.compile(
        r"^(https?://.+?/5G/UDO/)"
        r"DIN\d+/"
        r"([^_/]+)_(\d+)_S/"
        r"([A-Za-z])(.*)$"
    )
    
    match = pattern.match(url.strip())
    
    if match:
        return {
            'base_prefix': match.group(1),
            'model': match.group(2),
            'current_version': int(match.group(3)),
            'suffix': match.group(5),
            'url': url.strip()
        }
    return None

# ═══════════════════════════════════════════════════
# ⚡ ASYNC SCANNER
# ═══════════════════════════════════════════════════

async def async_check_url(session, url, semaphore):
    async with semaphore:
        try:
            async with session.head(
                url, 
                timeout=aiohttp.ClientTimeout(total=TIMEOUT), 
                allow_redirects=True
            ) as r:
                if r.status in (403, 405):
                    async with session.get(
                        url, 
                        timeout=aiohttp.ClientTimeout(total=TIMEOUT)
                    ) as r2:
                        size = r2.headers.get("Content-Length", "Unknown")
                        return r2.status == 200, url, size
                else:
                    size = r.headers.get("Content-Length", "Unknown")
                    return r.status == 200, url, size
        except:
            return False, url, "Error"

async def batch_scan_urls(urls, max_concurrent=1000):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    connector = aiohttp.TCPConnector(
        limit=max_concurrent,
        limit_per_host=max_concurrent,
        ttl_dns_cache=300,
        force_close=False,
        enable_cleanup_closed=True
    )
    
    timeout = aiohttp.ClientTimeout(total=TIMEOUT, connect=0.5)
    
    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        headers={'Connection': 'keep-alive'}
    ) as session:
        tasks = [async_check_url(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return results

def generate_urls(version, din_start, din_end, base_url, suffix):
    urls = []
    for din_num in range(din_start, din_end + 1):
        din = f"DIN{din_num:03d}"
        for letter in string.ascii_lowercase + string.ascii_uppercase:
            filename = f"{letter}{suffix}"
            url = base_url.format(din=din, ver=version, file=filename)
            urls.append((url, version))
    return urls

# ═══════════════════════════════════════════════════
# 🤖 FIRMWARE SCANNER
# ═══════════════════════════════════════════════════

class FirmwareScanner:
    def __init__(self):
        self.current_hits = []
        self.scanning = False
        self.stop_requested = False
    
    def stop_scan(self):
        self.stop_requested = True
        self.scanning = False
        return "## Scan Stopped"
    
    def start_scan(self, firmware_url, version_input, din_input, stop_on_first):
        self.scanning = True
        self.stop_requested = False
        self.current_hits = []
        
        if not firmware_url or not firmware_url.strip():
            yield "## Error\n\nPlease enter a firmware URL.", "", "", ""
            return
        
        firmware_info = parse_firmware_url(firmware_url)
        if not firmware_info:
            yield "## Invalid URL\n\nPlease check the URL format.", "", "", ""
            return
        
        if not version_input or not version_input.strip():
            yield "## Error\n\nPlease enter version or range.", "", "", ""
            return
        
        if not din_input or not din_input.strip():
            yield "## Error\n\nPlease enter DIN or range.", "", "", ""
            return
        
        try:
            if '-' in version_input:
                parts = version_input.split('-')
                ver_start = int(parts[0].strip())
                ver_end = int(parts[1].strip())
            else:
                ver_start = ver_end = int(version_input.strip())
        except:
            yield "## Invalid Version\n\nEnter a number or range.", "", "", ""
            return
        
        try:
            if '-' in din_input:
                parts = din_input.split('-')
                din_start = int(parts[0].strip())
                din_end = int(parts[1].strip())
            else:
                din_start = din_end = int(din_input.strip())
        except:
            yield "## Invalid DIN\n\nEnter a number or range.", "", "", ""
            return
        
        total_versions = ver_end - ver_start + 1
        din_count = din_end - din_start + 1
        total_requests = total_versions * din_count * 52
        optimal_workers = calculate_optimal_workers(total_requests)
        
        base_url = firmware_info['base_prefix'] + "{din}/" + firmware_info['model'] + "_{ver}_S/{file}"
        suffix = firmware_info['suffix']
        model = firmware_info['model']
        
        all_hits = []
        overall_start = time.time()
        total_checked = 0
        
        yield (
            f"## Scanning Started\n\n"
            f"Model: {model}\n"
            f"Versions: {ver_start}-{ver_end}\n"
            f"DINs: {din_start}-{din_end}\n"
            f"Total URLs: {total_requests:,}\n"
            f"Workers: {optimal_workers}",
            "",
            f"## Config\n\n"
            f"| Setting | Value |\n"
            f"|---------|-------|\n"
            f"| Model | {model} |\n"
            f"| Versions | {ver_start}-{ver_end} |\n"
            f"| DINs | {din_start}-{din_end} |\n"
            f"| Total | {total_requests:,} |\n"
            f"| Workers | {optimal_workers} |",
            ""
        )
        
        for version in range(ver_start, ver_end + 1):
            if self.stop_requested:
                break
            
            urls = generate_urls(version, din_start, din_end, base_url, suffix)
            total_urls = len(urls)
            should_stop = stop_on_first and ((ver_start == ver_end) or (din_start == din_end))
            batch_size = min(BATCH_SIZE, total_urls)
            version_hits = []
            
            for i in range(0, total_urls, batch_size):
                if self.stop_requested:
                    break
                
                batch = urls[i:i+batch_size]
                batch_urls = [url for url, _ in batch]
                
                try:
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    results = loop.run_until_complete(
                        batch_scan_urls(batch_urls, optimal_workers)
                    )
                    loop.close()
                except:
                    results = []
                    with ThreadPoolExecutor(max_workers=optimal_workers) as executor:
                        def check_sync(url):
                            try:
                                r = requests.head(url, timeout=TIMEOUT, allow_redirects=True)
                                if r.status_code in (403, 405):
                                    r = requests.get(url, timeout=TIMEOUT, stream=True)
                                return r.status_code == 200, url, r.headers.get("Content-Length", "Unknown")
                            except:
                                return False, url, "Error"
                        
                        futures = [executor.submit(check_sync, url) for url in batch_urls]
                        results = [f.result() for f in futures]
                
                for result in results:
                    if isinstance(result, tuple) and len(result) == 3:
                        ok, url, size = result
                        if ok:
                            hit = (version, url, size)
                            version_hits.append(hit)
                            all_hits.append(hit)
                            
                            with open("found_200_urls.txt", "a", encoding="utf-8") as f:
                                f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]\n")
                                f.write(f"Version: {version}\nURL: {url}\nSize: {size}\n{'─'*50}\n")
                            
                            if should_stop:
                                self.stop_requested = True
                                break
                
                total_checked += len(batch)
                elapsed = time.time() - overall_start
                rps = total_checked / elapsed if elapsed > 0 else 0
                progress = (total_checked / total_requests) * 100
                eta = (total_requests - total_checked) / rps if rps > 0 else 0
                
                bar_width = 30
                filled = int(bar_width * min(progress, 100) / 100)
                bar = '█' * filled + '░' * (bar_width - filled)
                
                progress_html = f"""
## Scanning v{version}

### {bar} {progress:.1f}%

- Checked: {total_checked:,} / {total_requests:,}
- Speed: {rps:,.0f} req/s
- ETA: {format_time(eta)}
- Elapsed: {format_time(elapsed)}
- Hits: {len(all_hits)}
"""
                
                stats_html = f"""
## Analytics

| Metric | Value |
|--------|-------|
| Model | {model} |
| Version | {version} |
| DIN | {din_start:03d}-{din_end:03d} |
| Progress | {progress:.1f}% |
| Speed | {rps:,.0f} req/s |
| Workers | {optimal_workers} |
| Hits | {len(all_hits)} |
| Elapsed | {format_time(elapsed)} |
"""
                
                results_html = ""
                if version_hits:
                    results_html = "## FIRMWARE FOUND!\n\n"
                    for j, (ver, url, size) in enumerate(version_hits[-5:], 1):
                        results_html += f"""
### Hit #{len(all_hits) - len(version_hits) + j}
- Version: {ver}
- Size: {format_size(size)}
- URL: `{url}`
"""
                
                yield progress_html, results_html, stats_html, ""
                
                if should_stop and version_hits:
                    break
            
            if should_stop and version_hits:
                break
        
        self.scanning = False
        overall_elapsed = time.time() - overall_start
        
        summary_html = f"""
## Scan Complete!

### Performance
- Duration: {format_time(overall_elapsed)}
- Checked: {total_checked:,}
- Speed: {int(total_checked/overall_elapsed) if overall_elapsed > 0 else 0:,} req/s
- Hits: {len(all_hits)}

### Discovered Firmware
"""
        
        if all_hits:
            seen = set()
            unique_hits = []
            for hit in all_hits:
                if hit[1] not in seen:
                    seen.add(hit[1])
                    unique_hits.append(hit)
            
            for j, (ver, url, size) in enumerate(unique_hits[:20], 1):
                summary_html += f"""
**#{j}** v{ver} | {format_size(size)}
`{url}`
"""
            if len(unique_hits) > 20:
                summary_html += f"\n*...and {len(unique_hits) - 20} more*"
            
            summary_html += "\nSaved to `found_200_urls.txt`"
        else:
            summary_html += "\nNo firmware found."
        
        yield progress_html, results_html, stats_html, summary_html

# ═══════════════════════════════════════════════════
# 🚀 GPU FUNCTION FOR HUGGING FACE
# ═══════════════════════════════════════════════════

import spaces

@spaces.GPU
def gpu_scanner(input_text):
    """GPU-accelerated scanner for Hugging Face Spaces"""
    return input_text

# Initialize GPU on startup
gpu_scanner("init")

# ═══════════════════════════════════════════════════
# 🎨 GRADIO UI
# ═══════════════════════════════════════════════════

def create_ui():
    scanner = FirmwareScanner()
    
    with gr.Blocks(title="FOTA Scanner") as app:
        gr.HTML("""
        <div style="text-align: center; padding: 20px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; border-radius: 10px; margin-bottom: 20px;">
            <h1>FOTA Firmware Scanner</h1>
            <p>30-Second Completion • 5,000+ Workers • Auto-Save</p>
        </div>
        """)
        
        firmware_url = gr.Textbox(
            label="Firmware URL",
            placeholder="Enter firmware URL here...",
            lines=2,
        )
        
        with gr.Row():
            version_input = gr.Textbox(
                label="Version / Range",
                placeholder="Enter version or range",
                scale=1,
            )
            din_input = gr.Textbox(
                label="DIN / Range",
                placeholder="Enter DIN or range",
                scale=1,
            )
        
        stop_on_first = gr.Checkbox(
            label="Stop on first hit",
            value=False,
        )
        
        with gr.Row():
            scan_btn = gr.Button("START SCAN", variant="primary", size="lg")
            stop_btn = gr.Button("STOP", variant="stop", size="lg")
        
        progress_display = gr.Markdown("## Ready\n\nFill all fields and click START SCAN.")
        
        with gr.Tabs():
            with gr.Tab("Results"):
                results_display = gr.Markdown("")
            with gr.Tab("Analytics"):
                stats_display = gr.Markdown("")
            with gr.Tab("Summary"):
                summary_display = gr.Markdown("")
        
        scan_event = scan_btn.click(
            fn=scanner.start_scan,
            inputs=[firmware_url, version_input, din_input, stop_on_first],
            outputs=[progress_display, results_display, stats_display, summary_display]
        )
        
        stop_btn.click(
            fn=scanner.stop_scan,
            inputs=[],
            outputs=[progress_display],
            cancels=[scan_event]
        )
    
    return app

# ═══════════════════════════════════════════════════
# 🚀 LAUNCH
# ═══════════════════════════════════════════════════

if __name__ == "__main__":
    mp.set_start_method('spawn', force=True)
    
    app = create_ui()
    app.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
        debug=False
    )