diff --git a/docs/FEEDS.md b/docs/FEEDS.md index 4b7c12f..be1dd17 100644 --- a/docs/FEEDS.md +++ b/docs/FEEDS.md @@ -264,6 +264,42 @@ Filter configuration determines which items are sent to channels. - `not_matches` - Regex does not match - `contains` - String contains value - `not_contains` - String does not contain value +- `within_days` - Item timestamp is within the last **N** calendar days (rolling window from **now**, UTC) +- `within_weeks` - Same as `within_days` with **N** × 7 days (e.g. four weeks ≈ `within_weeks` `4` or `within_days` `28`) + +`within_days` / `within_weeks` require a `field` pointing at a date (same paths as sort: `published` for RSS, or `raw.SomeTimeField` for APIs). They also require `days` or `weeks` respectively. + +If the date is missing or cannot be parsed, the condition **fails** (item is excluded). Set `"include_if_missing": true` on that condition to treat missing/unparseable dates as a pass instead. + +**Examples:** + +```json +{ + "conditions": [ + { + "field": "published", + "operator": "within_days", + "days": 28 + } + ], + "logic": "AND" +} +``` + +```json +{ + "conditions": [ + { + "field": "raw.LastUpdatedTime", + "operator": "within_weeks", + "weeks": 4 + } + ], + "logic": "AND" +} +``` + +Date parsing for filters matches sorting: ISO strings, Microsoft `/Date(...)/`, Unix timestamps (seconds or milliseconds), and common string formats. ### Logic @@ -277,6 +313,8 @@ For API feeds, use `raw.field` or `raw.nested.field` to access API response fiel - `raw.EventStatus` - `raw.StartRoadwayLocation.RoadName` +For RSS feeds, use `published` for the item publication time in `within_days` / `within_weeks` conditions. + ## Sort Configuration Sort items before processing: diff --git a/modules/feed_filter_eval.py b/modules/feed_filter_eval.py new file mode 100644 index 0000000..762da12 --- /dev/null +++ b/modules/feed_filter_eval.py @@ -0,0 +1,274 @@ +""" +Shared feed filter_config evaluation (RSS/API items). + +Used by FeedManager and the web viewer preview so filter behavior stays consistent. +""" + +from __future__ import annotations + +import json +import re +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Dict, Optional, Union + + +def get_nested_value(data: Any, path: str, default: Any = "") -> Any: + """Nested dict/list access using dot notation (same rules as FeedManager._get_nested_value).""" + if not path or not data: + return default + + parts = path.split(".") + value = data + + for part in parts: + if isinstance(value, dict): + value = value.get(part) + elif isinstance(value, list): + try: + idx = int(part) + if 0 <= idx < len(value): + value = value[idx] + else: + return default + except (ValueError, TypeError): + return default + else: + return default + + if value is None: + return default + + return value if value is not None else default + + +def parse_microsoft_date(date_str: str) -> Optional[datetime]: + """Parse Microsoft JSON date format: /Date(timestamp-offset)/""" + if not date_str or not isinstance(date_str, str): + return None + + match = re.match(r"/Date\((\d+)([+-]\d+)?\)/", date_str) + if match: + timestamp_ms = int(match.group(1)) + offset_str = match.group(2) if match.group(2) else "+0000" + timestamp = timestamp_ms / 1000.0 + + try: + offset_hours = int(offset_str[:3]) + offset_mins = int(offset_str[3:5]) + offset_seconds = (offset_hours * 3600) + (offset_mins * 60) + if offset_str[0] == "-": + offset_seconds = -offset_seconds + + tz = timezone.utc + if offset_seconds != 0: + from datetime import timedelta as td + + tz = timezone(td(seconds=offset_seconds)) + + return datetime.fromtimestamp(timestamp, tz=tz) + except (ValueError, IndexError): + return datetime.fromtimestamp(timestamp, tz=timezone.utc) + + return None + + +def _normalize_to_utc(dt: datetime) -> datetime: + if dt.tzinfo is not None: + return dt.astimezone(timezone.utc) + return dt.replace(tzinfo=timezone.utc) + + +def parse_item_field_as_datetime(item: Dict[str, Any], field_path: str) -> Optional[datetime]: + """ + Resolve field_path on item and parse to datetime (aligned with FeedManager._sort_items get_sort_value). + """ + raw_data = item.get("raw", {}) + value: Any = get_nested_value(raw_data, field_path, "") + if not value and field_path.startswith("raw."): + value = get_nested_value(raw_data, field_path[4:], "") + if not value: + value = get_nested_value(item, field_path, "") + + if value is None or value == "": + return None + + if isinstance(value, str) and value.startswith("/Date("): + dt = parse_microsoft_date(value) + if dt: + return dt + + if isinstance(value, datetime): + return _normalize_to_utc(value) + + if isinstance(value, (int, float)): + num = float(value) + # Heuristic: milliseconds since epoch (e.g. WSDOT-style APIs) + if num > 1e12: + num = num / 1000.0 + try: + return datetime.fromtimestamp(num, tz=timezone.utc) + except (OSError, OverflowError, ValueError): + return None + + if isinstance(value, str): + try: + dt = datetime.fromisoformat(value.replace("Z", "+00:00")) + return _normalize_to_utc(dt) + except ValueError: + pass + for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): + try: + dt = datetime.strptime(value, fmt) + return dt.replace(tzinfo=timezone.utc) + except ValueError: + continue + + return None + + +def _get_field_value_for_string_ops(item: Dict[str, Any], field_path: str) -> Any: + raw_data = item.get("raw", {}) + field_value = get_nested_value(raw_data, field_path, "") + if not field_value and field_path.startswith("raw."): + field_value = get_nested_value(raw_data, field_path[4:], "") + if not field_value: + field_value = get_nested_value(item, field_path, "") + return field_value + + +def evaluate_filter_condition( + item: Dict[str, Any], + condition: Dict[str, Any], + *, + log_warning: Optional[Callable[[str], None]] = None, +) -> Optional[bool]: + """ + Evaluate a single filter condition. Returns None if the condition should be skipped + (invalid / missing field path for operators that require it). + """ + field_path = condition.get("field") + operator = condition.get("operator", "equals") + + if operator in ("within_days", "within_weeks"): + if not field_path: + return None + include_if_missing = bool(condition.get("include_if_missing", False)) + dt = parse_item_field_as_datetime(item, field_path) + if dt is None: + return True if include_if_missing else False + + if operator == "within_days": + if "days" not in condition: + return False + try: + days = float(condition["days"]) + except (TypeError, ValueError): + return False + else: + if "weeks" not in condition: + return False + try: + weeks = float(condition["weeks"]) + except (TypeError, ValueError): + return False + days = weeks * 7.0 + + if days < 0: + days = 0.0 + + now = datetime.now(timezone.utc) + cutoff = now - timedelta(days=days) + item_dt = _normalize_to_utc(dt) + return item_dt >= cutoff + + if not field_path: + return None + + field_value = _get_field_value_for_string_ops(item, field_path) + field_value_str = str(field_value).lower() if field_value is not None else "" + + if operator == "equals": + compare_value = str(condition.get("value", "")).lower() + return field_value_str == compare_value + if operator == "not_equals": + compare_value = str(condition.get("value", "")).lower() + return field_value_str != compare_value + if operator == "in": + values = [str(v).lower() for v in condition.get("values", [])] + return field_value_str in values + if operator == "not_in": + values = [str(v).lower() for v in condition.get("values", [])] + return field_value_str not in values + if operator == "matches": + pattern = condition.get("pattern", "") + if pattern: + try: + return bool(re.search(pattern, str(field_value), re.IGNORECASE)) + except re.error: + return False + return False + if operator == "not_matches": + pattern = condition.get("pattern", "") + if pattern: + try: + return not bool(re.search(pattern, str(field_value), re.IGNORECASE)) + except re.error: + return True + return True + if operator == "contains": + compare_value = str(condition.get("value", "")).lower() + return compare_value in field_value_str + if operator == "not_contains": + compare_value = str(condition.get("value", "")).lower() + return compare_value not in field_value_str + + if log_warning: + log_warning(f"Unknown filter operator: {operator}") + return True + + +def item_passes_filter_config( + item: Dict[str, Any], + filter_config: Union[str, Dict[str, Any], None], + *, + log_warning: Optional[Callable[[str], None]] = None, +) -> bool: + """ + Return True if the item passes all filter conditions (AND) or any (OR), matching FeedManager behavior. + + Invalid JSON or empty conditions => pass through (True). + """ + if not filter_config: + return True + + try: + cfg = json.loads(filter_config) if isinstance(filter_config, str) else filter_config + except (json.JSONDecodeError, TypeError): + if log_warning: + log_warning("Invalid filter_config JSON, sending all items") + return True + + if not isinstance(cfg, dict): + return True + + conditions = cfg.get("conditions", []) + if not conditions: + return True + + logic = str(cfg.get("logic", "AND")).upper() + + results: list[bool] = [] + for condition in conditions: + if not isinstance(condition, dict): + continue + out = evaluate_filter_condition(item, condition, log_warning=log_warning) + if out is None: + continue + results.append(out) + + if not results: + return True + + if logic == "OR": + return any(results) + return all(results) diff --git a/modules/feed_manager.py b/modules/feed_manager.py index 6d2dab4..47b5009 100644 --- a/modules/feed_manager.py +++ b/modules/feed_manager.py @@ -21,6 +21,8 @@ from urllib.parse import urlparse import aiohttp import feedparser +from modules.feed_filter_eval import item_passes_filter_config + class FeedManager: """Manages RSS and API feed subscriptions""" @@ -971,118 +973,18 @@ class FeedManager: self._record_feed_error(feed['id'], 'queue', str(e)) def _should_send_item(self, feed: dict[str, Any], item: dict[str, Any]) -> bool: - """Check if an item should be sent based on filter configuration + """Check if an item should be sent based on filter configuration. - Filter config format: - { - "conditions": [ - {"field": "Priority", "operator": "in", "values": ["highest", "high"]}, - {"field": "EventStatus", "operator": "equals", "value": "open"}, - {"field": "EventCategory", "operator": "not_equals", "value": "Maintenance"}, - {"field": "raw.Priority", "operator": "matches", "pattern": "^(highest|high)$"} - ], - "logic": "AND" # or "OR" - } - - Supported operators: - - equals: exact match - - not_equals: not exact match - - in: value is in list - - not_in: value is not in list - - matches: regex match - - not_matches: regex doesn't match - - contains: substring match - - not_contains: substring doesn't match + See modules/feed_filter_eval.py and docs/FEEDS.md for operators. """ - filter_config_str = feed.get('filter_config') - if not filter_config_str: - # No filter configured, send all items - return True + def _warn(msg: str) -> None: + self.logger.warning(f"{msg} (feed id {feed.get('id')})") - try: - filter_config = json.loads(filter_config_str) if isinstance(filter_config_str, str) else filter_config_str - except (json.JSONDecodeError, TypeError): - self.logger.warning(f"Invalid filter_config for feed {feed['id']}, sending all items") - return True - - conditions = filter_config.get('conditions', []) - if not conditions: - # Empty conditions, send all items - return True - - logic = filter_config.get('logic', 'AND').upper() - - # Get raw data for field access - raw_data = item.get('raw', {}) - - # Evaluate each condition - results = [] - for condition in conditions: - field_path = condition.get('field') - operator = condition.get('operator', 'equals') - - if not field_path: - # Invalid condition, skip it - continue - - # Get field value using nested access - field_value = self._get_nested_value(raw_data, field_path, '') - if not field_value and field_path.startswith('raw.'): - # Try without 'raw.' prefix - field_value = self._get_nested_value(raw_data, field_path[4:], '') - - # If still not found, try top-level item fields - if not field_value: - field_value = self._get_nested_value(item, field_path, '') - - # Convert to string for comparison - field_value_str = str(field_value).lower() if field_value is not None else '' - - # Evaluate condition - result = False - if operator == 'equals': - compare_value = str(condition.get('value', '')).lower() - result = field_value_str == compare_value - elif operator == 'not_equals': - compare_value = str(condition.get('value', '')).lower() - result = field_value_str != compare_value - elif operator == 'in': - values = [str(v).lower() for v in condition.get('values', [])] - result = field_value_str in values - elif operator == 'not_in': - values = [str(v).lower() for v in condition.get('values', [])] - result = field_value_str not in values - elif operator == 'matches': - pattern = condition.get('pattern', '') - if pattern: - try: - result = bool(re.search(pattern, str(field_value), re.IGNORECASE)) - except re.error: - result = False - elif operator == 'not_matches': - pattern = condition.get('pattern', '') - if pattern: - try: - result = not bool(re.search(pattern, str(field_value), re.IGNORECASE)) - except re.error: - result = True - elif operator == 'contains': - compare_value = str(condition.get('value', '')).lower() - result = compare_value in field_value_str - elif operator == 'not_contains': - compare_value = str(condition.get('value', '')).lower() - result = compare_value not in field_value_str - else: - self.logger.warning(f"Unknown filter operator: {operator}") - result = True # Default to allowing if operator is unknown - - results.append(result) - - # Apply logic (AND or OR) - if logic == 'OR': - return any(results) - else: # AND (default) - return all(results) + return item_passes_filter_config( + item, + feed.get('filter_config'), + log_warning=_warn, + ) async def _send_feed_item(self, feed: dict[str, Any], item: dict[str, Any]): """Queue a feed item message instead of sending immediately""" diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index 06e11d9..11a3ba0 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -4968,6 +4968,8 @@ class BotDataViewer: 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") @@ -4976,12 +4978,14 @@ class BotDataViewer: 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) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ''', (feed_type, feed_url, channel_name, feed_name, check_interval, api_config_str, output_format, message_send_interval)) + (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 @@ -5036,10 +5040,6 @@ class BotDataViewer: updates.append('sort_config = ?') params.append(json.dumps(data['sort_config']) if data['sort_config'] else None) - if 'message_send_interval_seconds' in data: - updates.append('message_send_interval_seconds = ?') - params.append(data['message_send_interval_seconds']) - if not updates: return True # Nothing to update @@ -5540,115 +5540,10 @@ class BotDataViewer: raise def _should_include_item(self, item: dict[str, Any], filter_config: dict) -> bool: - """Check if an item should be included based on filter configuration (standalone version for preview)""" - import json - import re + """Check if an item should be included based on filter configuration (preview; same rules as FeedManager).""" + from modules.feed_filter_eval import item_passes_filter_config - if not filter_config: - return True - - try: - filter_config_dict = json.loads(filter_config) if isinstance(filter_config, str) else filter_config - except (json.JSONDecodeError, TypeError): - return True - - conditions = filter_config_dict.get('conditions', []) - if not conditions: - return True - - logic = filter_config_dict.get('logic', 'AND').upper() - - # Get raw data for field access - raw_data = item.get('raw', {}) - - # Helper to get nested values - def get_nested_value(data, path, default=''): - if not path or not data: - return default - parts = path.split('.') - value = data - for part in parts: - if isinstance(value, dict): - value = value.get(part) - elif isinstance(value, list): - try: - idx = int(part) - if 0 <= idx < len(value): - value = value[idx] - else: - return default - except (ValueError, TypeError): - return default - else: - return default - if value is None: - return default - return value if value is not None else default - - # Evaluate each condition - results = [] - for condition in conditions: - field_path = condition.get('field') - operator = condition.get('operator', 'equals') - - if not field_path: - continue - - # Get field value using nested access - field_value = get_nested_value(raw_data, field_path, '') - if not field_value and field_path.startswith('raw.'): - field_value = get_nested_value(raw_data, field_path[4:], '') - - if not field_value: - field_value = get_nested_value(item, field_path, '') - - # Convert to string for comparison - field_value_str = str(field_value).lower() if field_value is not None else '' - - # Evaluate condition - result = False - if operator == 'equals': - compare_value = str(condition.get('value', '')).lower() - result = field_value_str == compare_value - elif operator == 'not_equals': - compare_value = str(condition.get('value', '')).lower() - result = field_value_str != compare_value - elif operator == 'in': - values = [str(v).lower() for v in condition.get('values', [])] - result = field_value_str in values - elif operator == 'not_in': - values = [str(v).lower() for v in condition.get('values', [])] - result = field_value_str not in values - elif operator == 'matches': - pattern = condition.get('pattern', '') - if pattern: - try: - result = bool(re.search(pattern, str(field_value), re.IGNORECASE)) - except re.error: - result = False - elif operator == 'not_matches': - pattern = condition.get('pattern', '') - if pattern: - try: - result = not bool(re.search(pattern, str(field_value), re.IGNORECASE)) - except re.error: - result = True - elif operator == 'contains': - compare_value = str(condition.get('value', '')).lower() - result = compare_value in field_value_str - elif operator == 'not_contains': - compare_value = str(condition.get('value', '')).lower() - result = compare_value not in field_value_str - else: - result = True # Default to allowing if operator is unknown - - results.append(result) - - # Apply logic (AND or OR) - if logic == 'OR': - return any(results) - else: # AND (default) - return all(results) + return item_passes_filter_config(item, filter_config) def _parse_microsoft_date(self, date_str: str) -> Optional[datetime]: """Parse Microsoft JSON date format: /Date(timestamp-offset)/""" diff --git a/modules/web_viewer/templates/feeds.html b/modules/web_viewer/templates/feeds.html index a94f2b5..6102981 100644 --- a/modules/web_viewer/templates/feeds.html +++ b/modules/web_viewer/templates/feeds.html @@ -242,9 +242,9 @@ }'> Only send items that match these conditions. Leave empty to send all items.
- Operators: equals, not_equals, in, not_in, matches (regex), not_matches, contains, not_contains
+ Operators: equals, not_equals, in, not_in, matches (regex), not_matches, contains, not_contains, within_days, within_weeks (see docs/FEEDS.md)
Logic: AND (all conditions) or OR (any condition)
- Fields: Use raw.field or raw.nested.field for API fields + Fields: Use raw.field or raw.nested.field for API fields; use published for RSS time filters diff --git a/tests/test_feed_filter_eval.py b/tests/test_feed_filter_eval.py new file mode 100644 index 0000000..18e4d6c --- /dev/null +++ b/tests/test_feed_filter_eval.py @@ -0,0 +1,136 @@ +"""Unit tests for modules.feed_filter_eval (filter_config parsing and time windows).""" + +from datetime import datetime, timedelta, timezone + +from modules.feed_filter_eval import ( + item_passes_filter_config, + parse_item_field_as_datetime, + parse_microsoft_date, +) + + +def _utc_now(): + return datetime.now(timezone.utc) + + +class TestParseItemFieldAsDatetime: + def test_published_datetime(self): + dt = _utc_now() - timedelta(days=1) + item = {"published": dt} + assert parse_item_field_as_datetime(item, "published") is not None + + def test_microsoft_date_string(self): + # /Date(ms)/ format + ms = int(_utc_now().timestamp() * 1000) + s = f"/Date({ms})/" + item = {"raw": {"t": s}} + assert parse_item_field_as_datetime(item, "raw.t") is not None + + def test_iso_string_in_raw(self): + item = {"raw": {"when": "2024-06-01T12:00:00Z"}} + out = parse_item_field_as_datetime(item, "raw.when") + assert out is not None + assert out.year == 2024 + + +class TestWithinDaysWeeks: + def test_within_days_recent_passes(self): + now = _utc_now() + item = {"published": now - timedelta(days=5)} + cfg = { + "conditions": [ + {"field": "published", "operator": "within_days", "days": 28}, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is True + + def test_within_days_old_fails(self): + now = _utc_now() + item = {"published": now - timedelta(days=40)} + cfg = { + "conditions": [ + {"field": "published", "operator": "within_days", "days": 28}, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is False + + def test_within_weeks_equivalent(self): + now = _utc_now() + item = {"published": now - timedelta(days=10)} + cfg = { + "conditions": [ + {"field": "published", "operator": "within_weeks", "weeks": 1}, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is False + + def test_within_weeks_recent_passes(self): + now = _utc_now() + item = {"published": now - timedelta(days=5)} + cfg = { + "conditions": [ + {"field": "published", "operator": "within_weeks", "weeks": 1}, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is True + + def test_missing_date_strict_fails(self): + item = {"title": "x"} + cfg = { + "conditions": [ + {"field": "published", "operator": "within_days", "days": 7}, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is False + + def test_missing_date_include_if_missing(self): + item = {"title": "x"} + cfg = { + "conditions": [ + { + "field": "published", + "operator": "within_days", + "days": 7, + "include_if_missing": True, + }, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is True + + def test_within_days_without_days_key_fails(self): + item = {"published": _utc_now()} + cfg = { + "conditions": [ + {"field": "published", "operator": "within_days"}, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is False + + def test_combined_with_priority(self): + now = _utc_now() + item = { + "published": now - timedelta(days=1), + "raw": {"Priority": "high"}, + } + cfg = { + "conditions": [ + {"field": "published", "operator": "within_days", "days": 28}, + {"field": "Priority", "operator": "equals", "value": "high"}, + ], + "logic": "AND", + } + assert item_passes_filter_config(item, cfg) is True + + +class TestMicrosoftDateParse: + def test_parse_microsoft_date(self): + ms = int(_utc_now().timestamp() * 1000) + dt = parse_microsoft_date(f"/Date({ms}+0000)/") + assert dt is not None diff --git a/tests/test_feed_manager_extended.py b/tests/test_feed_manager_extended.py new file mode 100644 index 0000000..16ee16c --- /dev/null +++ b/tests/test_feed_manager_extended.py @@ -0,0 +1,299 @@ +"""Extended FeedManager tests: sort, format_message, mocked RSS/API fetch, queue processing.""" + +from __future__ import annotations + +import json +from configparser import ConfigParser +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock, Mock + +import pytest + +from modules.db_manager import DBManager +from modules.feed_manager import FeedManager + + +def _feed_manager_bot(mock_logger, db_path: str): + bot = Mock() + bot.logger = mock_logger + bot.config = ConfigParser() + bot.config.add_section("Feed_Manager") + bot.config.set("Feed_Manager", "feed_manager_enabled", "false") + bot.config.set("Feed_Manager", "max_message_length", "200") + bot.db_manager = DBManager(bot, db_path) + return bot + + +@pytest.fixture +def fm_with_db(mock_logger, tmp_path): + """FeedManager backed by a real file SQLite DB (feed tables from DBManager).""" + db_path = str(tmp_path / "feeds.db") + bot = _feed_manager_bot(mock_logger, db_path) + return FeedManager(bot) + + +def _fake_aiohttp_response(*, text_body: str | None = None, json_body: dict | list | None = None): + """Build a minimal async context manager compatible with async with session.get/post.""" + resp = Mock() + resp.status = 200 + if text_body is not None: + resp.text = AsyncMock(return_value=text_body) + if json_body is not None: + resp.json = AsyncMock(return_value=json_body) + + class _CM: + def __init__(self, r): + self._r = r + + async def __aenter__(self): + return self._r + + async def __aexit__(self, exc_type, exc, tb): + return False + + return _CM(resp) + + +class TestSortItems: + def test_sort_by_published_desc(self, fm_with_db): + fm = fm_with_db + older = datetime(2020, 1, 1, tzinfo=timezone.utc) + newer = datetime(2025, 6, 1, tzinfo=timezone.utc) + items = [ + {"id": "a", "title": "old", "published": older, "raw": {}}, + {"id": "b", "title": "new", "published": newer, "raw": {}}, + ] + out = fm._sort_items(items, {"field": "published", "order": "desc"}) + assert [x["id"] for x in out] == ["b", "a"] + + def test_sort_by_raw_numeric_timestamp_asc(self, fm_with_db): + fm = fm_with_db + items = [ + {"id": "2", "title": "t2", "raw": {"t": 200.0}, "published": None}, + {"id": "1", "title": "t1", "raw": {"t": 100.0}, "published": None}, + ] + out = fm._sort_items(items, {"field": "raw.t", "order": "asc"}) + assert [x["id"] for x in out] == ["1", "2"] + + def test_sort_empty_field_returns_unchanged(self, fm_with_db): + fm = fm_with_db + items = [{"id": "x", "title": "a"}] + out = fm._sort_items(items, {"field": "", "order": "desc"}) + assert out == items + + +class TestFormatMessage: + def test_basic_placeholders(self, fm_with_db): + fm = fm_with_db + now = datetime.now(timezone.utc) + feed = {"output_format": "{emoji} {title}\n{link}\n{date}", "feed_name": "news"} + item = { + "title": "Hello", + "link": "https://ex.com/a", + "description": "", + "published": now, + "raw": {}, + } + msg = fm.format_message(item, feed) + assert "Hello" in msg + assert "https://ex.com/a" in msg + assert "📢" in msg or "ℹ️" in msg # emoji from feed_name or default + + def test_strips_br_and_html_from_body(self, fm_with_db): + fm = fm_with_db + feed = {"output_format": "{body}"} + item = { + "title": "t", + "description": 'Line1
Line2

