From de18090304c76194b7163ea23af3009e901be3b8 Mon Sep 17 00:00:00 2001 From: neo <397763@gmail.com> Date: Tue, 9 Jun 2026 21:10:21 +0500 Subject: [PATCH] =?UTF-8?q?=F0=9F=8F=97=20=D0=A4=D0=B0=D0=B7=D0=B0=201-4:?= =?UTF-8?q?=20=D0=90=D1=80=D1=85=D0=B8=D1=82=D0=B5=D0=BA=D1=82=D1=83=D1=80?= =?UTF-8?q?=D0=B0,=20=D1=8F=D0=B4=D1=80=D0=BE,=20=D0=BF=D1=80=D0=BE=D0=B2?= =?UTF-8?q?=D0=B5=D1=80=D0=BA=D0=B8,=20API,=20=D0=B3=D0=B5=D0=BD=D0=B5?= =?UTF-8?q?=D1=80=D0=B0=D1=82=D0=BE=D1=80=20=D0=BE=D1=82=D1=87=D0=B5=D1=82?= =?UTF-8?q?=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - scanner/engine.py — Playwright-движок сканирования - scanner/models.py — Pydantic модели - scanner/checks/ — 5 проверок (cookie, privacy, forms, server, external) - api/server.py — FastAPI REST API - reports/generator.py — PDF через WeasyPrint + Jinja2 - main.py — CLI + сервер Проверки: ✅ cookie_banner — поиск cookie-баннера ✅ privacy_link — проверка ссылки на политику ✅ form_checkbox — проверка чекбоксов согласия ✅ server_location — геолокация сервера (RФ или нет) ✅ external_scripts — внешние скрипты аналитики --- .gitignore | 30 ++++ README.md | 133 +++++++++++++++++ api/__init__.py | 0 api/schemas.py | 37 +++++ api/server.py | 102 +++++++++++++ main.py | 153 +++++++++++++++++++ reports/__init__.py | 0 reports/generator.py | 83 +++++++++++ reports/templates/report.html.j2 | 228 +++++++++++++++++++++++++++++ requirements.txt | 6 + scanner/__init__.py | 0 scanner/checks/__init__.py | 25 ++++ scanner/checks/base.py | 76 ++++++++++ scanner/checks/cookie_banner.py | 89 +++++++++++ scanner/checks/external_scripts.py | 110 ++++++++++++++ scanner/checks/form_checkbox.py | 117 +++++++++++++++ scanner/checks/privacy_link.py | 92 ++++++++++++ scanner/checks/server_location.py | 139 ++++++++++++++++++ scanner/engine.py | 127 ++++++++++++++++ scanner/models.py | 118 +++++++++++++++ 20 files changed, 1665 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 api/__init__.py create mode 100644 api/schemas.py create mode 100644 api/server.py create mode 100644 main.py create mode 100644 reports/__init__.py create mode 100644 reports/generator.py create mode 100644 reports/templates/report.html.j2 create mode 100644 requirements.txt create mode 100644 scanner/__init__.py create mode 100644 scanner/checks/__init__.py create mode 100644 scanner/checks/base.py create mode 100644 scanner/checks/cookie_banner.py create mode 100644 scanner/checks/external_scripts.py create mode 100644 scanner/checks/form_checkbox.py create mode 100644 scanner/checks/privacy_link.py create mode 100644 scanner/checks/server_location.py create mode 100644 scanner/engine.py create mode 100644 scanner/models.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5c888d5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +.eggs/ +*.egg + +# Virtual env +.venv/ +venv/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# Data +data/output/results.json +data/output/reports/audit_*.pdf + +# Env +.env +.env.local + +# OS +.DS_Store +Thumbs.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..5aebe89 --- /dev/null +++ b/README.md @@ -0,0 +1,133 @@ +# 152-ФЗ Scanner — Compliance Audit Tool + +Автоматизированный веб-сканер для аудита сайтов на соответствие 152-ФЗ (О персональных данных). +Генерирует PDF-отчеты с рекомендациями для коммерческой продажи услуг по доработке. + +## 🏗 Архитектура + +``` +152-fz-scanner/ +├── main.py # Точка входа (CLI / FastAPI) +├── scanner/ # Ядро сканирования +│ ├── engine.py # Playwright-движок (управление браузером) +│ ├── checks/ # Модули проверок +│ │ ├── base.py # Базовый класс для всех проверок +│ │ ├── cookie_banner.py # Проверка баннера cookie +│ │ ├── privacy_link.py # Проверка ссылки на политику +│ │ ├── privacy_page.py # Проверка содержимого страницы политики +│ │ ├── form_checkbox.py # Проверка чекбокса согласия в формах +│ │ ├── server_location.py # Проверка геолокации IP (через API) +│ │ └── external_scripts.py # Проверка внешних скриптов (Google, FB, etc.) +│ └── models.py # Pydantic модели (ScanResult, CheckResult, SiteReport) +├── api/ # REST API (FastAPI) +│ ├── server.py # Основной файл сервера +│ ├── routes.py # Эндпоинты +│ └── schemas.py # Pydantic схемы для JSON +├── reports/ # Генерация отчетов +│ ├── generator.py # Логика генерации PDF +│ ├── templates/ # HTML-шаблоны отчетов +│ │ └── report.html.j2 +│ └── static/ # Статика (CSS, логотипы) +│ └── styles.css +├── data/ # Входные/выходные данные +│ ├── input/ # CSV с URL сайтов для сканирования +│ └── output/ # JSON-результаты и PDF-отчеты +│ ├── results.json +│ └── reports/ +│ └── audit_*.pdf +├── requirements.txt # Зависимости +├── .gitignore +└── README.md +``` + +## 🛠 Стек технологий + +| Компонент | Технология | Назначение | +|-----------|-----------|-----------| +| Движок браузера | `playwright` | Рендеринг JS, эмуляция пользователя, анти-бот (stealth) | +| HTTP-клиент | `curl_cffi` | Быстрые проверки (HTTPS, redirect, IP-геолокация) | +| Web-фреймворк | `fastapi` + `uvicorn` | REST API для запуска сканирования и получения отчетов | +| ORM/Модели | `pydantic` | Валидация данных, JSON-схемы | +| PDF-генерация | `weasyprint` | Конвертация HTML-шаблона в PDF-отчет | +| Шаблоны | `jinja2` | HTML-генерация отчета | +| CSV/Парсинг | `pandas` | Чтение входных CSV (список сайтов) | + +## 📋 План разработки + +### Фаза 1: Ядро (День 1-2) +- [ ] Настройка окружения (venv, requirements.txt) +- [ ] Написание базовой модели `ScanResult` (Pydantic) +- [ ] Реализация `scanner/engine.py` (Playwright + stealth) +- [ ] Реализация первой проверки: `cookie_banner.py` +- [ ] CLI-утилита: `python main.py --url https://example.com --check cookie` +- [ ] Тест: сканирование 10 сайтов, вывод CSV + +### Фаза 2: Расширение проверок (День 3-4) +- [ ] `privacy_link.py` — поиск ссылки в footer/header +- [ ] `privacy_page.py` — парсинг содержимого страницы политики +- [ ] `form_checkbox.py` — проверка всех форм на чекбоксы согласия +- [ ] `server_location.py` — проверка IP через MaxMind DB или API +- [ ] `external_scripts.py` — поиск Google Analytics, FB Pixel и т.д. +- [ ] Интеграция: сканирование сайта по всем 5 проверкам +- [ ] Генерация JSON-результата + +### Фаза 3: Генератор отчетов (День 5) +- [ ] HTML-шаблон отчета (Jinja2 + CSS) +- [ ] `reports/generator.py` — конвертация JSON в PDF (WeasyPrint) +- [ ] Настройка стилей (брендинг, цвета, таблица рисков) +- [ ] Тест: генерация PDF-отчета для одного сайта +- [ ] Тест: генерация 10 PDF-отчетов в batch-режиме + +### Фаза 4: API Сервер (День 6) +- [ ] `main.py` — FastAPI сервер (uvicorn) +- [ ] Эндпоинт `POST /api/scan` — запуск сканирования +- [ ] Эндпоинт `GET /api/report/{scan_id}` — скачивание PDF +- [ ] Эндпоинт `GET /api/results` — список всех сканирований +- [ ] Эндпоинт `POST /api/scan/csv` — загрузка CSV с URL + авто-сканирование +- [ ] Документация (Swagger UI доступен по умолчанию) + +### Фаза 5: Интеграция и доработка (День 7) +- [ ] Проверка работы в headless-режиме на сервере +- [ ] Настройка логирования +- [ ] Добавление rate-limiting и timeout (чтобы не висели сайты) +- [ ] Тестирование на реальном списке 50-100 сайтов Уфы +- [ ] Документация в репозитории + +## 🔑 Ключевые решения + +1. **Playwright вместо requests**: Необходим для рендеринга JS (cookie-баннеры появляются динамически). +2. **Stealth-режим**: Обязательно для обхода простых анти-бот-защит (Cloudflare, Akamai). +3. **Отдельные модули проверок**: Каждая проверка — отдельный класс, наследующий `BaseCheck`. Легко добавлять новые. +4. **Async-first**: Playwright работает асинхронно, что ускоряет пакетное сканирование в 5-10 раз. +5. **PDF через HTML/CSS**: Проще стилизовать, чем генерировать PDF программно (weasyprint). + +## 📦 Зависимости + +``` +fastapi>=0.104.0 +uvicorn[standard]>=0.24.0 +playwright>=1.40.0 +curl_cffi>=0.6.0 +pydantic>=2.5.0 +weasyprint>=60.0 +jinja2>=3.1.2 +pandas>=2.1.0 +maxminddb>=2.3.0 +``` + +## 🚀 Быстрый старт + +```bash +# Установка +pip install -r requirements.txt +playwright install chromium + +# Запуск API-сервера +uvicorn main:app --host 0.0.0.0 --port 8000 + +# Тестирование (CLI) +python main.py --url https://example.com + +# Сканирование CSV +python main.py --csv data/input/sites.csv +``` diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/schemas.py b/api/schemas.py new file mode 100644 index 0000000..e897b7a --- /dev/null +++ b/api/schemas.py @@ -0,0 +1,37 @@ +from pydantic import BaseModel, Field +from typing import Optional + + +class HealthResponse(BaseModel): + status: str + version: str = "1.0.0" + checks_count: int = 5 + + +class ScanRequest(BaseModel): + url: str + checks: Optional[list[str]] = None + + +class CSVScanRequest(BaseModel): + urls: list[str] + checks: Optional[list[str]] = None + + +class ScanResultResponse(BaseModel): + scan_id: str + url: str + score: int + is_compliant: bool + summary: str + checks_passed: int + checks_failed: int + checks_errors: int + + +class BatchScanResultResponse(BaseModel): + scan_id: str + total: int + results: list[ScanResultResponse] + compliant_count: int + non_compliant_count: int diff --git a/api/server.py b/api/server.py new file mode 100644 index 0000000..fb84b28 --- /dev/null +++ b/api/server.py @@ -0,0 +1,102 @@ +""" +FastAPI сервер для 152-ФЗ Scanner. +REST API для запуска сканирования и получения отчетов. +""" +from fastapi import FastAPI, HTTPException, BackgroundTasks +from scanner.engine import ScannerEngine +from api.schemas import ( + HealthResponse, + ScanRequest, + ScanResultResponse, + BatchScanResultResponse, +) + +app = FastAPI( + title="152-ФЗ Scanner API", + description="Автоматизированный сканер для аудита сайтов по 152-ФЗ", + version="1.0.0", +) + +scanner = ScannerEngine() + + +@app.get("/health", response_model=HealthResponse, tags=["Health"]) +async def health_check(): + """Проверка состояния API.""" + return HealthResponse(status="ok") + + +@app.post("/api/scan", tags=["Scan"]) +async def scan_single(request: ScanRequest): + """ + Просканировать один URL. + + Возвращает результаты всех проверок и итоговый балл. + """ + try: + result = await scanner.scan_url(request.url, request.checks) + + checks_passed = sum(1 for c in result.checks if c.status.value == "pass") + checks_failed = sum(1 for c in result.checks if c.status.value == "fail") + checks_errors = sum(1 for c in result.checks if c.status.value == "error") + + return ScanResultResponse( + scan_id=result.url.replace("://", "_").replace("/", "_")[:30], + url=result.url, + score=result.score, + is_compliant=result.is_compliant, + summary=result.summary, + checks_passed=checks_passed, + checks_failed=checks_failed, + checks_errors=checks_errors, + ) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.post("/api/scan/batch", tags=["Scan"]) +async def scan_batch(request: BatchScanResultResponse): + """ + Просканировать несколько URL. + + Запрос: {"urls": ["https://site1.ru", "https://site2.ru"], "checks": [...]} + """ + from scanner.engine import scanner_engine + + try: + results = await scanner_engine.scan_urls_batch(request.urls, request.checks) + + scan_results = [] + for r in results: + checks_passed = sum(1 for c in r.checks if c.status.value == "pass") + checks_failed = sum(1 for c in r.checks if c.status.value == "fail") + checks_errors = sum(1 for c in r.checks if c.status.value == "error") + + scan_results.append(ScanResultResponse( + scan_id=r.url.replace("://", "_").replace("/", "_")[:30], + url=r.url, + score=r.score, + is_compliant=r.is_compliant, + summary=r.summary, + checks_passed=checks_passed, + checks_failed=checks_failed, + checks_errors=checks_errors, + )) + + compliant_count = sum(1 for r in scan_results if r.is_compliant) + + return BatchScanResultResponse( + scan_id="batch", + total=len(scan_results), + results=scan_results, + compliant_count=compliant_count, + non_compliant_count=len(scan_results) - compliant_count, + ) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/results", tags=["Results"]) +async def get_results(): + """Получить список всех выполненных сканирований.""" + return {"message": "Результаты хранятся в памяти (v1). В будущем будет JSON-файл или БД."} diff --git a/main.py b/main.py new file mode 100644 index 0000000..039d87a --- /dev/null +++ b/main.py @@ -0,0 +1,153 @@ +""" +Точка входа для 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 reports.generator import PDFReportGenerator + + +async def scan_single(url: str, checks: list[str] | None = None) -> SiteScan: + """Просканировать один URL и вывести результат.""" + engine = ScannerEngine() + result = await engine.scan_url(url, checks) + + # Выводим результаты в терминал + print(f"\n{'='*60}") + print(f"🔍 Сканирование: {url}") + print(f"{'='*60}") + print(f"📊 Балл: {result.score}/100") + print(f"{'✅ Соответствует' if result.is_compliant else '❌ НЕ соответствует'} 152-ФЗ") + print(f"📝 {result.summary}") + print(f"\n{'─'*40}") + + for check in result.checks: + emoji = { + "pass": "✅", + "fail": "❌", + "skip": "⏭", + "error": "⚠️", + }.get(check.status.value, " ") + print(f"{emoji} {check.title}: {check.message}") + + print(f"{'='*60}\n") + return result + + +def scan_csv(filepath: str): + """Парксить CSV-файл с URL и запустить сканирование.""" + csv_path = Path(filepath) + if not csv_path.exists(): + print(f"❌ Файл не найден: {filepath}") + sys.exit(1) + + urls = [] + 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://")): + urls.append(url) + + if not urls: + print("❌ В CSV не найдено валидных URL") + sys.exit(1) + + print(f"📋 Загружено {len(urls)} URL из {csv_path}") + + async def batch_scan(): + engine = ScannerEngine() + results = await engine.scan_urls_batch(urls) + + 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) + + print(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)) + print(f" 📄 PDF: {pdf_path}") + except Exception as e: + print(f" ⚠️ Ошибка генерации PDF для {result['url']}: {e}") + + # Сводка + compliant = sum(1 for r in output if r["is_compliant"]) + non_compliant = len(output) - compliant + print(f"\n📊 Сводка:") + print(f" Всего сайтов: {len(output)}") + print(f" ✅ Соответствуют: {compliant}") + print(f" ❌ НЕ соответствуют: {non_compliant}") + + # Показываем топ нарушений + print(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]): + print(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="Какие проверки запустить") + + # Подкоманда: csv + csv_parser = subparsers.add_parser("csv", help="Сканировать CSV с URL") + csv_parser.add_argument("filepath", help="Путь к CSV-файлу") + + # Подкоманда: 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)) + elif args.command == "csv": + scan_csv(args.filepath) + elif args.command == "server": + import uvicorn + uvicorn.run("api.server:app", host=args.host, port=args.port, reload=True) + else: + parser.print_help() diff --git a/reports/__init__.py b/reports/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/reports/generator.py b/reports/generator.py new file mode 100644 index 0000000..dc2997e --- /dev/null +++ b/reports/generator.py @@ -0,0 +1,83 @@ +""" +Генератор PDF-отчетов из JSON-данных сканирования. +Использует Jinja2 + WeasyPrint для создания красивых отчетов. +""" +from __future__ import annotations + +import os +from datetime import datetime +from pathlib import Path +from typing import Any + +from jinja2 import Environment, FileSystemLoader +from weasyprint import HTML + +# Путь к корню проекта +BASE_DIR = Path(__file__).parent.parent +TEMPLATE_DIR = Path(__file__).parent / "templates" + + +class PDFReportGenerator: + """Генерирует PDF-отчеты из результатов сканирования.""" + + def __init__(self): + self.jinja_env = Environment( + loader=FileSystemLoader(str(TEMPLATE_DIR)), + autoescape=True, + ) + + def generate(self, scan_result: dict[str, Any], output_dir: str | None = None) -> Path: + """ + Создать PDF-отчет из результатов сканирования. + + Args: + scan_result: dict с данными сканирования (SiteScan.model_dump()) + output_dir: папка для сохранения (по умолчанию data/output/reports) + + Returns: + Path до созданного PDF-файла + """ + if output_dir is None: + output_dir = BASE_DIR / "data" / "output" / "reports" + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + # Подставляем форматированные данные в шаблон + template = self.jinja_env.get_template("report.html.j2") + html_content = template.render(**self._prepare_context(scan_result)) + + # Генерируем PDF + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + domain_safe = scan_result.get("url", "unknown").replace("://", "_").replace("/", "_")[:30] + pdf_path = output_dir / f"audit_{timestamp}_{domain_safe}.pdf" + + HTML(string=html_content, base_url=str(TEMPLATE_DIR)).write_pdf(str(pdf_path)) + + return pdf_path + + @staticmethod + def _prepare_context(scan_result: dict) -> dict: + """Подготавливает контекст для шаблона.""" + checks = scan_result.get("checks", []) + is_compliant = scan_result.get("is_compliant", True) + score = scan_result.get("score", 0) + scanned_at = scan_result.get("scanned_at", "") + + # Форматируем дату + if isinstance(scanned_at, str): + try: + dt = datetime.fromisoformat(scanned_at.replace("Z", "+00:00")) + scanned_at = dt.strftime("%d.%m.%Y %H:%M") + except (ValueError, AttributeError): + pass + + return { + "site_scan": scan_result, + "url": scan_result.get("url", ""), + "score": score, + "is_compliant": is_compliant, + "summary": scan_result.get("summary", ""), + "checks": checks, + "scanned_at": scanned_at, + "scanner_version": "1.0.0", + } diff --git a/reports/templates/report.html.j2 b/reports/templates/report.html.j2 new file mode 100644 index 0000000..0458c8c --- /dev/null +++ b/reports/templates/report.html.j2 @@ -0,0 +1,228 @@ + + + + + Аудит по 152-ФЗ — {{ site_scan.url }} + + + +
+

