152-fz-scanner/scanner/checks/operator_info.py

"""
Проверка контактных данных оператора: ИНН, ОГРН, DPO.
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 OperatorInfoCheck(BaseCheck):
    """Проверяет наличие реквизитов оператора и DPO на сайте."""

    name = "operator_info"
    title = "Контакты оператора и DPO"
    description = "Проверяет ИНН/ОГРН оператора и контакты ответственного за ПД"
    risk_on_fail = RiskLevel.HIGH
    default_weight = 15

    # Regex для российских реквизитов
    INN_PATTERN = r"ИНН\s*:?\s*(\d{10}|\d{12})"
    OGRN_PATTERN = r"ОГРН(ИП)?\s*:?\s*(\d{13}|\d{15})"
    DPO_PATTERNS = [
        r"ответственн[ыо][ейм]\s*за\s*(обработку\s*)?(персональных?\s*)?данных?",
        r"(data\s*protection\s*officer|dpo)",
        r"лиц[ое]\s*ответственн[ао][ея]\s*за\s*пд",
    ]
    CONTACT_PATTERN = r"([a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+)"

    async def check(self, page: Page, url: str) -> CheckResult:
        try:
            # 1. Главная страница
            page_text = await page.evaluate("() => document.body.innerText")

            # 2. Ищем ссылку на политику конфиденциальности
            policy_text = ""
            try:
                links = await page.query_selector_all("a[href]")
                policy_patterns = [
                    r"политик[аые]\s*конфиденциальн", r"privacy\s*policy",
                    r"персональные\s*данные", r"обработк[ае]\s*данных",
                ]
                policy_url = None
                for link in links:
                    href = await link.get_attribute("href") or ""
                    text = (await link.inner_text()).strip()
                    combined = (text + " " + href).lower()
                    if any(re.search(p, combined, re.UNICODE | re.IGNORECASE) for p in policy_patterns):
                        from urllib.parse import urljoin
                        policy_url = urljoin(url, href)
                        break

                if policy_url:
                    import aiohttp
                    from bs4 import BeautifulSoup
                    async with aiohttp.ClientSession() as session:
                        async with session.get(
                            policy_url,
                            headers={"User-Agent": "Mozilla/5.0"},
                            timeout=aiohttp.ClientTimeout(total=8),
                        ) as resp:
                            if resp.status == 200:
                                html = await resp.text()
                                soup = BeautifulSoup(html, "html.parser")
                                for tag in soup(["script", "style", "nav", "footer", "header"]):
                                    tag.decompose()
                                policy_text = soup.get_text(separator=" ").strip()
            except Exception:
                policy_text = ""  # Ошибка загрузки политики — работаем с главной

            # 3. Объединяем текст главной + политики
            combined_text = page_text + "\n" + policy_text

            # 4. Ищем реквизиты
            inn_match = re.search(self.INN_PATTERN, combined_text, re.IGNORECASE)
            inn = inn_match.group(1) if inn_match else None

            ogrn_match = re.search(self.OGRN_PATTERN, combined_text, re.IGNORECASE)
            ogrn = ogrn_match.group(2) if ogrn_match else None

            dpo_found = any(
                re.search(pattern, combined_text, re.IGNORECASE)
                for pattern in self.DPO_PATTERNS
            )

            contacts = re.findall(self.CONTACT_PATTERN, combined_text)
            contacts = [c for c in contacts if not any(
                skip in c.lower() for skip in ("example", "test", "noreply", "no-reply")
            )][:3]

            # 5. Результат
            has_inn = inn is not None
            has_ogrn = ogrn is not None
            has_dpo = dpo_found
            has_contact = len(contacts) > 0
            present = sum([has_inn, has_ogrn, has_dpo, has_contact])

            details = {
                "inn": inn, "ogrn": ogrn,
                "dpo_found": dpo_found, "contacts": contacts,
                "policy_checked": bool(policy_text),
            }

            if present >= 3:
                return self._pass_result(
                    message=f"Реквизиты оператора: ИНН={'есть' if has_inn else 'нет'}, "
                            f"ОГРН={'есть' if has_ogrn else 'нет'}, "
                            f"DPO={'есть' if has_dpo else 'нет'}",
                    details=details,
                )

            if present >= 1:
                missing = []
                if not has_inn: missing.append("ИНН")
                if not has_ogrn: missing.append("ОГРН")
                if not has_dpo: missing.append("контакты DPO")
                return self._fail_result(
                    message=f"Реквизиты неполные: отсутствуют {', '.join(missing)}",
                    risk_level=RiskLevel.HIGH,
                    details=details,
                )

            return self._fail_result(
                message="Реквизиты оператора не найдены. Отсутствуют ИНН, ОГРН, DPO и контакты.",
                risk_level=RiskLevel.CRITICAL,
                details=details,
            )

        except Exception as e:
            return self._error_result(f"Ошибка проверки реквизитов: {str(e)}")