152-fz-scanner/main.py

"""
Точка входа для 152-ФЗ Scanner.
Поддерживает запуск через CLI (один URL или CSV) и через API-сервер.
"""
import asyncio
import csv
import json
import sys
from pathlib import Path

from scanner.engine import ScannerEngine
from scanner.models import SiteScan
from scanner.output import out
from reports.generator import PDFReportGenerator


async def scan_single(url: str, checks: list[str] | None = None, deep: bool = False) -> SiteScan:
    """Просканировать один URL и вывести результат.
    
    Args:
        url: URL для сканирования
        checks: список проверок (None = QUICK_CHECKS)
        deep: запустить LLM-анализ политики после основных проверок
    """
    engine = ScannerEngine()
    result = await engine.scan_url(url, checks)

    # Выводим результаты в терминал
    out(f"\n{'='*60}")
    out(f"🔍 Сканирование: {url}")
    out(f"{'='*60}")
    out(f"📊 Балл: {result.score}/100")
    out(f"{'✅ Соответствует' if result.is_compliant else '❌ НЕ соответствует'} 152-ФЗ")
    out(f"📝 {result.summary}")
    out(f"\n{'─'*40}")

    for check in result.checks:
        emoji = {
            "pass": "✅",
            "fail": "❌",
            "skip": "⏭",
            "error": "⚠️",
        }.get(check.status.value, "  ")
        out(f"{emoji} {check.title}: {check.message}")

    out(f"{'='*60}\n")
    
    # Опциональный LLM-анализ политики
    if deep:
        out("🤖 Запуск глубокого LLM-анализа политики...")
        try:
            from scanner.checks.llm_policy import run_llm_analysis
            
            # Ищем URL политики в результатах проверок
            policy_url = None
            for check in result.checks:
                if check.name == "privacy_link" and check.status.value == "pass":
                    policy_url = check.details.get("policy_url")
                    break
            
            if policy_url:
                llm_result = await run_llm_analysis(policy_url)
                out(f"\n{'─'*40}")
                out(f"🧠 {llm_result['title']}")
                out(f"   Статус: {llm_result['status']}")
                out(f"   Риск: {llm_result['risk_level']}")
                out(f"   {llm_result['message']}")
                if llm_result.get('details', {}).get('present_count'):
                    d = llm_result['details']
                    out(f"   Разделов: {d['present_count']}/16")
                    if d.get('missing'):
                        out(f"   Отсутствуют: {', '.join(d['missing'][:5])}")
            else:
                out("   ⚠️ URL политики не найден — LLM-анализ пропущен")
        except Exception as e:
            out(f"   ⚠️ Ошибка LLM-анализа: {e}")
    
    return result


def is_likely_hotel(url: str) -> bool:
    """Проверяет, похож ли URL на сайт отеля (фильтр не-отелей)."""
    domain = url.replace("https://", "").replace("http://", "").split("/")[0].lower()
    
    # Известные не-отели — пропускаем их
    not_hotel_patterns = [
        "wikipedia", "google", "example", "demo", "test", "toscrape",
        "herokuapp", "phpstore", "myshopify", "wordpress", "joomla",
        "drupal", "typo3", "prestashop", "magento", "opencart",
        "wix", "squarespace", "webflow", "shopify", "woocommerce",
        "tilda", "constructor", "builder", "playwright", "node",
        "localhost", "127.0.0.1", "ip-api", "ipify", "my-ip",
        "reg.ru", "nic.ru", "beget", "timeweb", "selectel",
    ]
    
    for pattern in not_hotel_patterns:
        if pattern in domain:
            return False
    
    # Разрешаем всё остальное — пользователь курирует список
    return True