📋 Аудит соответствия 152-ФЗ

+

Автоматизированный сканер | {{ site_scan.url }}

+

Дата проверки: {{ scanned_at }}

+

Версия сканера: {{ scanner_version }}

+
+ +
+
{{ score }}
+
+ {{ '✅ Сайт соответствует 152-ФЗ' if is_compliant else '❌ Сайт НЕ соответствует 152-ФЗ' }} +
+
+ +
+

📊 Резюме проверки

+

{{ summary }}

+
+ +

📝 Результаты проверок

+ + + + + + + + + + + {% for check in checks %} + + + + + + + {% endfor %} + +
ПроверкаСтатусРискДетали
{{ check.title }} + {% if check.status == 'pass' %} + ✅ Пройдено + {% elif check.status == 'fail' %} + ❌ Не пройдено + {% elif check.status == 'skip' %} + ⏭ Пропущено + {% elif check.status == 'error' %} + ⚠️ Ошибка + {% endif %} + + {% if check.risk_level == 'critical' %} + 🔴 Критический + {% elif check.risk_level == 'high' %} + 🟠 Высокий + {% elif check.risk_level == 'medium' %} + 🟡 Средний + {% else %} + 🟢 Низкий + {% endif %} + {{ check.message }}
+ +
+

Нужна помощь с приведением сайта в соответствие 152-ФЗ?

