""" Опциональный LLM-анализ политики конфиденциальности. Запускается ПОСЛЕ основных проверок, только если явно запрошен. Проверяет 16 обязательных разделов политики по 152-ФЗ. """ from __future__ import annotations import re import json from pathlib import Path from typing import Any import aiohttp from bs4 import BeautifulSoup from scanner.checks.base import BaseCheck from scanner.models import CheckResult, CheckStatus, RiskLevel # 16 обязательных разделов политики конфиденциальности (Reova, 2026) REQUIRED_SECTIONS = [ {"key": "general", "name": "Общие положения", "article": "Ст. 3"}, {"key": "operator", "name": "Наименование оператора", "article": "Ст. 18.1"}, {"key": "purposes", "name": "Цели обработки", "article": "Ст. 5 ч.2"}, {"key": "categories", "name": "Категории ПД", "article": "Ст. 18.1"}, {"key": "subjects", "name": "Категории субъектов", "article": "Ст. 18.1"}, {"key": "basis", "name": "Правовые основания", "article": "Ст. 6"}, {"key": "methods", "name": "Способы обработки", "article": "Ст. 3"}, {"key": "storage", "name": "Сроки хранения", "article": "Ст. 5 ч.7"}, {"key": "destruction","name": "Порядок уничтожения", "article": "Ст. 21"}, {"key": "rights", "name": "Права субъекта", "article": "Ст. 14, 15"}, {"key": "security", "name": "Меры безопасности", "article": "Ст. 19"}, {"key": "transfer", "name": "Трансграничная передача", "article": "Ст. 12"}, {"key": "thirdparty", "name": "Передача третьим лицам", "article": "Ст. 6 ч.3"}, {"key": "cookies", "name": "Cookies", "article": "Разъяснения РКН"}, {"key": "dpo", "name": "Ответственный за ПД (DPO)", "article": "Ст. 22.1"}, {"key": "disputes", "name": "Порядок разрешения споров", "article": "Ст. 17"}, ] class LLMPolicyCheck(BaseCheck): """Глубокий LLM-анализ текста политики конфиденциальности.""" name = "llm_policy_analysis" title = "LLM-анализ политики конфиденциальности" description = "Проверяет 16 обязательных разделов политики через LLM" risk_on_fail = RiskLevel.HIGH default_weight = 20 def __init__(self, llm_url: str = "http://192.168.1.152:80/v1/chat/completions", llm_model: str = "Qwen3.6-35B-A3B-UD-IQ4_NL.gguf"): self.llm_url = llm_url self.llm_model = llm_model async def check(self, page, url: str) -> CheckResult: """Этот метод вызывается через API, не через Playwright.""" return self._skip_result(message="LLM-анализ запускается отдельно через analyze_policy()") async def analyze_policy(self, policy_url: str) -> CheckResult: """ Загрузить страницу политики и проанализировать через LLM. Args: policy_url: URL страницы политики конфиденциальности Returns: CheckResult с детальным анализом 16 разделов """ try: # 1. Скачиваем текст политики import aiohttp async with aiohttp.ClientSession() as session: async with session.get( policy_url, headers={"User-Agent": "Mozilla/5.0"}, timeout=aiohttp.ClientTimeout(total=10), ) as resp: if resp.status != 200: return self._error_result(f"Не удалось загрузить политику: HTTP {resp.status}") 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() if len(policy_text) < 100: return self._fail_result( message="Текст политики слишком короткий для анализа (<100 символов).", details={"char_count": len(policy_text)}, ) # 2. Отправляем на LLM-анализ sections_list = "\n".join( f"- {s['key']}: {s['name']} ({s['article']})" for s in REQUIRED_SECTIONS ) system_prompt = ( "Ты юридический аудитор. Проанализируй текст политики конфиденциальности " "на соответствие 152-ФЗ. Проверь наличие каждого из 16 обязательных разделов. " "Отвечай СТРОГО в формате JSON без markdown-обёрток." ) user_prompt = f"""Проверь, какие из этих разделов присутствуют в тексте политики: {sections_list} Для каждого раздела укажи: - present: true/false (присутствует ли) - evidence: короткая цитата из текста (до 100 символов), если present=true - quality: "full"/"partial"/"missing" Также дай: - overall_score: общая оценка от 0 до 100 - summary: краткое резюме на русском (2-3 предложения) Текст политики (первые 8000 символов): {policy_text[:8000]} Ответь ТОЛЬКО JSON, без markdown:""" analysis = await self._call_llm(system_prompt, user_prompt) if not analysis: return self._error_result("LLM не вернул результат анализа.") # 3. Формируем результат present_count = sum(1 for s in analysis.get("sections", []) if s.get("present")) missing_sections = [s["name"] for s in analysis.get("sections", []) if not s.get("present")] partial_sections = [s["name"] for s in analysis.get("sections", []) if s.get("quality") == "partial"] overall = analysis.get("overall_score", 0) summary = analysis.get("summary", "Анализ завершён.") if present_count >= 14 and not missing_sections: return self._pass_result( message=f"Политика содержит {present_count}/16 обязательных разделов. {summary}", details={ "present_count": present_count, "total_sections": 16, "overall_score": overall, "missing": missing_sections, "partial": partial_sections, "llm_summary": summary, }, ) if present_count >= 10: return self._fail_result( message=f"Политика неполная: {present_count}/16 разделов. Отсутствуют: {', '.join(missing_sections[:5])}. {summary}", risk_level=RiskLevel.HIGH, details={ "present_count": present_count, "total_sections": 16, "overall_score": overall, "missing": missing_sections, "partial": partial_sections, "llm_summary": summary, }, ) return self._fail_result( message=f"Политика критически неполная: {present_count}/16 разделов. Отсутствуют: {', '.join(missing_sections[:8])}. {summary}", risk_level=RiskLevel.CRITICAL, details={ "present_count": present_count, "total_sections": 16, "overall_score": overall, "missing": missing_sections, "partial": partial_sections, "llm_summary": summary, }, ) except Exception as e: return self._error_result(f"Ошибка LLM-анализа: {str(e)}") async def _call_llm(self, system_prompt: str, user_prompt: str) -> dict | None: """Вызвать LLM для анализа текста политики.""" try: import aiohttp payload = { "model": self.llm_model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], "temperature": 0.1, "max_tokens": 2048, } async with aiohttp.ClientSession() as session: async with session.post( self.llm_url, json=payload, timeout=aiohttp.ClientTimeout(total=60), ) as resp: if resp.status != 200: return None data = await resp.json() content = data.get("choices", [{}])[0].get("message", {}).get("content", "") # Извлекаем JSON из ответа (LLM может обернуть в ```json) content = content.strip() if content.startswith("```"): content = re.sub(r"^```(?:json)?\s*", "", content) content = re.sub(r"\s*```$", "", content) return json.loads(content) except Exception: return None async def run_llm_analysis(policy_url: str, llm_url: str | None = None) -> dict[str, Any]: """ Удобная функция для запуска LLM-анализа из CLI/API. Returns: Словарь с результатом check и деталями """ checker = LLMPolicyCheck() if llm_url: checker.llm_url = llm_url result = await checker.analyze_policy(policy_url) return { "status": result.status.value, "title": result.title, "message": result.message, "risk_level": result.risk_level.value, "details": result.details, }