#!/usr/bin/python # -*- coding: utf-8 -*- # --------------------------------------------------------------------------- # # # # Isida-NG Jabber Bot (fork of iSida) # # Original iSida Copyright (C) 2009-2015 diSabler # # Isida-NG Copyright (C) 2026 Luciferus # # Project home: https://git.ubunix.pro/Luciferus/Isida-NG # # # # RSS/Aton feeds plugin # # # # This program is free software: you can redistribute it and/or modify # # it under the terms of the GNU General Public License as published by # # the Free Software Foundation, either version 3 of the License, or # # (at your option) any later version. # # # # This program is distributed in the hope that it will be useful, # # but WITHOUT ANY WARRANTY; without even the implied warranty of # # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # # GNU General Public License for more details. # # # # You should have received a copy of the GNU General Public License # # along with this program. If not, see . # # # # --------------------------------------------------------------------------- # import re import time import hashlib import urllib.request import urllib.parse import urllib.error from urllib.parse import urlparse, urljoin import feedparser # --------------------------------------------------------------------------- # # HTTP HELPERS # # --------------------------------------------------------------------------- # def _fetch_url_raw(url): """Возвращает сырые байты или None при ошибке.""" try: req = urllib.request.Request(url, headers={ 'User-Agent': 'curl/8.5.0', 'Accept': '*/*', 'Accept-Language': 'ru-RU,ru;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive', }) with urllib.request.urlopen(req, timeout=20) as resp: data = resp.read() if resp.headers.get('Content-Encoding') == 'gzip': import gzip try: data = gzip.decompress(data) except Exception: pass return data except Exception as e: pprint('[RSS] fetch error for %s: %s' % (url, e), 'red') return None def _ensure_url(url): url = url.strip() if not url.startswith(('http://', 'https://')): return 'https://' + url return url def _url_base(url): p = urlparse(url) return '%s://%s' % (p.scheme or 'https', p.netloc) # --------------------------------------------------------------------------- # # FEED PARSING С ИСПОЛЬЗОВАНИЕМ feedparser # # --------------------------------------------------------------------------- # def _parse_feed_with_feedparser(url): """ Загружает и парсит фид с помощью feedparser. Возвращает dict с ключами: url, title, type, items или None при ошибке. """ try: feed = feedparser.parse(url) if feed.bozo: if not feed.entries and not feed.feed: return None if not feed.entries and not feed.feed.get('title'): return None feed_type = 'RSS' if hasattr(feed, 'version') and feed.version: if 'atom' in feed.version.lower(): feed_type = 'ATOM' title = feed.feed.get('title', '') or feed.feed.get('subtitle', '') or feed.feed.get('name', '') or 'RSS Feed' items = [] for entry in feed.entries: title_text = entry.get('title', 'Без заголовка') link = entry.get('link', '') if not link and entry.get('links'): for l in entry.links: if l.get('rel') == 'alternate' or not l.get('rel'): link = l.get('href', '') break description = entry.get('description', '') or entry.get('summary', '') or entry.get('content', [{}])[0].get('value', '') description = re.sub(r'<[^>]+>', ' ', description).strip() items.append({ 'title': title_text.strip(), 'link': link.strip(), 'description': description[:500] }) return { 'url': url, 'title': title.strip(), 'type': feed_type, 'items': items, } except Exception as e: pprint('[RSS] feedparser error for %s: %s' % (url, e), 'red') return None # --------------------------------------------------------------------------- # # ПОИСК ФИДОВ В HTML # # --------------------------------------------------------------------------- # def _find_feeds_in_html_bytes(raw_data, base_url): """Ищет ссылки на фиды в сырых байтах (без декодирования).""" if not raw_data: return [] try: content = raw_data.decode('utf-8', errors='ignore') except: content = raw_data.decode('latin-1', errors='ignore') feeds = [] seen = set() def add(feed_type, href, title): if not href: return if href.startswith('mailto:') or href.startswith('javascript:'): return if href.startswith('/'): href = _url_base(base_url) + href elif not href.startswith(('http://', 'https://')): href = urljoin(base_url, href) key = href.lower() if key in seen: return seen.add(key) title = title or ('RSS Feed' if feed_type == 'RSS' else 'Atom Feed') feeds.append({'type': feed_type, 'url': href, 'title': title}) # Ищем все href, содержащие feed, rss или atom for href in re.findall(r'href=[\"\']([^\"\']+)[\"\']', content, re.I): if re.search(r'(feed|rss|atom)', href, re.I): ftype = 'ATOM' if 'atom' in href.lower() else 'RSS' add(ftype, href, '') return feeds # --------------------------------------------------------------------------- # # РЕЗОЛВЕР ФИДА # # --------------------------------------------------------------------------- # def _resolve_feed(url): """Находит фид по URL (прямой фид или сайт с ссылками).""" url = _ensure_url(url) # 1) Пробуем напрямую через feedparser parsed = _parse_feed_with_feedparser(url) if parsed and parsed['items']: return parsed # 2) Пробуем найти фид на странице raw_data = _fetch_url_raw(url) if raw_data is None: return None feeds = _find_feeds_in_html_bytes(raw_data, url) for f in feeds: parsed = _parse_feed_with_feedparser(f['url']) if parsed and parsed['items']: return parsed return None def _get_feed_items(url, limit=10): resolved = _resolve_feed(url) if resolved is None: return None, None items = resolved['items'][:limit] if limit > 0 else resolved['items'] return resolved['title'], items # --------------------------------------------------------------------------- # # COMMANDS # # --------------------------------------------------------------------------- # def rss(type, jid, nick, text): parts = text.strip().split(' ', 1) subcmd = parts[0].lower() if parts else 'help' args = parts[1] if len(parts) > 1 else '' if subcmd == 'show': rss_show(type, jid, nick, args) elif subcmd == 'add': rss_add(type, jid, nick, args) elif subcmd == 'del': rss_del(type, jid, nick, args) elif subcmd == 'get': rss_get(type, jid, nick, args) elif subcmd == 'new': rss_new(type, jid, nick, args) elif subcmd == 'clear': rss_clear(type, jid, nick, args) elif subcmd == 'all': rss_all(type, jid, nick, args) else: rss_help(type, jid, nick, '') def rss_show(type, jid, nick, text): room = getRoom(jid) feeds = cur_execute_fetchall('SELECT url, update, type, time FROM feed WHERE room=%s', (room,)) if not feeds: feeds = cur_execute_fetchall('SELECT url, update, type, time FROM feed WHERE room=%s', ('global',)) if not feeds: send_msg(type, jid, nick, L('Расписание новостей в %s пусто' % room, '%s/%s' % (jid, nick))) return msg = L('Расписание новостей в %s:' % room, '%s/%s' % (jid, nick)) for feed in feeds: msg += '\n%s (%s) %s — %s' % (feed[0], feed[1], feed[2], disp_time(feed[3], '%s/%s' % (jid, nick))) send_msg(type, jid, nick, msg) def _parse_interval(s): s = s.strip().lower() n = '' i = 0 while i < len(s) and (s[i].isdigit()): n += s[i]; i += 1 if not n: return None unit = s[i:] n = int(n) if unit in ('m', 'min', 'minutes'): return n * 60 if unit in ('h', 'hour', 'hours'): return n * 3600 if unit in ('d', 'day', 'days'): return n * 86400 return n def rss_add(type, jid, nick, text): parts = text.strip().split(' ', 2) if len(parts) < 3: send_msg(type, jid, nick, L('Usage: rss add