+

Мы исправим все найденные нарушения, подготовим документы и настроим cookie-баннеры.

+

💰 Стоимость: от 7 000 ₽ (зависит от объема нарушений)

+
+ 📧 Написать на email + 💬 Telegram +
+
+ +
+

Документ сгенерирован автоматически. Не является юридической консультацией.

+

© 2026 152-ФЗ Scanner | Powered by AI Compliance

+
+ + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b972f82 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +fastapi>=0.104.0 +uvicorn[standard]>=0.24.0 +playwright>=1.40.0 +pydantic>=2.5.0 +jinja2>=3.1.2 +weasyprint>=60.0 diff --git a/scanner/__init__.py b/scanner/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scanner/checks/__init__.py b/scanner/checks/__init__.py new file mode 100644 index 0000000..e614249 --- /dev/null +++ b/scanner/checks/__init__.py @@ -0,0 +1,25 @@ +from scanner.checks.base import BaseCheck +from scanner.checks.cookie_banner import CookieBannerCheck +from scanner.checks.privacy_link import PrivacyLinkCheck +from scanner.checks.form_checkbox import FormCheckboxCheck +from scanner.checks.server_location import ServerLocationCheck +from scanner.checks.external_scripts import ExternalScriptsCheck + +# Реестр всех проверок — новые проверки автоматически регистрируются здесь +ALL_CHECKS: dict[str, type[BaseCheck]] = { + "cookie_banner": CookieBannerCheck, + "privacy_link": PrivacyLinkCheck, + "form_checkbox": FormCheckboxCheck, + "server_location": ServerLocationCheck, + "external_scripts": ExternalScriptsCheck, +} + +__all__ = [ + "BaseCheck", + "CookieBannerCheck", + "PrivacyLinkCheck", + "FormCheckboxCheck", + "ServerLocationCheck", + "ExternalScriptsCheck", + "ALL_CHECKS", +] diff --git a/scanner/checks/base.py b/scanner/checks/base.py new file mode 100644 index 0000000..47f3ccc --- /dev/null +++ b/scanner/checks/base.py @@ -0,0 +1,76 @@ +""" +Базовый класс для всех проверок 152-ФЗ. +Каждая проверка наследуется от этого класса и реализует метод check(). +""" +from __future__ import annotations + +from abc import ABC, abstractmethod +from playwright.async_api import Page + +from scanner.models import CheckResult, CheckStatus, RiskLevel + + +class BaseCheck(ABC): + """Базовый класс для проверки соответствия 152-ФЗ.""" + + name: str = "base" # Уникальное имя (используется в реестре) + title: str = "Base Check" # Отображаемое название + description: str = "" # Описание что проверяет + risk_on_fail: RiskLevel = RiskLevel.LOW # Уровень риска при провале + default_weight: int = 10 # Вес (влияние на общий балл) + + @abstractmethod + async def check(self, page: Page, url: str) -> CheckResult: + """ + Выполнить проверку сайта. + + Args: + page: Playwright Page объект (сайт уже загружен) + url: URL проверяемого сайта + + Returns: + CheckResult с результатом проверки + """ + pass + + def _pass_result(self, message: str = "", details: dict = {}) -> CheckResult: + """Создать успешный результат.""" + return CheckResult( + name=self.name, + title=self.title, + status=CheckStatus.PASS, + risk_level=self.risk_on_fail, + message=message or f"✓ {self.title} — проверено, нарушений не найдено", + details=details, + ) + + def _fail_result(self, message: str, details: dict = {}, risk_level: RiskLevel | None = None) -> CheckResult: + """Создать результат с провалом.""" + return CheckResult( + name=self.name, + title=self.title, + status=CheckStatus.FAIL, + risk_level=risk_level or self.risk_on_fail, + message=f"✗ {message}", + details=details, + ) + + def _skip_result(self, message: str = "") -> CheckResult: + """Создать результат с пропуском.""" + return CheckResult( + name=self.name, + title=self.title, + status=CheckStatus.SKIP, + risk_level=RiskLevel.LOW, + message=message or f"⏭ {self.title} — пропущено", + ) + + def _error_result(self, message: str) -> CheckResult: + """Создать результат с ошибкой.""" + return CheckResult( + name=self.name, + title=self.title, + status=CheckStatus.ERROR, + risk_level=RiskLevel.HIGH, + message=f"⚠ {message}", + ) diff --git a/scanner/checks/cookie_banner.py b/scanner/checks/cookie_banner.py new file mode 100644 index 0000000..4ea7608 --- /dev/null +++ b/scanner/checks/cookie_banner.py @@ -0,0 +1,89 @@ +""" +Проверка наличия cookie-баннера (уведомления о cookies). +Согласно 152-ФЗ, сайт обязан получать согласие пользователя +на обработку персональных данных, включая cookies. +""" +from __future__ import annotations + +import re +from playwright.async_api import Page + +from scanner.checks.base import BaseCheck +from scanner.models import CheckResult, RiskLevel + + +class CookieBannerCheck(BaseCheck): + """Проверяет наличие cookie-баннера на сайте.""" + + name = "cookie_banner" + title = "Cookie-баннер" + description = "Проверяет, отображается ли на сайте уведомление о cookie" + risk_on_fail = RiskLevel.HIGH + default_weight = 15 + + # Паттерны для поиска cookie-баннера + COOKIE_PATTERNS = [ + r"cookie", + r"куки", + r"уведомление\s*о\s*cookie", + r"согласие\s*на\s*использование\s*cookie", + r"we\s*use\s*cookies", + r"privacy\s*policy", + r"данные\s*пользователя", + r"обработка\s*персональных\s*данных", + r"personal\s*data", + r"consent", + r"согласие", + ] + + async def check(self, page: Page, url: str) -> CheckResult: + try: + # Получаем весь текст страницы (включая динамически загруженный) + page_text = await page.evaluate("() => document.body.innerText") + page_html = await page.evaluate("() => document.documentElement.outerHTML") + + # Ищем cookie-баннер по видимым элементам + cookie_elements = await page.query_selector_all( + '[id*="cookie"], [class*="cookie"], [id*="consent"], [class*="consent"], [id*="privacy"], [class*="privacy"], [id*="policy"], [class*="policy"]' + ) + + found_banner = False + banner_texts = [] + + for el in cookie_elements: + text = await el.inner_text() + if text: + banner_texts.append(text[:200]) + + # Также ищем по тексту всей страницы + has_keyword = any( + re.search(pattern, page_text, re.IGNORECASE | re.UNICODE) + for pattern in self.COOKIE_PATTERNS + ) + + if banner_texts: + found_banner = True + return self._pass_result( + message=f"Найден cookie-баннер ({len(banner_texts)} элементов)", + details={ + "banner_count": len(banner_texts), + "sample_text": banner_texts[0][:100], + }, + ) + + if has_keyword: + return self._pass_result( + message="На странице найдены упоминания cookie/согласия", + details={"keyword_found": True}, + ) + + return self._fail_result( + message="Cookie-баннер не найден. Сайт собирает данные без явного согласия.", + details={ + "page_text_sample": page_text[:200], + "search_patterns": self.COOKIE_PATTERNS, + }, + ) + + except Exception as e: + return self._error_result(message=f"Ошибка при проверке cookie-баннера: {str(e)}") diff --git a/scanner/checks/external_scripts.py b/scanner/checks/external_scripts.py new file mode 100644 index 0000000..d2fc6ec --- /dev/null +++ b/scanner/checks/external_scripts.py @@ -0,0 +1,110 @@ +""" +Проверка внешних скриптов, собирающих данные пользователей. +Google Analytics, Facebook Pixel, Яндекс.Метрика и другие — +отправляют данные на сервера за пределами РФ. +""" +from __future__ import annotations + +import re +from playwright.async_api import Page + +from scanner.checks.base import BaseCheck +from scanner.models import CheckResult, RiskLevel + + +class ExternalScriptsCheck(BaseCheck): + """Проверяет наличие внешних скриптов, передающих данные за пределы РФ.""" + + name = "external_scripts" + title = "Внешние скрипты аналитики" + description = "Ищет скрипты Google Analytics, Facebook Pixel и другие иностранные сервисы" + risk_on_fail = RiskLevel.HIGH + default_weight = 15 + + EXTERNAL_PATTERNS = [ + r"google-analytics\.com", + r"googletagmanager\.com", + r"googlesyndication\.com", + r"googleadservices\.com", + r"googleapis\.com", + r"google\.com/recaptcha", + r"facebook\.com/tr", + r"fbcdn\.net", + r"connect\.facebook\.net", + r"mc\.yandex\.ru", + r"monitoring\.yandex\.ru", + r"top\.mail\.ru", + r"clarity\.microsoft\.com", + r"hotjar\.com", + r"mixpanel\.com", + r"segment\.com", + r"amplitude\.com", + r"zendesk\.com", + r"intercom\.io", + r"chatra\.com", + r"tilda\.cdn", + r"wix\.com", + r"wordpress\.com", + r"shopify\.com", + ] + + async def check(self, page: Page, url: str) -> CheckResult: + try: + html = await page.evaluate("() => document.documentElement.outerHTML") + + found_scripts = [] + for pattern in self.EXTERNAL_PATTERNS: + matches = re.findall(pattern, html, re.IGNORECASE) + if matches: + found_scripts.append({ + "pattern": pattern, + "matches": len(matches), + }) + + if not found_scripts: + return self._pass_result( + message="Внешние скрипты аналитики не найдены.", + ) + + critical_services = [ + "google-analytics", + "googletagmanager", + "facebook", + "clarity", + "segment", + "amplitude", + ] + + has_critical = False + for script in found_scripts: + if any(re.search(c, script["pattern"]) for c in critical_services): + has_critical = True + break + + if has_critical: + return self._fail_result( + message=( + f"Найдены внешние скрипты передачи данных: " + f"{', '.join(s['pattern'] for s in found_scripts)}. " + f"Они передают персональные данные за пределы РФ." + ), + details={ + "scripts_found": found_scripts, + "has_critical": True, + }, + ) + + return self._fail_result( + message=( + f"Найдены внешние скрипты: " + f"{', '.join(s['pattern'] for s in found_scripts)}. " + f"Требуется согласие пользователя." + ), + details={ + "scripts_found": found_scripts, + "has_critical": False, + }, + ) + + except Exception as e: + return self._error_result(message=f"Ошибка: {str(e)}") diff --git a/scanner/checks/form_checkbox.py b/scanner/checks/form_checkbox.py new file mode 100644 index 0000000..0a18bf7 --- /dev/null +++ b/scanner/checks/form_checkbox.py @@ -0,0 +1,117 @@ +""" +Проверка наличия чекбокса согласия на обработку персональных данных. +Согласно 152-ФЗ, обработка данных допустима только при наличии согласия субъекта. +Формы на сайте (обратная связь, регистрация, подписка и т.д.) +должны содержать явное согласие пользователя. +""" +from __future__ import annotations + +import re +from playwright.async_api import Page + +from scanner.checks.base import BaseCheck +from scanner.models import CheckResult, RiskLevel + + +class FormCheckboxCheck(BaseCheck): + """Проверяет наличие чекбокса согласия в формах на сайте.""" + + name = "form_checkbox" + title = "Чекбокс согласия в формах" + description = "Проверяет, есть ли чекбокс согласия на обработку ПД в формах" + risk_on_fail = RiskLevel.CRITICAL + default_weight = 20 + + CONSENT_PATTERNS = [ + r"согласие\s*на\s*обработку\s*данных", + r"согласие\s*на\s*обработку\s*персональных?\s*данных?", + r"я\s*согласен\s*на\s*", + r"agree\s*to\s*the\s*", + r"agree\s*to\s*privacy", + r"я\s*принимаю\s*условия", + r"processing\s*of\s*personal\s*data", + r"обработк[аею]\s*персональных?\s*данных?", + r"политик[ууе]?\s*конфиденциальност[иы]", + ] + + async def check(self, page: Page, url: str) -> CheckResult: + try: + # Получаем все формы на странице + forms = await page.query_selector_all("form") + + if not forms: + return self._skip_result( + message="Формы на странице не найдены.", + ) + + forms_needing_consent = 0 + forms_with_consent = 0 + forms_without_consent = [] + + for form_idx, form in enumerate(forms): + # Проверяем, есть ли чекбокс в форме + checkboxes = await form.query_selector_all('input[type="checkbox"]') + + has_consent = False + checkbox_texts = [] + + for cb in checkboxes: + cb_text = await cb.inner_text() if not await cb.get_attribute("value") else await cb.get_attribute("value") + cb_label = await form.evaluate( + f'(el) => {{\n' + f' const cb = el.closest("label") || el.parentElement;\n' + f' return cb ? cb.innerText : "";\n' + f'}}', form + ) + + checkbox_texts.append(cb_text or cb_label or "без текста") + + combined = f"{cb_text} {cb_label}".lower() + if any(re.search(p, combined, re.UNICODE | re.IGNORECASE) for p in self.CONSENT_PATTERNS): + has_consent = True + + forms_needing_consent += 1 + if has_consent: + forms_with_consent += 1 + else: + forms_without_consent.append({ + "form_index": form_idx, + "checkboxes": checkbox_texts[:3], + }) + + if forms_with_consent == forms_needing_consent: + return self._pass_result( + message=f"Все формы ({forms_with_consent}) содержат чекбокс согласия.", + details={ + "forms_count": forms_with_consent, + "checked_forms": forms_with_consent, + }, + ) + + if forms_with_consent == 0: + return self._fail_result( + message=( + f"Формы ({forms_needing_consent}) не содержат чекбокс согласия на обработку ПД. " + f"Это прямое нарушение 152-ФЗ." + ), + details={ + "total_forms": forms_needing_consent, + "without_consent": forms_without_consent, + "sample_texts": [f["checkboxes"][0] for f in forms_without_consent[:3]], + }, + ) + + # Частичное покрытие + return self._fail_result( + message=( + f"Только {forms_with_consent} из {forms_needing_consent} форм содержат чекбокс согласия." + ), + details={ + "total_forms": forms_needing_consent, + "with_consent": forms_with_consent, + "without_consent_count": forms_needing_consent - forms_with_consent, + }, + ) + + except Exception as e: + return self._error_result(message=f"Ошибка при проверке чекбоксов: {str(e)}") diff --git a/scanner/checks/privacy_link.py b/scanner/checks/privacy_link.py new file mode 100644 index 0000000..fea155e --- /dev/null +++ b/scanner/checks/privacy_link.py @@ -0,0 +1,92 @@ +""" +Проверка наличия ссылки на Политику конфиденциальности. +Согласно 152-ФЗ, оператор обязан опубликовать документ, +определяющий его политику в отношении обработки персональных данных. +""" +from __future__ import annotations + +import re +from urllib.parse import urljoin + +from playwright.async_api import Page + +from scanner.checks.base import BaseCheck +from scanner.models import CheckResult, RiskLevel + + +class PrivacyLinkCheck(BaseCheck): + """Проверяет наличие и доступность ссылки на Политику конфиденциальности.""" + + name = "privacy_link" + title = "Ссылка на Политику конфиденциальности" + description = "Проверяет, есть ли на сайте ссылка на политику конфиденциальности" + risk_on_fail = RiskLevel.HIGH + default_weight = 15 + + PRIVACY_PATTERNS = [ + r"privacy", + r"политик[аые]", + r"конфиденциальн[аыо]", + r"персональные\s*данные", + r"personal\s*data", + r"data\s*policy", + r"обработка\s*данных", + r"user\s*privacy", + ] + + async def check(self, page: Page, url: str) -> CheckResult: + try: + # Получаем все ссылки на странице + links = await page.query_selector_all("a[href]") + link_texts = [] + + for link in links: + href = await link.get_attribute("href") or "" + text = (await link.inner_text()).strip() + if text: + link_texts.append({"text": text, "href": href}) + + # Ищем ссылки на политику + found_links = [] + for lt in link_texts: + combined = (lt["text"] + " " + lt["href"]).lower() + if any(re.search(p, combined, re.UNICODE) for p in self.PRIVACY_PATTERNS): + full_url = urljoin(url, lt["href"]) + found_links.append({"text": lt["text"], "url": full_url}) + + if not found_links: + return self._fail_result( + message="Ссылка на Политику конфиденциальности не найдена.", + details={"total_links": len(link_texts), "page_text_sample": link_texts[:3]}, + ) + + # Проверяем, что ссылка ведет на рабочую страницу + policy_url = found_links[0]["url"] + try: + await page.goto(policy_url, timeout=5000) + policy_text = await page.evaluate("() => document.body.innerText") + await page.goto(url, timeout=5000) # Возвращаемся на главную + + if len(policy_text) < 50: + return self._fail_result( + message=f"Страница политики ({policy_url}) слишком короткая ({len(policy_text)} символов).", + details={"policy_url": policy_url, "char_count": len(policy_text)}, + ) + + return self._pass_result( + message=f"Найдена ссылка на Политику конфиденциальности ({policy_url})", + details={ + "policy_url": policy_url, + "link_text": found_links[0]["text"], + "char_count": len(policy_text), + }, + ) + + except Exception as e: + return self._fail_result( + message=f"Страница политики не доступна: {str(e)}", + details={"policy_url": policy_url}, + ) + + except Exception as e: + return self._error_result(message=f"Ошибка при проверке ссылки на политику: {str(e)}") diff --git a/scanner/checks/server_location.py b/scanner/checks/server_location.py new file mode 100644 index 0000000..9846668 --- /dev/null +++ b/scanner/checks/server_location.py @@ -0,0 +1,139 @@ +""" +Проверка геолокации сервера. +Согласно 152-ФЗ (ст. 18), сбор персональных данных граждан РФ +должен производиться с использованием баз данных, находящихся на территории РФ. +""" +from __future__ import annotations + +from playwright.async_api import Page + +from scanner.checks.base import BaseCheck +from scanner.models import CheckResult, RiskLevel + + +class ServerLocationCheck(BaseCheck): + """Проверяет, находится ли сервер сайта в России.""" + + name = "server_location" + title = "Геолокация сервера" + description = "Проверяет, находится ли сервер сайта на территории РФ" + risk_on_fail = RiskLevel.CRITICAL + default_weight = 25 + + # Популярные иностранные провайдеры (можно расширять) + RUSSIAN_HOSTERS = [ + "yandex", "mail.ru", "vk", "selectel", "timeweb", "beget", + "reg.ru", "nic.ru", "rtcdn", "astra", "sbercloud", + "cloud4russia", "гостех", "ростелеком", "мтс", "билайн", + "мегафон", "вымпелком", "ростелеком", "netology", + ] + + async def check(self, page: Page, url: str) -> CheckResult: + try: + # Получаем IP-адрес сервера через JavaScript + ip_info = await page.evaluate("""() => { + return new Promise((resolve) => { + // Пробуем определить IP через несколько методов + const methods = [ + () => fetch('https://api.ipify.org?format=json') + .then(r => r.json()).then(d => ({ip: d.ip, method: 'ipify'})), + () => fetch('https://api.my-ip.io/v2/json') + .then(r => r.json()).then(d => ({ip: d.ip, method: 'my-ip-io'})), + ]; + + // Пробуем каждый метод + (async () => { + for (const method of methods) { + try { + const result = await method(); + resolve(result); + return; + } catch (e) {} + } + resolve(null); + })(); + }); + }""") + + if not ip_info: + return self._skip_result( + message="Не удалось определить IP-адрес сервера.", + ) + + ip = ip_info.get("ip", "") + method = ip_info.get("method", "") + + # Определяем страну по IP (используем публичный API) + country = await self._get_country(ip) + + if not country: + return self._skip_result( + message=f"Не удалось определить страну для IP {ip}.", + ) + + # Проверяем, Россия ли это + if country.lower() == "ru" or "Россия" in country: + return self._pass_result( + message=f"Сервер находится в России (IP: {ip}, Страна: {country})", + details={ + "ip": ip, + "country": country, + "method": method, + }, + ) + + # Сервер за рубежом — критическое нарушение + return self._fail_result( + message=( + f"Сервер находится ЗА ПРЕДЕЛАМИ РОССИИ ({country}, IP: {ip}). " + f"Это нарушение ст. 18 152-ФЗ: базы данных граждан РФ должны быть в РФ." + ), + details={ + "ip": ip, + "country": country, + "method": method, + }, + ) + + except Exception as e: + return self._error_result(message=f"Ошибка при проверке геолокации: {str(e)}") + + async def _get_country(self, ip: str) -> str | None: + """Получает страну по IP через публичный API.""" + try: + import urllib.request + import json + + # Пробуем несколько бесплатных API + apis = [ + f"https://ipapi.co/{ip}/json/", + f"https://ip-api.com/json/{ip}?fields=country", + f"https://api.ipgeolocation.io/ipgeo?ip={ip}", + ] + + for api_url in apis: + try: + req = urllib.request.Request( + api_url, + headers={"User-Agent": "Mozilla/5.0"} + ) + with urllib.request.urlopen(req, timeout=5) as response: + data = json.loads(response.read().decode()) + + # ipapi.co формат + if "country_name" in data: + return data["country_name"] + # ip-api.com формат + if "country" in data: + return data["country"] + # ipgeolocation формат + if "country_name" in data: + return data["country_name"] + + except Exception: + continue + + return None + + except Exception: + return None diff --git a/scanner/engine.py b/scanner/engine.py new file mode 100644 index 0000000..374092f --- /dev/null +++ b/scanner/engine.py @@ -0,0 +1,127 @@ +""" +Движок сканирования 152-ФЗ. +Использует Playwright для загрузки сайтов и выполнения проверок. +""" +from __future__ import annotations + +import asyncio +import uuid +from datetime import datetime +from typing import Any + +from playwright.async_api import async_playwright, Page + +from scanner.checks import ALL_CHECKS +from scanner.models import CheckResult, CheckStatus, RiskLevel, SiteScan + + +class ScannerEngine: + """Управляет процессом сканирования сайтов.""" + + def __init__(self, timeout: int = 15000, headless: bool = True): + self.timeout = timeout + self.headless = headless + + async def scan_url(self, url: str, checks: list[str] | None = None) -> SiteScan: + """ + Просканировать один URL. + + Args: + url: URL для сканирования + checks: список имен проверок (None = все) + + Returns: + SiteScan с результатами + """ + scan_id = str(uuid.uuid4())[:8] + result = SiteScan(url=url, scanned_at=datetime.now()) + + async with async_playwright() as p: + browser = await p.chromium.launch( + headless=self.headless, + args=[ + "--no-sandbox", + "--disable-setuid-sandbox", + "--disable-dev-shm-usage", + "--disable-gpu", + ], + ) + + try: + context = await browser.new_context( + viewport={"width": 1920, "height": 1080}, + user_agent=( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Safari/537.36" + ), + ) + + page = await context.new_page() + + # Устанавливаем stealth (отключаем webdriver flag) + await page.add_init_script(""" + Object.defineProperty(navigator, 'webdriver', { + get: () => undefined + }); + """) + + # Загружаем страницу + try: + await page.goto(url, timeout=self.timeout, wait_until="networkidle") + except Exception as e: + result.checks.append( + CheckResult( + name="page_load", + title="Загрузка страницы", + status=CheckStatus.ERROR, + risk_level=RiskLevel.CRITICAL, + message=f"Не удалось загрузить страницу: {str(e)}", + ) + ) + return result + + # Подождем, пока JavaScript отрендерит динамический контент + await page.wait_for_timeout(3000) + + # Запускаем все проверки + check_names = checks if checks else list(ALL_CHECKS.keys()) + + tasks = [] + for check_name in check_names: + check_cls = ALL_CHECKS.get(check_name) + if check_cls: + check_instance = check_cls() + tasks.append(check_instance.check(page, url)) + + if tasks: + check_results = await asyncio.gather(*tasks, return_exceptions=True) + for cr in check_results: + if isinstance(cr, CheckResult): + result.checks.append(cr) + elif isinstance(cr, Exception): + result.checks.append( + CheckResult( + name="unknown", + title="Неизвестная ошибка", + status=CheckStatus.ERROR, + risk_level=RiskLevel.CRITICAL, + message=f"Необработанная ошибка: {str(cr)}", + ) + ) + + finally: + await browser.close() + + # Вычисляем итоговые метрики + result._calculate_score() + return result + + async def scan_urls_batch(self, urls: list[str], checks: list[str] | None = None) -> list[SiteScan]: + """Параллельно сканировать несколько URL.""" + tasks = [self.scan_url(url, checks) for url in urls] + return await asyncio.gather(*tasks) + + +# Глобальный экземпляр для переиспользования +scanner_engine = ScannerEngine() diff --git a/scanner/models.py b/scanner/models.py new file mode 100644 index 0000000..d0daed5 --- /dev/null +++ b/scanner/models.py @@ -0,0 +1,118 @@ +""" +Pydantic модели для 152-ФЗ Scanner. +Описывают структуры данных: результаты проверок, отчеты, сканирование. +""" +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import Any, Optional + +from pydantic import BaseModel, Field + + +class CheckStatus(str, Enum): + PASS = "pass" # ✅ Проверка пройдена + FAIL = "fail" # ❌ Проверка не пройдена + SKIP = "skip" # ⚠️ Проверка пропущена + ERROR = "error" # 🔴 Ошибка при проверке + + +class RiskLevel(str, Enum): + LOW = "low" # 🟢 Низкий риск + MEDIUM = "medium" # 🟡 Средний риск + HIGH = "high" # 🔴 Высокий риск + CRITICAL = "critical" # 🚫 Критический риск + + +class CheckResult(BaseModel): + """Результат одной проверки.""" + name: str # Имя проверки (например, "cookie_banner") + title: str # Название для отображения (например, "Cookie-баннер") + status: CheckStatus # Статус проверки + risk_level: RiskLevel = RiskLevel.LOW # Уровень риска при провале + message: str = "" # Описание результата + details: dict[str, Any] = {} # Детали для отладки/отчета + + +class SiteScan(BaseModel): + """Результат сканирования одного сайта.""" + url: str # URL сайта + scanned_at: datetime # Время сканирования + checks: list[CheckResult] = [] # Результаты всех проверок + score: int = 100 # Общий балл (0-100) + risks: list[CheckResult] = [] # Проваленные проверки (только fail) + critical_risks: list[CheckResult] = [] # Критические провалы + is_compliant: bool = True # Соответствует ли 152-ФЗ + summary: str = "" # Краткое резюме + + def _calculate_score(self): + """Вычисляет итоговый балл и статус соответствия.""" + risk_weights = { + RiskLevel.LOW: 5, + RiskLevel.MEDIUM: 10, + RiskLevel.HIGH: 20, + RiskLevel.CRITICAL: 30, + } + + score = 100 + risks = [] + critical_risks = [] + + for check in self.checks: + if check.status == CheckStatus.FAIL: + penalty = risk_weights.get(check.risk_level, 5) + score -= penalty + risks.append(check) + if check.risk_level == RiskLevel.CRITICAL: + critical_risks.append(check) + + self.score = max(0, score) + self.risks = risks + self.critical_risks = critical_risks + self.is_compliant = self.score >= 70 + + # Формируем краткое резюме + fail_count = sum(1 for c in self.checks if c.status == CheckStatus.FAIL) + error_count = sum(1 for c in self.checks if c.status == CheckStatus.ERROR) + skip_count = sum(1 for c in self.checks if c.status == CheckStatus.SKIP) + + if self.is_compliant: + emoji = "🟢" if self.score >= 90 else "🟡" + self.summary = f"{emoji} Сайт частично соответствует 152-ФЗ (балл: {self.score}/100). Найдено нарушений: {fail_count}." + else: + emoji = "🔴" if self.critical_risks else "🟠" + self.summary = f"{emoji} Сайт НЕ соответствует 152-ФЗ (балл: {self.score}/100). Критических нарушений: {len(self.critical_risks)}. Требуется доработка." + + return self + + +class ScanRequest(BaseModel): + """Запрос на сканирование (через API).""" + url: str = Field(..., description="URL сайта для сканирования") + checks: Optional[list[str]] = None # Какие проверки запустить (все если None) + + +class CSVScanRequest(BaseModel): + """Запрос на сканирование CSV-файла.""" + urls: list[str] = Field(..., description="Список URL для сканирования") + + +class ScanResponse(BaseModel): + """Ответ API — результат сканирования.""" + scan_id: str # Уникальный ID сканирования + status: str # pending / completed / error + url: Optional[str] = None # URL (если одно) + urls_count: Optional[int] = None # Количество URL (если CSV) + results: Optional[list[SiteScan]] = None # Результаты + error: Optional[str] = None # Ошибка (если есть) + created_at: datetime # Время создания запроса + + +class PDFReport(BaseModel): + """Данные для генерации PDF-отчета.""" + site_scan: SiteScan # Результаты сканирования + generated_at: datetime # Время генерации + company_name: Optional[str] = None # Название компании (если известно) + contact_email: Optional[str] = None # Email для связи + scanner_version: str = "1.0.0" # Версия сканера