""" Точка входа для 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()