core/parser.py - master - dava/MagDipl

core/parser.py
453 строки20 KB
david
beta_version
23 фев 2026, 11:20
23 фев 2026, 11:20777dfb5
import requests
import logging
from bs4 import BeautifulSoup
import re
import xml.etree.ElementTree as ET

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LentaParser:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
            "Connection": "keep-alive",
        })

    def parse_rubric(self, rubric_slug: str, limit: int = 1000):
        """
        Пытаемся получить статьи Lenta.ru.
        1) (если доступно) JSON API
        2) RSS
        3) HTML (страница рубрики) — самый надёжный fallback
        """
        # rubric_slug может быть как slug ("politics"), так и URL ("https://lenta.ru/rubrics/politics/")
        slug = (rubric_slug or "").strip()
        if slug.startswith("http://") or slug.startswith("https://"):
            # ожидаем .../rubrics/<slug>/
            parts = [p for p in slug.split("/") if p]
            if "rubrics" in parts:
                idx = parts.index("rubrics")
                if idx + 1 < len(parts):
                    slug = parts[idx + 1]
            else:
                # fallback: берём последний сегмент
                slug = parts[-1] if parts else slug

        # Пробуем разные варианты URL JSON API (исторически менялись; часто сейчас 404)
        urls_to_try = [
            f"https://lenta.ru/api/v1/rubric/{slug}/?limit={limit}",
            f"https://lenta.ru/api/v1/rubrics/{slug}/?limit={limit}",
            f"https://lenta.ru/api/v1/news/{slug}/?limit={limit}",
        ]
        
        for url in urls_to_try:
            try:
                logger.info(f"Пробую URL: {url}")
                response = self.session.get(url, timeout=20)
                logger.info(f"Status code: {response.status_code}")
                
                if response.status_code != 200:
                    logger.warning(f"Ошибка {response.status_code} для {url}")
                    continue

                try:
                    data = response.json()
                except Exception as e:
                    logger.error(f"Ошибка парсинга JSON: {e}")
                    continue

                # В разные моменты структура ответа может отличаться: ищем список статей в нескольких местах
                raw_articles = None
                if isinstance(data, dict):
                    # Пробуем разные ключи
                    for key in ["articles", "items", "news", "results", "data"]:
                        if key in data:
                            val = data[key]
                            if isinstance(val, list):
                                raw_articles = val
                                logger.info(f"Найдены статьи в ключе '{key}': {len(val)} шт.")
                                break
                            elif isinstance(val, dict) and "articles" in val:
                                raw_articles = val["articles"]
                                logger.info(f"Найдены статьи в data.articles: {len(raw_articles)} шт.")
                                break
                elif isinstance(data, list):
                    raw_articles = data
                    logger.info(f"Ответ - список из {len(data)} элементов")
                
                if not isinstance(raw_articles, list) or len(raw_articles) == 0:
                    logger.warning(f"Не найдены статьи в ответе. Ключи: {list(data.keys()) if isinstance(data, dict) else 'N/A'}")
                    continue

                articles = []
                for item in raw_articles:
                    if not isinstance(item, dict):
                        continue
                    
                    # Пробуем разные поля для текста
                    text = (
                        item.get("description")
                        or item.get("announce")
                        or item.get("text")
                        or item.get("body")
                        or item.get("lead")
                        or item.get("summary")
                        or ""
                    )
                    
                    # Если нет description, пробуем взять title + text
                    if not text and item.get("title"):
                        text = item.get("title", "")
                    
                    if not text or len(text.strip()) < 10:
                        continue
                    
                    title = (
                        item.get("title")
                        or item.get("headline")
                        or item.get("name")
                        or "No title"
                    )
                    
                    articles.append({
                        "title": title,
                        "text": text.strip(),
                        "category": slug
                    })
                
                logger.info(f"Успешно собрано {len(articles)} статей для категории {slug}")
                return articles
                
            except requests.exceptions.RequestException as e:
                logger.error(f"Ошибка запроса к {url}: {e}")
                continue
            except Exception as e:
                logger.error(f"Неожиданная ошибка для {url}: {e}")
                continue
        
        # Если JSON API не сработал, пробуем RSS
        logger.info(f"Пробую RSS для категории {slug}")
        articles = self._parse_rss(slug, limit)
        if articles:
            return articles

        # Если RSS не сработал, парсим HTML страницы рубрики
        logger.info(f"Пробую HTML для категории {slug}")
        articles = self._parse_html_rubric(slug, limit)
        if articles:
            return articles
        
        logger.warning(f"Не удалось получить статьи для категории {slug}")
        return []
    
    def _parse_rss(self, slug: str, limit: int = 100):
        """Альтернативный метод через RSS фиды"""
        rss_urls = [
            f"https://lenta.ru/rss/news/{slug}",
            f"https://lenta.ru/rss/rubrics/{slug}",
            f"https://lenta.ru/rss/top7",
            # иногда RSS бывает с .xml или со слешем
            f"https://lenta.ru/rss/news/{slug}.xml",
            f"https://lenta.ru/rss/rubrics/{slug}.xml",
            f"https://lenta.ru/rss/top7.xml",
        ]
        
        articles = []
        for rss_url in rss_urls:
            try:
                logger.info(f"Пробую RSS: {rss_url}")
                response = self.session.get(rss_url, timeout=10)
                if response.status_code != 200:
                    continue
                
                # Парсим RSS XML
                root = ET.fromstring(response.content)
                
                # Находим все item элементы
                items = root.findall('.//item') or root.findall('.//{http://www.w3.org/2005/Atom}entry')
                
                if not items:
                    continue
                
                for item in items[:limit]:
                    # Извлекаем title
                    title_elem = item.find('title') or item.find('{http://www.w3.org/2005/Atom}title')
                    title = title_elem.text if title_elem is not None else "No title"
                    
                    # Извлекаем описание
                    desc_elem = item.find('description') or item.find('{http://www.w3.org/2005/Atom}summary') or item.find('content')
                    text = desc_elem.text if desc_elem is not None else ""
                    
                    # Извлекаем ссылку
                    link_elem = item.find('link') or item.find('{http://www.w3.org/2005/Atom}link')
                    link = link_elem.text if link_elem is not None else (link_elem.get('href', '') if link_elem is not None else '')
                    
                    # Если описание короткое, пробуем получить полный текст
                    if not text or len(text) < 50:
                        if link:
                            full_text = self._fetch_article_text(link)
                            if full_text:
                                text = full_text
                    
                    if not text or len(text.strip()) < 20:
                        continue
                    
                    articles.append({
                        "title": title.strip(),
                        "text": text.strip(),
                        "category": slug
                    })
                
                if articles:
                    logger.info(f"Через RSS собрано {len(articles)} статей")
                    return articles
                    
            except ET.ParseError as e:
                logger.warning(f"Ошибка парсинга XML RSS {rss_url}: {e}")
                continue
            except Exception as e:
                logger.error(f"Ошибка при парсинге RSS {rss_url}: {e}")
                continue
        
        return []

    def _parse_html_rubric(self, slug: str, limit: int = 100):
        """
        Парсинг HTML страницы рубрики:
        1) берём ссылку на рубрику
        2) вытаскиваем ссылки на новости
        3) по каждой ссылке вытаскиваем текст статьи
        """
        # Пробуем разные варианты URL
        rubric_urls = [
            f"https://lenta.ru/rubrics/{slug}/",
            f"https://lenta.ru/rubric/{slug}/",
            f"https://lenta.ru/{slug}/",
        ]
        
        rubric_url = None
        r = None
        for url in rubric_urls:
            try:
                logger.info(f"HTML: пробую {url}")
                r = self.session.get(url, timeout=20)
                if r.status_code == 200:
                    rubric_url = url
                    break
            except Exception as e:
                logger.debug(f"HTML: ошибка запроса {url}: {e}")
                continue
        
        if not r or r.status_code != 200:
            logger.warning(f"HTML: все варианты URL для {slug} вернули ошибку, пробую главную страницу")
            return self._parse_main_page(slug, limit)

        soup = BeautifulSoup(r.text, "html.parser")

        hrefs = []
        # Ищем ссылки на новости разными способами
        for a in soup.find_all("a", href=True):
            href = a.get("href", "").strip()
            if not href:
                continue
            
            # Полные URL
            if href.startswith("https://lenta.ru/news/"):
                hrefs.append(href)
            # Относительные ссылки на новости вида /news/YYYY/MM/DD/slug/
            elif re.match(r"^/news/\d{4}/\d{2}/\d{2}/", href):
                hrefs.append("https://lenta.ru" + href)
            # Любые /news/ ссылки
            elif href.startswith("/news/") and len(href) > 10:
                hrefs.append("https://lenta.ru" + href)
        
        # Также ищем в специальных контейнерах статей
        article_containers = soup.find_all(["article", "div"], class_=re.compile(r"card|item|news|article", re.I))
        for container in article_containers:
            link = container.find("a", href=True)
            if link:
                href = link.get("href", "").strip()
                if href.startswith("/news/") or href.startswith("https://lenta.ru/news/"):
                    if href.startswith("/"):
                        href = "https://lenta.ru" + href
                    if href not in hrefs:
                        hrefs.append(href)

        # дедуп + ограничение
        seen = set()
        links = []
        for u in hrefs:
            if u in seen:
                continue
            seen.add(u)
            links.append(u)
            if len(links) >= max(5, limit * 2):
                break

        logger.info(f"HTML: найдено {len(links)} уникальных ссылок на новости")

        if not links:
            logger.warning(f"HTML: не нашёл ссылок на новости на странице {rubric_url}")
            logger.debug(f"HTML: найдено всего {len(hrefs)} ссылок, но ни одна не подошла")
            # Пробуем парсить главную страницу как fallback
            logger.info("HTML: пробую парсить главную страницу Lenta.ru")
            return self._parse_main_page(slug, limit)

        articles = []
        for i, url in enumerate(links):
            if len(articles) >= limit:
                break
            logger.debug(f"Обрабатываю статью {i+1}/{min(len(links), limit)}: {url[:80]}...")
            text = self._fetch_article_text(url)
            if not text:
                logger.debug(f"Не удалось получить текст с {url}")
                continue

            title = self._fetch_article_title(url)
            articles.append({
                "title": title or "No title",
                "text": text,
                "category": slug,
            })
            logger.debug(f"Добавлена статья: {title[:50]}... ({len(text)} символов)")

        logger.info(f"HTML: собрано {len(articles)} статей из рубрики {slug}")
        return articles

    def _fetch_article_title(self, url: str) -> str:
        """Пытается вытащить заголовок статьи."""
        try:
            response = self.session.get(url, timeout=10)
            if response.status_code != 200:
                return ""
            soup = BeautifulSoup(response.text, "html.parser")
            h1 = soup.find("h1")
            if h1 and h1.get_text(strip=True):
                return h1.get_text(strip=True)
            og = soup.find("meta", attrs={"property": "og:title"})
            if og and og.get("content"):
                return og["content"].strip()
            return ""
        except Exception:
            return ""
    
    def _fetch_article_text(self, url: str) -> str:
        """Получает текст статьи со страницы"""
        try:
            response = self.session.get(url, timeout=10)
            if response.status_code != 200:
                return ""
            
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # Ищем основной контент статьи - пробуем разные селекторы
            content_selectors = [
                'article',
                '.b-topic__content',
                '.article-body',
                '[itemprop="articleBody"]',
                '.text',
                '.b-text',
                '.topic-body',
                'div[itemprop="articleBody"]',
                '.js-topic__text',
            ]
            
            for selector in content_selectors:
                content = soup.select_one(selector)
                if content:
                    # Удаляем скрипты и стили
                    for script in content(["script", "style", "noscript"]):
                        script.decompose()
                    text = content.get_text(separator=' ', strip=True)
                    if len(text) > 100:
                        return text
            
            # Fallback: ищем все параграфы в article или main
            main_content = soup.find('main') or soup.find('article')
            if main_content:
                paragraphs = main_content.find_all('p')
                if paragraphs:
                    text = ' '.join([p.get_text(strip=True) for p in paragraphs if p.get_text(strip=True)])
                    if len(text) > 100:
                        return text
            
            return ""
        except Exception as e:
            logger.debug(f"Ошибка при получении текста с {url}: {e}")
            return ""
    
    def parse_all_news(self, limit: int = 200) -> list:
        """
        Парсит ВСЕ новости с главной страницы Lenta.ru без категорий.
        Категория будет определена классификатором позже.
        """
        try:
            logger.info(f"Парсинг всех новостей с Lenta.ru (лимит: {limit})")
            response = self.session.get("https://lenta.ru/", timeout=15)
            if response.status_code != 200:
                logger.warning(f"Главная страница вернула {response.status_code}")
                return []
            
            soup = BeautifulSoup(response.text, "html.parser")
            
            # Ищем все ссылки на новости
            news_links = []
            for a in soup.find_all("a", href=True):
                href = a.get("href", "").strip()
                # Берём только ссылки на новости
                if href.startswith("/news/") and len(href) > 10:
                    full_url = "https://lenta.ru" + href
                    if full_url not in news_links:
                        news_links.append(full_url)
            
            logger.info(f"Найдено {len(news_links)} ссылок на новости на главной странице")
            
            if not news_links:
                logger.warning("Не найдено ссылок на новости")
                return []
            
            articles = []
            total_to_check = min(len(news_links), limit * 2)  # Проверяем больше, т.к. не все подойдут
            
            for i, url in enumerate(news_links[:total_to_check]):
                if len(articles) >= limit:
                    break
                
                if (i + 1) % 10 == 0:
                    logger.info(f"Обработано {i+1}/{total_to_check} статей, собрано {len(articles)}")
                
                text = self._fetch_article_text(url)
                if not text or len(text) < 50:
                    continue
                
                title = self._fetch_article_title(url)
                if not title or title == "No title":
                    # Пробуем извлечь из URL или текста
                    title = url.split("/")[-1].replace("-", " ").title()
                
                articles.append({
                    "title": title,
                    "text": text,
                    "category": "unknown",  # Будет определена классификатором
                })
            
            logger.info(f"✅ Собрано {len(articles)} статей (без категорий, будут классифицированы)")
            return articles
        except Exception as e:
            logger.error(f"Ошибка при парсинге всех новостей: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return []
    
    def _parse_main_page(self, category_slug: str, limit: int = 100) -> list:
        """Парсит главную страницу Lenta.ru и фильтрует по категориям (legacy метод)."""
        # Используем новый метод, но помечаем категорией
        articles = self.parse_all_news(limit)
        for article in articles:
            article["category"] = category_slug
        return articles