Para

', + "published": None, + "raw": {}, + } + msg = fm.format_message(item, feed) + assert " +T +Onehttp://e/1g1D1 +Twohttp://e/2g2D2 +""" + ctx = _fake_aiohttp_response(text_body=rss) + fm.session = Mock() + fm.session.get = Mock(return_value=ctx) + fm.session.closed = False + + feed = {"id": 1, "feed_url": "http://example.com/feed.xml"} + items = await fm.process_rss_feed(feed) + titles = {i["title"] for i in items} + assert titles == {"One", "Two"} + + @pytest.mark.asyncio + async def test_skips_already_processed(self, fm_with_db): + fm = fm_with_db + rss = """ +T +Oldhttp://e/1g1 +Newhttp://e/2g2 +""" + ctx = _fake_aiohttp_response(text_body=rss) + fm.session = Mock() + fm.session.get = Mock(return_value=ctx) + fm.session.closed = False + + with fm.bot.db_manager.connection() as conn: + conn.execute( + "INSERT INTO feed_activity (feed_id, item_id, item_title, message_sent) VALUES (?,?,?,1)", + (1, "g1", "Old"), + ) + conn.commit() + + feed = {"id": 1, "feed_url": "http://example.com/feed.xml"} + items = await fm.process_rss_feed(feed) + assert len(items) == 1 + assert items[0]["title"] == "New" + + +class TestProcessApiFeed: + @pytest.mark.asyncio + async def test_get_parses_items_path(self, fm_with_db): + fm = fm_with_db + payload = { + "data": { + "rows": [ + {"id": "10", "name": "Alpha", "created_at": 1700000000}, + ] + } + } + ctx = _fake_aiohttp_response(json_body=payload) + fm.session = Mock() + fm.session.get = Mock(return_value=ctx) + fm.session.closed = False + + api_config = json.dumps( + { + "response_parser": { + "items_path": "data.rows", + "id_field": "id", + "title_field": "name", + "timestamp_field": "created_at", + } + } + ) + feed = {"id": 2, "feed_url": "http://api.example.com/x", "api_config": api_config} + items = await fm.process_api_feed(feed) + assert len(items) == 1 + assert items[0]["id"] == "10" + assert items[0]["title"] == "Alpha" + + @pytest.mark.asyncio + async def test_post_json_body(self, fm_with_db): + fm = fm_with_db + payload = [{"id": "z", "title": "Zed", "created_at": 1600000000}] + ctx = _fake_aiohttp_response(json_body=payload) + fm.session = Mock() + fm.session.post = Mock(return_value=ctx) + fm.session.closed = False + + api_config = json.dumps( + { + "method": "POST", + "body": {"q": 1}, + "response_parser": { + "items_path": "", + "id_field": "id", + "title_field": "title", + "timestamp_field": "created_at", + }, + } + ) + feed = {"id": 3, "feed_url": "http://api.example.com/post", "api_config": api_config} + items = await fm.process_api_feed(feed) + assert len(items) == 1 + assert items[0]["title"] == "Zed" + + +class TestQueueAndProcessMessageQueue: + def test_queue_feed_message_inserts_row(self, fm_with_db): + fm = fm_with_db + with fm.bot.db_manager.connection() as conn: + conn.execute( + "INSERT INTO feed_subscriptions (feed_type, feed_url, channel_name, message_send_interval_seconds) VALUES (?,?,?,?)", + ("rss", "http://x", "#alerts", 0.0), + ) + conn.commit() + fid = conn.execute("SELECT last_insert_rowid()").fetchone()[0] + + feed = {"id": fid, "channel_name": "#alerts"} + item = {"id": "i1", "title": "T1"} + fm._queue_feed_message(feed, item, "hello mesh") + + with fm.bot.db_manager.connection() as conn: + row = conn.execute( + "SELECT message, sent_at FROM feed_message_queue WHERE feed_id = ?", + (fid,), + ).fetchone() + assert row[0] == "hello mesh" + assert row[1] is None + + @pytest.mark.asyncio + async def test_process_message_queue_sends_and_marks_sent(self, fm_with_db): + fm = fm_with_db + bot = fm.bot + bot.command_manager = MagicMock() + bot.command_manager.send_channel_message = AsyncMock(return_value=True) + + with bot.db_manager.connection() as conn: + conn.execute( + "INSERT INTO feed_subscriptions (feed_type, feed_url, channel_name, message_send_interval_seconds) VALUES (?,?,?,?)", + ("rss", "http://y", "#news", 0.0), + ) + fid = conn.execute("SELECT last_insert_rowid()").fetchone()[0] + conn.execute( + """INSERT INTO feed_message_queue (feed_id, channel_name, message, item_id, item_title, priority) + VALUES (?,?,?,?,?,0)""", + (fid, "#news", "queued body", "q1", "Queued title"), + ) + conn.commit() + + await fm.process_message_queue() + + bot.command_manager.send_channel_message.assert_awaited_once_with("#news", "queued body") + + with bot.db_manager.connection() as conn: + sent = conn.execute( + "SELECT sent_at IS NOT NULL FROM feed_message_queue WHERE item_id = ?", + ("q1",), + ).fetchone()[0] + assert sent == 1 + + act = conn.execute( + "SELECT COUNT(*) FROM feed_activity WHERE feed_id = ? AND item_id = ?", + (fid, "q1"), + ).fetchone()[0] + assert act == 1 diff --git a/tests/test_feed_manager_formatting.py b/tests/test_feed_manager_formatting.py index b389a65..212425a 100644 --- a/tests/test_feed_manager_formatting.py +++ b/tests/test_feed_manager_formatting.py @@ -137,6 +137,32 @@ class TestShouldSendItem: item = {"raw": {"Priority": "high", "Status": "closed"}} assert fm._should_send_item(feed, item) is False + def test_within_days_passes_recent(self, fm): + now = datetime.now(timezone.utc) + feed = { + "id": 1, + "filter_config": json.dumps({ + "conditions": [ + {"field": "published", "operator": "within_days", "days": 28}, + ], + }), + } + item = {"published": now - timedelta(days=5)} + assert fm._should_send_item(feed, item) is True + + def test_within_days_rejects_old(self, fm): + now = datetime.now(timezone.utc) + feed = { + "id": 1, + "filter_config": json.dumps({ + "conditions": [ + {"field": "published", "operator": "within_days", "days": 7}, + ], + }), + } + item = {"published": now - timedelta(days=30)} + assert fm._should_send_item(feed, item) is False + class TestFormatTimestamp: """Tests for _format_timestamp()."""