neo
de18090304
- 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 — внешние скрипты аналитики
103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""
|
|
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-файл или БД."}
|