Upload scraper.py

scraper.py

@@ -0,0 +1,218 @@
+"""
+Скрапер для сайта ФИПИ (fipi.ru)
+Извлекает задания по русскому языку для ЕГЭ (задание 27)
+"""
+
+import httpx
+from bs4 import BeautifulSoup
+from typing import List, Dict, Optional
+from datetime import datetime
+import re
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class FIPIScraper:
+    """Парсер для сайта ФИПИ"""
+    
+    def __init__(self, base_url: str = "https://fipi.ru"):
+        self.base_url = base_url
+        self.headers = {
+            "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",
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
+            "Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
+        }
+    
+    async def fetch_page(self, url: str) -> Optional[str]:
+        """Получение HTML страницы"""
+        # Создаем клиент с отключенной проверкой SSL (для fipi.ru поддоменов)
+        import ssl
+        ssl_context = ssl.create_default_context()
+        ssl_context.check_hostname = False
+        ssl_context.verify_mode = ssl.CERT_NONE
+        
+        async with httpx.AsyncClient(
+            headers=self.headers, 
+            timeout=30.0,
+            verify=ssl_context
+        ) as client:
+            try:
+                response = await client.get(url)
+                response.raise_for_status()
+                return response.text
+            except httpx.HTTPError as e:
+                logger.error(f"Ошибка при получении {url}: {e}")
+                return None
+    
+    def parse_task_page(self, html: str, url: str) -> Optional[Dict]:
+        """Парсинг страницы с заданием"""
+        soup = BeautifulSoup(html, 'lxml')
+
+        # Извлечение заголовка - приоритет h1 в .content
+        title_tag = soup.select_one('.content h1') or soup.find('h1')
+        title = title_tag.get_text(strip=True) if title_tag else "Без названия"
+        
+        # Если заголовок пустой, пробуем извлечь из title документа
+        if not title or title == "Без названия":
+            title_doc = soup.find('title')
+            if title_doc:
+                title = title_doc.get_text(strip=True)
+
+        # Извлечение основного контента - приоритет .content
+        content_div = soup.select_one('.content') or soup.find('div', class_='field--name-body')
+        if not content_div:
+            content_div = soup.find('main') or soup.find('body')
+
+        # Очистка текста - удаляем скрипты и стили
+        for element in content_div.find_all(['script', 'style', 'nav', 'header', 'footer']):
+            element.decompose()
+        
+        content = content_div.get_text(separator='\n', strip=True) if content_div else ""
+
+        # Извлечение изображения (если есть)
+        images = []
+        for img in content_div.find_all('img'):
+            src = img.get('src') or img.get('data-src')
+            if src:
+                if not src.startswith('http'):
+                    src = self.base_url + src
+                images.append(src)
+
+        # Извлечение ссылок на задания
+        task_links = []
+        for link in content_div.find_all('a', href=True):
+            href = link['href']
+            link_text = link.get_text(strip=True)
+            if any(pattern in href for pattern in ['/ege/', '/oge/', '/task/', '/demo/', '/bank/']):
+                if not href.startswith('http'):
+                    href = self.base_url + href
+                task_links.append({"text": link_text, "url": href})
+
+        # Определение типа задания
+        task_type = self._detect_task_type(title, content)
+
+        # Извлечение вариантов (если есть)
+        variants = self._extract_variants(content)
+
+        return {
+            "title": title,
+            "content": content,
+            "source_url": url,
+            "task_type": task_type,
+            "images": images,
+            "variants": variants,
+            "task_links": task_links,
+            "scraped_at": datetime.utcnow().isoformat(),
+        }
+    
+    def _detect_task_type(self, title: str, content: str) -> str:
+        """Определение типа задания"""
+        text = (title + " " + content).lower()
+        
+        if any(word in text for word in ["сочинение", "эссе", "напишит"]):
+            return "writing"
+        elif any(word in text for word in ["тест", "выбер", "вариант"]):
+            return "test"
+        elif any(word in text for word in ["ауди", "слуш"]):
+            return "listening"
+        elif any(word in text for word in ["чит", "текст"]):
+            return "reading"
+        else:
+            return "other"
+    
+    def _extract_variants(self, content: str) -> List[str]:
+        """Извлечение вариантов ответов"""
+        variants = []
+        
+        # Паттерн для вариантов типа "1) ... 2) ..."
+        pattern = r'(\d+)[\.\)]\s*([^\n\d]+)'
+        matches = re.findall(pattern, content)
+        
+        for _, variant in matches:
+            variants.append(variant.strip())
+        
+        return variants[:10]  # Ограничение на 10 вариантов
+    
+    async def scrape_tasks(self, subject: str = "russian") -> List[Dict]:
+        """
+        Скрапинг заданий по предмету
+
+        Args:
+            subject: Код предмета (по умолчанию russian)
+
+        Returns:
+            Список заданий
+        """
+        tasks = []
+
+        # Актуальные URLs для скрапинга (fipi.ru) - только работающие
+        urls_to_scrape = [
+            f"{self.base_url}/ege/otkrytyy-bank-zadaniy-ege",
+            f"{self.base_url}/oge/otkrytyy-bank-zadaniy-oge",
+        ]
+
+        for url in urls_to_scrape:
+            logger.info(f"Скрапинг {url}")
+            html = await self.fetch_page(url)
+
+            if html:
+                task = self.parse_task_page(html, url)
+                if task:
+                    tasks.append(task)
+                    
+                    # Если есть ссылки на задания, скачиваем их
+                    for link_info in task.get('task_links', [])[:5]:  # Ограничиваем количество
+                        link_url = link_info.get('url')
+                        if link_url:
+                            logger.info(f"  -> Скачиваем задание: {link_url}")
+                            link_html = await self.fetch_page(link_url)
+                            if link_html:
+                                subtask = self.parse_task_page(link_html, link_url)
+                                if subtask:
+                                    tasks.append(subtask)
+
+        logger.info(f"Найдено {len(tasks)} заданий")
+        return tasks
+    
+    async def scrape_task_by_id(self, task_id: str) -> Optional[Dict]:
+        """Скрапинг конкретного задания по ID"""
+        url = f"{self.base_url}/task/{task_id}"
+        logger.info(f"Скрапинг задания {task_id}")
+        
+        html = await self.fetch_page(url)
+        if html:
+            return self.parse_task_page(html, url)
+        
+        return None
+    
+    async def search_tasks(self, query: str) -> List[Dict]:
+        """Поиск заданий по ключевому слову"""
+        tasks = []
+        # Используем правильный URL для поиска на fipi.ru
+        search_url = f"{self.base_url}/search?q={query}"
+
+        html = await self.fetch_page(search_url)
+        if not html:
+            # Пробуем альтернативный поиск через банк заданий
+            logger.info("Поиск не доступен, пробуем парсинг банка заданий")
+            return await self.scrape_tasks()
+
+        soup = BeautifulSoup(html, 'lxml')
+
+        # Поиск ссылок на задания с правильными паттернами
+        for link in soup.find_all('a', href=True):
+            href = link['href']
+            # Проверяем на наличие валидных URL заданий
+            if any(pattern in href for pattern in ['/ege/', '/oge/', '/task/', '/demo/', '/bank/']):
+                if not href.startswith('http'):
+                    href = self.base_url + href
+
+                task_html = await self.fetch_page(href)
+                if task_html:
+                    task = self.parse_task_page(task_html, href)
+                    if task:
+                        tasks.append(task)
+
+        return tasks