mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-04 03:52:34 +00:00
agessaman
5c5bd0969c
The twelve feed methods (subscription CRUD, activity, errors, statistics, preview and item formatting) and the two module helpers they use move from app.py into modules/web_viewer/feeds.py; BotDataViewer inherits the mixin, so every method stays on the viewer for routes and tests. Method bodies are unchanged apart from dropping function-local 'import sqlite3' lines the module import now covers, and _validate_feed_interval's annotation (object to Any, for mypy, which now checks this module). app.py keeps every name it imported for them so modules.web_viewer.app.<name> still resolves. The two tests that patched create_safe_requests_session and safe_requests_request to steer _preview_feed_items now patch them where the method lives; the SafeUrlPolicy.validate patches target the shared class and are unchanged. app.py drops to about 8,200 lines.
568 lines
22 KiB
Python
568 lines
22 KiB
Python
"""Feed subscription storage, preview and formatting for the web viewer.
|
|
|
|
Mixed into BotDataViewer, which supplies ``config``, ``logger``,
|
|
``_get_db_connection`` and ``_db_connection``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from typing import Any
|
|
|
|
from modules.feed_filter_eval import item_passes_filter_config
|
|
from modules.feed_format import format_feed_message, sort_feed_items
|
|
from modules.feed_manager import _useful_feed_content_type
|
|
from modules.feed_parse import (
|
|
api_item_fields,
|
|
feed_allow_private_urls,
|
|
feed_max_parsed_items,
|
|
feed_max_response_bytes,
|
|
rss_entry_published,
|
|
)
|
|
from modules.security_utils import SafeUrlPolicy, create_safe_requests_session, safe_requests_request
|
|
|
|
|
|
def _validate_feed_interval(raw: Any) -> int:
|
|
"""Coerce a feed poll interval, rejecting values that break the poller.
|
|
|
|
feed_manager compares ``current_time - last_check >= interval``: anything
|
|
<= 0 leaves every feed permanently due and hammers the source URL, and a
|
|
None (JSON ``null``) raises a TypeError that aborts the whole poll cycle for
|
|
every feed, not just this one.
|
|
"""
|
|
try:
|
|
interval = int(raw)
|
|
except (TypeError, ValueError):
|
|
raise ValueError("check_interval_seconds must be a positive integer")
|
|
if interval <= 0:
|
|
raise ValueError("check_interval_seconds must be a positive integer")
|
|
return interval
|
|
|
|
|
|
def _read_limited_requests_response(
|
|
response: Any,
|
|
*,
|
|
max_bytes: int,
|
|
feed_type: str,
|
|
) -> bytes:
|
|
"""Read a streamed, decompressed requests response under a hard byte cap."""
|
|
response.raise_for_status()
|
|
content_type = response.headers.get('Content-Type', '')
|
|
if not _useful_feed_content_type(content_type, feed_type):
|
|
raise ValueError(f"Unexpected {feed_type.upper()} content type: {content_type}")
|
|
|
|
declared_length = response.headers.get('Content-Length')
|
|
if declared_length:
|
|
try:
|
|
content_length = int(declared_length)
|
|
except ValueError:
|
|
content_length = None
|
|
if content_length is not None and content_length > max_bytes:
|
|
raise ValueError(f"Feed response exceeds {max_bytes} byte limit")
|
|
|
|
chunks: list[bytes] = []
|
|
total = 0
|
|
for chunk in response.iter_content(chunk_size=64 * 1024, decode_unicode=False):
|
|
if not chunk:
|
|
continue
|
|
total += len(chunk)
|
|
if total > max_bytes:
|
|
raise ValueError(f"Feed response exceeds {max_bytes} byte limit")
|
|
chunks.append(chunk)
|
|
return b''.join(chunks)
|
|
|
|
|
|
class FeedSubscriptionsMixin:
|
|
"""Feed subscription CRUD, activity/error queries, preview and item formatting."""
|
|
|
|
config: Any
|
|
logger: Any
|
|
|
|
def _get_feed_subscriptions(self, channel_filter=None):
|
|
"""Get all feed subscriptions, optionally filtered by channel"""
|
|
try:
|
|
with self._db_connection() as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.cursor()
|
|
|
|
if channel_filter:
|
|
cursor.execute('''
|
|
SELECT * FROM feed_subscriptions
|
|
WHERE channel_name = ?
|
|
ORDER BY id
|
|
''', (channel_filter,))
|
|
else:
|
|
cursor.execute('''
|
|
SELECT * FROM feed_subscriptions
|
|
ORDER BY id
|
|
''')
|
|
|
|
rows = cursor.fetchall()
|
|
feeds = []
|
|
for row in rows:
|
|
feed = dict(row)
|
|
# Get feed count for this channel
|
|
cursor.execute('''
|
|
SELECT COUNT(*) FROM feed_activity
|
|
WHERE feed_id = ?
|
|
''', (feed['id'],))
|
|
feed['item_count'] = cursor.fetchone()[0]
|
|
|
|
# Get error count
|
|
cursor.execute('''
|
|
SELECT COUNT(*) FROM feed_errors
|
|
WHERE feed_id = ? AND resolved_at IS NULL
|
|
''', (feed['id'],))
|
|
feed['error_count'] = cursor.fetchone()[0]
|
|
|
|
feeds.append(feed)
|
|
|
|
return {'feeds': feeds, 'total': len(feeds)}
|
|
except Exception as e:
|
|
self.logger.error(f"Error getting feed subscriptions: {e}")
|
|
return {'feeds': [], 'total': 0, 'error': str(e)}
|
|
|
|
def _get_feed_subscription(self, feed_id):
|
|
"""Get a single feed subscription by ID"""
|
|
try:
|
|
with self._db_connection() as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.cursor()
|
|
cursor.execute('SELECT * FROM feed_subscriptions WHERE id = ?', (feed_id,))
|
|
row = cursor.fetchone()
|
|
return dict(row) if row else None
|
|
except Exception as e:
|
|
self.logger.error(f"Error getting feed subscription: {e}")
|
|
return None
|
|
|
|
def _create_feed_subscription(self, data):
|
|
"""Create a new feed subscription"""
|
|
import json
|
|
conn = None
|
|
try:
|
|
feed_type = data.get('feed_type')
|
|
feed_url = data.get('feed_url')
|
|
channel_name = data.get('channel_name')
|
|
feed_name = data.get('feed_name')
|
|
check_interval = _validate_feed_interval(data.get('check_interval_seconds', 300))
|
|
api_config = data.get('api_config')
|
|
output_format = data.get('output_format')
|
|
message_send_interval = data.get('message_send_interval_seconds')
|
|
filter_config = data.get('filter_config')
|
|
sort_config = data.get('sort_config')
|
|
|
|
if not all([feed_type, feed_url, channel_name]):
|
|
raise ValueError("feed_type, feed_url, and channel_name are required")
|
|
|
|
conn = self._get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
api_config_str = json.dumps(api_config) if api_config else None
|
|
filter_config_str = json.dumps(filter_config) if filter_config else None
|
|
sort_config_str = json.dumps(sort_config) if sort_config else None
|
|
|
|
cursor.execute('''
|
|
INSERT INTO feed_subscriptions
|
|
(feed_type, feed_url, channel_name, feed_name, check_interval_seconds, api_config, output_format, message_send_interval_seconds, filter_config, sort_config)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
''', (feed_type, feed_url, channel_name, feed_name, check_interval, api_config_str, output_format, message_send_interval, filter_config_str, sort_config_str))
|
|
|
|
conn.commit()
|
|
return cursor.lastrowid
|
|
except Exception:
|
|
if conn:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _update_feed_subscription(self, feed_id, data):
|
|
"""Update a feed subscription"""
|
|
import json
|
|
conn = None
|
|
try:
|
|
conn = self._get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
updates = []
|
|
params = []
|
|
|
|
if 'channel_name' in data:
|
|
channel_name = str(data['channel_name']).strip() if data['channel_name'] is not None else ''
|
|
if not channel_name:
|
|
raise ValueError("channel_name cannot be empty")
|
|
updates.append('channel_name = ?')
|
|
params.append(channel_name)
|
|
|
|
if 'feed_name' in data:
|
|
updates.append('feed_name = ?')
|
|
params.append(data['feed_name'])
|
|
|
|
if 'check_interval_seconds' in data:
|
|
updates.append('check_interval_seconds = ?')
|
|
params.append(_validate_feed_interval(data['check_interval_seconds']))
|
|
|
|
if 'enabled' in data:
|
|
updates.append('enabled = ?')
|
|
params.append(1 if data['enabled'] else 0)
|
|
|
|
if 'api_config' in data:
|
|
updates.append('api_config = ?')
|
|
params.append(json.dumps(data['api_config']) if data['api_config'] else None)
|
|
|
|
if 'output_format' in data:
|
|
updates.append('output_format = ?')
|
|
params.append(data['output_format'] if data['output_format'] else None)
|
|
|
|
if 'message_send_interval_seconds' in data:
|
|
updates.append('message_send_interval_seconds = ?')
|
|
params.append(float(data['message_send_interval_seconds']) if data['message_send_interval_seconds'] else None)
|
|
|
|
if 'filter_config' in data:
|
|
updates.append('filter_config = ?')
|
|
params.append(json.dumps(data['filter_config']) if data['filter_config'] else None)
|
|
|
|
if 'sort_config' in data:
|
|
updates.append('sort_config = ?')
|
|
params.append(json.dumps(data['sort_config']) if data['sort_config'] else None)
|
|
|
|
if not updates:
|
|
return True # Nothing to update
|
|
|
|
updates.append('updated_at = CURRENT_TIMESTAMP')
|
|
params.append(feed_id)
|
|
|
|
query = f'UPDATE feed_subscriptions SET {", ".join(updates)} WHERE id = ?'
|
|
cursor.execute(query, params)
|
|
conn.commit()
|
|
|
|
return cursor.rowcount > 0
|
|
except Exception:
|
|
if conn:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _delete_feed_subscription(self, feed_id):
|
|
"""Delete a feed subscription"""
|
|
conn = None
|
|
try:
|
|
conn = self._get_db_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute('DELETE FROM feed_subscriptions WHERE id = ?', (feed_id,))
|
|
conn.commit()
|
|
return cursor.rowcount > 0
|
|
except Exception:
|
|
if conn:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _reset_feed_errors(self, feed_id=None):
|
|
"""Clear recorded feed errors. Pass a feed_id to clear one feed, or None for all.
|
|
|
|
Returns the number of error rows deleted.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = self._get_db_connection()
|
|
cursor = conn.cursor()
|
|
if feed_id is None:
|
|
cursor.execute('DELETE FROM feed_errors')
|
|
else:
|
|
cursor.execute('DELETE FROM feed_errors WHERE feed_id = ?', (feed_id,))
|
|
conn.commit()
|
|
return cursor.rowcount
|
|
except Exception:
|
|
if conn:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def _get_feed_activity(self, feed_id, limit=50):
|
|
"""Get activity log for a feed"""
|
|
try:
|
|
with self._db_connection() as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT * FROM feed_activity
|
|
WHERE feed_id = ?
|
|
ORDER BY processed_at DESC
|
|
LIMIT ?
|
|
''', (feed_id, limit))
|
|
rows = cursor.fetchall()
|
|
return [dict(row) for row in rows]
|
|
except Exception as e:
|
|
self.logger.error(f"Error getting feed activity: {e}")
|
|
return []
|
|
|
|
def _get_feed_errors(self, feed_id, limit=20):
|
|
"""Get error history for a feed"""
|
|
try:
|
|
with self._db_connection() as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
SELECT * FROM feed_errors
|
|
WHERE feed_id = ?
|
|
ORDER BY occurred_at DESC
|
|
LIMIT ?
|
|
''', (feed_id, limit))
|
|
rows = cursor.fetchall()
|
|
return [dict(row) for row in rows]
|
|
except Exception as e:
|
|
self.logger.error(f"Error getting feed errors: {e}")
|
|
return []
|
|
|
|
def _get_feed_statistics(self):
|
|
"""Get aggregate feed statistics"""
|
|
try:
|
|
with self._db_connection() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
stats = {}
|
|
|
|
# Total subscriptions
|
|
cursor.execute('SELECT COUNT(*) FROM feed_subscriptions')
|
|
stats['total_subscriptions'] = cursor.fetchone()[0]
|
|
|
|
# Enabled subscriptions
|
|
cursor.execute('SELECT COUNT(*) FROM feed_subscriptions WHERE enabled = 1')
|
|
stats['enabled_subscriptions'] = cursor.fetchone()[0]
|
|
|
|
# Items processed in last 24h
|
|
cursor.execute('''
|
|
SELECT COUNT(*) FROM feed_activity
|
|
WHERE processed_at > datetime('now', '-24 hours')
|
|
''')
|
|
stats['items_24h'] = cursor.fetchone()[0]
|
|
|
|
# Items processed in last 7d
|
|
cursor.execute('''
|
|
SELECT COUNT(*) FROM feed_activity
|
|
WHERE processed_at > datetime('now', '-7 days')
|
|
''')
|
|
stats['items_7d'] = cursor.fetchone()[0]
|
|
|
|
# Error count
|
|
cursor.execute('''
|
|
SELECT COUNT(*) FROM feed_errors
|
|
WHERE resolved_at IS NULL
|
|
''')
|
|
stats['active_errors'] = cursor.fetchone()[0]
|
|
|
|
# Most active channels
|
|
cursor.execute('''
|
|
SELECT channel_name, COUNT(*) as feed_count
|
|
FROM feed_subscriptions
|
|
WHERE enabled = 1
|
|
GROUP BY channel_name
|
|
ORDER BY feed_count DESC
|
|
LIMIT 10
|
|
''')
|
|
stats['top_channels'] = [{'channel': row[0], 'count': row[1]} for row in cursor.fetchall()]
|
|
|
|
return stats
|
|
except Exception as e:
|
|
self.logger.error(f"Error getting feed statistics: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _get_feeds_by_channel(self, channel_idx):
|
|
"""Get all feeds for a specific channel index"""
|
|
# First get channel name from index
|
|
# This would require channel_manager access
|
|
# For now, return empty list
|
|
return []
|
|
|
|
def _preview_feed_items(self, feed_url: str, feed_type: str, output_format: str, api_config: dict[str, Any] | None = None, filter_config: dict[str, Any] | None = None, sort_config: dict[str, Any] | None = None) -> list[dict[str, Any]]:
|
|
"""Preview feed items with custom output format (standalone, doesn't require bot)"""
|
|
|
|
import feedparser
|
|
|
|
try:
|
|
items = []
|
|
|
|
# Validate URL for SSRF protection
|
|
url_policy = SafeUrlPolicy(allow_private=feed_allow_private_urls(self.config))
|
|
if not url_policy.validate(feed_url):
|
|
raise ValueError("Invalid or unsafe feed URL")
|
|
max_response_bytes = feed_max_response_bytes(self.config)
|
|
max_parsed_items = feed_max_parsed_items(self.config)
|
|
preview_parse_limit = min(20, max_parsed_items)
|
|
|
|
if feed_type == 'rss':
|
|
# Fetch RSS feed
|
|
with create_safe_requests_session(url_policy) as http_session:
|
|
response = safe_requests_request(
|
|
http_session,
|
|
'GET',
|
|
feed_url,
|
|
policy=url_policy,
|
|
timeout=30,
|
|
headers={'User-Agent': 'MeshCoreBot/1.0 FeedManager'},
|
|
stream=True,
|
|
)
|
|
with closing(response):
|
|
content = _read_limited_requests_response(
|
|
response,
|
|
max_bytes=max_response_bytes,
|
|
feed_type='rss',
|
|
)
|
|
parsed = feedparser.parse(content)
|
|
|
|
# Get items (we'll filter and limit later)
|
|
for entry in parsed.entries[:preview_parse_limit]:
|
|
items.append({
|
|
'title': entry.get('title', 'Untitled'),
|
|
'description': entry.get('description', ''),
|
|
'link': entry.get('link', ''),
|
|
'published': rss_entry_published(entry),
|
|
})
|
|
|
|
elif feed_type == 'api':
|
|
# Fetch API feed
|
|
if api_config is None:
|
|
raise ValueError("api_config is required for API feed type")
|
|
method = api_config.get('method', 'GET').upper()
|
|
headers = api_config.get('headers', {})
|
|
params = api_config.get('params', {})
|
|
body = api_config.get('body')
|
|
parser_config = api_config.get('response_parser', {})
|
|
|
|
with create_safe_requests_session(url_policy) as http_session:
|
|
response = safe_requests_request(
|
|
http_session,
|
|
method,
|
|
feed_url,
|
|
policy=url_policy,
|
|
headers=headers,
|
|
params=params,
|
|
json=body if method == 'POST' else None,
|
|
timeout=30,
|
|
stream=True,
|
|
)
|
|
with closing(response):
|
|
content = _read_limited_requests_response(
|
|
response,
|
|
max_bytes=max_response_bytes,
|
|
feed_type='api',
|
|
)
|
|
|
|
# Try to parse JSON, handle cases where response might be a string
|
|
try:
|
|
data = json.loads(content)
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
# If JSON parsing fails, try to get text and see if it's an error message
|
|
text_snippet = content[:200].decode('utf-8', errors='replace')
|
|
raise Exception(f"API returned non-JSON response: {text_snippet[:200]}")
|
|
|
|
# Check if response is an error message (string)
|
|
if isinstance(data, str):
|
|
raise Exception(f"API returned error message: {data[:200]}")
|
|
|
|
# Ensure data is a dict or list
|
|
if not isinstance(data, (dict, list)):
|
|
raise Exception(f"API response is not a valid JSON object or array: {type(data).__name__} - {str(data)[:200]}")
|
|
|
|
# Extract items using parser config
|
|
items_path = parser_config.get('items_path', '')
|
|
items_data: dict[Any, Any] | list[Any]
|
|
if items_path:
|
|
parts = items_path.split('.')
|
|
items_data = data
|
|
for part in parts:
|
|
if isinstance(items_data, dict):
|
|
items_data = items_data.get(part, [])
|
|
else:
|
|
raise Exception(f"Cannot navigate path '{items_path}': expected dict at '{part}', got {type(items_data).__name__}")
|
|
else:
|
|
# If no items_path, data should be a list or we wrap it
|
|
if isinstance(data, list):
|
|
items_data = data
|
|
elif isinstance(data, dict):
|
|
# If it's a dict, try to find common array fields
|
|
_found = data.get('items') or data.get('data') or data.get('results')
|
|
items_data = _found if _found is not None else [data]
|
|
else:
|
|
items_data = [data]
|
|
|
|
# Ensure items_data is a list
|
|
if not isinstance(items_data, list):
|
|
items_data = [items_data]
|
|
|
|
for item_data in items_data[:preview_parse_limit]:
|
|
# Ensure item_data is a dict
|
|
if not isinstance(item_data, dict):
|
|
# If it's not a dict, try to convert or skip
|
|
if isinstance(item_data, str):
|
|
# If it's a string, create a simple dict
|
|
item_data = {'title': item_data, 'description': item_data}
|
|
else:
|
|
# Try to convert to dict or skip
|
|
continue
|
|
|
|
items.append(api_item_fields(item_data, parser_config))
|
|
|
|
# Apply sorting if configured
|
|
if sort_config:
|
|
items = sort_feed_items(items, sort_config, log_warning=self.logger.warning)
|
|
|
|
# Apply filter if configured
|
|
if filter_config:
|
|
items = [item for item in items if item_passes_filter_config(item, filter_config)]
|
|
|
|
# Limit to first 3 items after filtering
|
|
items = items[:3]
|
|
|
|
# Format items using output format (shared with FeedManager)
|
|
formatted_items = []
|
|
for item in items:
|
|
formatted = self._format_feed_item(item, output_format, feed_name='')
|
|
formatted_items.append({
|
|
'original': item,
|
|
'formatted': formatted
|
|
})
|
|
|
|
return formatted_items
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error previewing feed: {e}")
|
|
raise
|
|
|
|
def _format_feed_item(self, item: dict[str, Any], format_str: str, feed_name: str = '') -> str:
|
|
"""Format a feed item using the shared feed formatter (parity with FeedManager)."""
|
|
try:
|
|
max_length = self.config.getint(
|
|
'Feed_Manager', 'max_message_length', fallback=130
|
|
)
|
|
except Exception:
|
|
max_length = 130
|
|
try:
|
|
shorten_feed_urls = (
|
|
self.config.getboolean('Feed_Manager', 'shorten_urls', fallback=False)
|
|
if self.config.has_section('Feed_Manager')
|
|
else False
|
|
)
|
|
except ValueError:
|
|
shorten_feed_urls = False
|
|
|
|
return format_feed_message(
|
|
item,
|
|
format_str,
|
|
feed_name=feed_name or '',
|
|
max_message_length=max_length,
|
|
shorten_feed_urls=shorten_feed_urls,
|
|
config=self.config,
|
|
logger=self.logger,
|
|
)
|