def scan_csv(filepath: str, checks: list[str] | None = None):
    """Парксить CSV-файл с URL и запустить сканирование."""
    csv_path = Path(filepath)
    if not csv_path.exists():
        out(f"❌ Файл не найден: {filepath}")
        sys.exit(1)

    urls = []
    skipped = []
    with open(csv_path, "r", encoding="utf-8") as f:
        reader = csv.reader(f)
        for row in reader:
            if row:
                url = row[0].strip()
                if url and url.startswith(("http://", "https://")):
                    if is_likely_hotel(url):
                        urls.append(url)
                    else:
                        skipped.append(url)

    if skipped:
        out(f"⚠️ Пропущено {len(skipped)} URL (не похожи на отели):")
        for u in skipped[:5]:
            out(f"    {u}")
        if len(skipped) > 5:
            out(f"    ...и ещё {len(skipped) - 5}")
        out()

    if not urls:
        out("❌ В CSV не найдено валидных URL")
        sys.exit(1)

    out(f"📋 Загружено {len(urls)} URL из {csv_path}")

    async def batch_scan():
        engine = ScannerEngine()
        results = await engine.scan_urls_batch(urls, checks)

        output = []
        for r in results:
            output.append({
                "url": r.url,
                "score": r.score,
                "is_compliant": r.is_compliant,
                "summary": r.summary,
                "checks": [c.model_dump() for c in r.checks],
            })

        # Сохраняем JSON
        output_dir = Path(__file__).parent / "data" / "output"
        output_dir.mkdir(parents=True, exist_ok=True)

        json_path = output_dir / "results.json"
        with open(json_path, "w", encoding="utf-8") as f:
            json.dump(output, f, ensure_ascii=False, indent=2)

        out(f"\n📄 Результаты сохранены в: {json_path}")

        # Генерируем PDF для каждого сайта
        generator = PDFReportGenerator()
        reports_dir = output_dir / "reports"

        for result in output:
            try:
                pdf_path = generator.generate(result, str(reports_dir))
                out(f"  📄 PDF: {pdf_path}")
            except Exception as e:
                out(f"  ⚠️ Ошибка генерации PDF для {result['url']}: {e}")

        # Сводка
        compliant = sum(1 for r in output if r["is_compliant"])
        non_compliant = len(output) - compliant
        out(f"\n📊 Сводка:")
        out(f"  Всего сайтов: {len(output)}")
        out(f"  ✅ Соответствуют: {compliant}")
        out(f"  ❌ НЕ соответствуют: {non_compliant}")

        # Показываем топ нарушений
        out(f"\n🔥 Топ нарушений:")
        check_failures = {}
        for r in output:
            for check in r["checks"]:
                if check["status"] == "fail":
                    name = check["title"]
                    check_failures[name] = check_failures.get(name, 0) + 1

        for name, count in sorted(check_failures.items(), key=lambda x: -x[1]):
            out(f"  ❌ {name}: {count} сайтов")

    asyncio.run(batch_scan())


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="152-ФЗ Scanner")
    subparsers = parser.add_subparsers(dest="command")

    # Подкоманда: single
    single_parser = subparsers.add_parser("scan", help="Сканировать один URL")
    single_parser.add_argument("url", help="URL сайта для сканирования")
    single_parser.add_argument("--checks", nargs="+", help="Какие проверки запустить")
    single_parser.add_argument("--deep", action="store_true", help="Запустить LLM-анализ политики после сканирования")

    # Подкоманда: csv
    csv_parser = subparsers.add_parser("csv", help="Сканировать CSV с URL")
    csv_parser.add_argument("filepath", help="Путь к CSV-файлу")
    csv_parser.add_argument("--checks", nargs="+", help="Какие проверки запустить")

    # Подкоманда: server
    server_parser = subparsers.add_parser("server", help="Запустить API-сервер")
    server_parser.add_argument("--port", type=int, default=8000, help="Порт (по умолчанию 8000)")
    server_parser.add_argument("--host", default="0.0.0.0", help="Хост (по умолчанию 0.0.0.0)")

    args = parser.parse_args()

    if args.command == "scan":
        asyncio.run(scan_single(args.url, args.checks, args.deep))
    elif args.command == "csv":
        scan_csv(args.filepath, args.checks)
    elif args.command == "server":
        import uvicorn
        uvicorn.run("api.server:app", host=args.host, port=args.port, reload=True)
    else:
        parser.print_help()