import sys import os import pandas as pd from datetime import datetime, timedelta from bs4 import BeautifulSoup import requests import re import time import feedparser import gradio as gr # Конфигурация RSS_URL = 'https://vecherka.su/rss/' # Обратите внимание: вероятно, опечатка — должно быть 'https://' CSV_FILE_PATH = 'bd.csv' def parse_article_date(published_date_str): """Парсит дату публикации статьи из RSS в объект datetime.""" date_formats = [ '%a, %d %b %Y %H:%M:%S %z', '%a, %d %b %Y %H:%M:%S %Z', '%d %b %Y' # запасной вариант ] for fmt in date_formats: try: return datetime.strptime(published_date_str, fmt) except ValueError: continue print(f"Could not parse date: '{published_date_str}'") return None def is_recent_article(parsed_date): """Проверяет, опубликована ли статья сегодня или вчера.""" today = datetime.now() yesterday = today - timedelta(days=1) article_date_str = parsed_date.strftime('%d-%m-%Y') today_str = today.strftime('%d-%m-%Y') yesterday_str = yesterday.strftime('%d-%m-%Y') return article_date_str == today_str or article_date_str == yesterday_str def extract_images_from_entry(entry): """Извлекает URL изображений из RSS‑записи (media_content, links, HTML‑контент).""" image_urls = [] # Из media_content if 'media_content' in entry and len(image_urls) < 3: for media in entry.media_content: if (media.get('type', '').startswith('image/') and media.get('url') and media.get('url') not in image_urls): image_urls.append(media['url']) if len(image_urls) == 3: break # Из links с rel='enclosure' if 'links' in entry and len(image_urls) < 3: for link_entry in entry.links: if (link_entry.get('rel') == 'enclosure' and link_entry.get('type', '').startswith('image/') and link_entry.get('href') and link_entry.get('href') not in image_urls): image_urls.append(link_entry['href']) if len(image_urls) == 3: break # Из HTML‑контента (summary или content) html_content = entry.get('summary', '') or ( entry.get('content', [{}])[0].get('value', '') if entry.get('content') else '') if html_content and len(image_urls) < 3: soup = BeautifulSoup(html_content, 'html.parser') img_tags = soup.find_all('img') for img in img_tags: if img.get('src') and img.get('src') not in image_urls: image_urls.append(img['src']) if len(image_urls) == 3: break return image_urls def fetch_article_text(news_link): """Загружает и извлекает текст статьи по ссылке.""" try: response = requests.get(news_link, timeout=10) response.raise_for_status() article_soup = BeautifulSoup(response.text, 'html.parser') detail_text_div = article_soup.find('div', class_='detail-text') if detail_text_div: full_text = detail_text_div.get_text(separator=' ', strip=True) # Удаляем предложения с «подписывайтесь» full_text = re.sub(r'[^.!?]*\bподписывайтесь\b[^.!?]*[?.!]', '', full_text, flags=re.IGNORECASE) full_text = re.sub(r'\s+', ' ', full_text).strip() # Пропускаем статьи с «Реклама» if re.search(r'\bРеклама\b', full_text, re.IGNORECASE): print(f"Skipping article due to 'Реклама' in full text.") return None return full_text else: print(f"Could not find 'detail-text' div for article: {news_link}") return None except requests.exceptions.RequestException as e: print(f"Error fetching content for {news_link}: {e}") return None except Exception as e: print(f"Error parsing content for {news_link}: {e}") return None def check_for_new_articles(): """Основная функция: проверяет RSS‑ленту на новые статьи и сохраняет их в CSV.""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Checking for new articles...") # Определяем диапазон дат (сегодня и вчера) today_date = datetime.now() yesterday_date = today_date - timedelta(days=1) today_str = today_date.strftime('%d-%m-%Y') yesterday_str = yesterday_date.strftime('%d-%m-%Y') return today_str, yesterday_str def load_existing_articles(): """Загружает существующие статьи из CSV для предотвращения дубликатов.""" processed_links = set() existing_df = None if os.path.exists(CSV_FILE_PATH): try: existing_df = pd.read_csv(CSV_FILE_PATH, encoding='utf-8-sig', sep=';') processed_links = set(existing_df['link'].tolist()) print(f"Loaded {len(processed_links)} existing articles from {CSV_FILE_PATH}.") except Exception as e: print(f"Error loading existing CSV: {e}. Starting with an empty processed_links set.") return processed_links, existing_df def fetch_and_parse_rss(): """Получает и парсит RSS‑ленту.""" feed = feedparser.parse(RSS_URL) if not feed.entries: print("No entries found in the RSS feed.") return [] print(f"Found {len(feed.entries)} entries in RSS feed.") return feed.entries def process_single_article(entry, today_str, yesterday_str, processed_links): """Обрабатывает одну статью: проверяет дату, извлекает данные, возвращает словарь с данными или None.""" title = getattr(entry, 'title', 'No Title') news_link = getattr(entry, 'link', None) # Пропускаем, если нет ссылки или статья уже обработана if not news_link or news_link in processed_links: return None published_date_str = getattr(entry, 'published', None) if not published_date_str: print(f"Skipping entry '{title}' due to missing publication date.") return None # Парсим дату публикации parsed_date = parse_article_date(published_date_str) if not parsed_date: return None # Пропускаем статью, если дату не удалось распарсить article_date_str = parsed_date.strftime('%d-%m-%Y') # Фильтруем статьи, опубликованные сегодня или вчера if article_date_str != today_str and article_date_str != yesterday_str: return None # Извлекаем URL изображений image_urls = extract_images_from_entry(entry) # Получаем полный текст статьи full_text = fetch_article_text(news_link) if not full_text: return None # Создаём короткий текст (первые 200 символов) short_text = full_text[:200] if len(full_text) > 200 else full_text return { 'title': title, 'published': article_date_str, 'image_urls': image_urls, 'link': news_link, 'full_text': full_text, 'Status': 'Off', 'short_text': short_text, 'Constant': '' } def save_new_articles(new_articles_data, existing_df): """Сохраняет новые статьи в CSV‑файл.""" if not new_articles_data: print("No new articles found to add.") return 0 new_df = pd.DataFrame(new_articles_data) # Преобразуем список URL изображений в строку через запятую new_df['image_urls'] = new_df['image_urls'].apply(lambda x: ', '.join(x)) if existing_df is not None and not existing_df.empty: # Дописываем в существующий файл без заголовка new_df.to_csv( CSV_FILE_PATH, mode='a', header=False, index=False, encoding='utf-8-sig', sep=';' ) else: # Создаём новый файл с заголовком new_df.to_csv( CSV_FILE_PATH, mode='w', header=True, index=False, encoding='utf-8-sig', sep=';' ) articles_added_count = len(new_articles_data) print(f"Added {articles_added_count} new articles to {CSV_FILE_PATH}.") return articles_added_count def check_for_new_articles(): """Основная функция: проверяет RSS‑ленту на новые статьи и сохраняет их в CSV. Возвращает количество добавленных статей.""" # Шаг 1: инициализация today_str, yesterday_str = check_for_new_articles_init() # Шаг 2: загрузка существующих статей processed_links, existing_df = load_existing_articles() # Шаг 3: получение RSS entries = fetch_and_parse_rss() if not entries: return 0 # Шаг 4: обработка каждой статьи new_articles_data = [] for entry in entries: article_data = process_single_article(entry, today_str, yesterday_str, processed_links) if article_data: new_articles_data.append(article_data) # Шаг 5: сохранение результатов return save_new_articles(new_articles_data, existing_df)