Coverage for src/local_deep_research/news/api.py: 90%
418 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Direct API functions for news system.
3These functions can be called directly by scheduler or wrapped by Flask endpoints.
4"""
6from typing import Dict, Any, Optional
7from datetime import datetime, timezone, UTC
8from loguru import logger
9from sqlalchemy.orm import Session
10import json
12from ..constants import ResearchStatus
13from ..llm.providers.base import normalize_provider
14from ..utilities.sql_utils import escape_like
15from .exceptions import (
16 InvalidLimitException,
17 SubscriptionNotFoundException,
18 SubscriptionCreationException,
19 SubscriptionUpdateException,
20 SubscriptionDeletionException,
21 DatabaseAccessException,
22 NewsFeedGenerationException,
23 NotImplementedException,
24 NewsAPIException,
25)
26from ..constants import DEFAULT_SEARCH_TOOL
27# Removed welcome feed import - no placeholders
28# get_db_setting not available in merged codebase
30# Generic detail surfaced to API clients in place of a raw exception string.
31# These messages are returned to the client via NewsAPIException.to_dict(), so
32# echoing str(e) would leak DB/driver internals (SQL, schema, paths) to callers
33# on shared/multi-user deployments (CWE-209). The real cause is always captured
34# server-side by the adjacent logger.exception(...) for diagnosis.
35_GENERIC_ERROR_DETAIL = "an internal error occurred"
38def _notify_scheduler_about_subscription_change(
39 action: str, user_id: Optional[str] = None
40):
41 """
42 Notify the scheduler about subscription changes.
44 Args:
45 action: The action performed (created, updated, deleted)
46 user_id: Optional user_id to use as fallback for username
47 """
48 try:
49 from flask import session as flask_session
50 from ..scheduler.background import get_background_job_scheduler
52 scheduler = get_background_job_scheduler()
53 if scheduler.is_running:
54 # Get username, with optional fallback to user_id
55 username = flask_session.get("username")
56 if not username and user_id:
57 username = user_id
59 # Get password from session password store
60 from ..database.session_passwords import session_password_store
62 session_id = flask_session.get("session_id")
63 password = None
64 if session_id and username: 64 ↛ 69line 64 didn't jump to line 69 because the condition on line 64 was always true
65 password = session_password_store.get_session_password(
66 username, session_id
67 )
69 if password and username:
70 # Update scheduler to reschedule subscriptions
71 scheduler.update_user_info(username, password)
72 logger.info(
73 f"Scheduler notified about {action} subscription for {username}"
74 )
75 else:
76 logger.warning(
77 f"Could not notify scheduler - no password available{' for ' + username if username else ''}"
78 )
79 except Exception:
80 logger.exception(
81 f"Could not notify scheduler about {action} subscription"
82 )
85def get_news_feed(
86 user_id: str = "anonymous",
87 limit: int = 20,
88 use_cache: bool = True,
89 focus: Optional[str] = None,
90 search_strategy: Optional[str] = None,
91 subscription_id: Optional[str] = None,
92) -> Dict[str, Any]:
93 """
94 Get personalized news feed by pulling from news_items table first, then research history.
96 Args:
97 user_id: User identifier
98 limit: Maximum number of cards to return
99 use_cache: Whether to use cached news
100 focus: Optional focus area for news
101 search_strategy: Override default recommendation strategy
103 Returns:
104 Dictionary with news items and metadata. Each item's ``findings``
105 field is the answer-only report content (post chat-mode-v2 refactor,
106 #3665 Fix B); structured top-N source links are in the separate
107 ``links`` array, not embedded in ``findings``.
108 """
109 # Validate limit - allow any positive number
110 if limit < 1:
111 raise InvalidLimitException(limit)
113 try:
114 logger.info(
115 f"get_news_feed called with user_id={user_id}, limit={limit}"
116 )
118 # News is always enabled for now - per-user settings will be handled later
119 # if not get_db_setting("news.enabled", True):
120 # return {"error": "News system is disabled", "news_items": []}
122 # Import database functions
123 from ..database.session_context import get_user_db_session
124 from ..database.models import ResearchHistory
126 news_items = []
127 remaining_limit = limit
129 # Query research history from user's database for news items
130 logger.info("Getting news items from research history")
131 try:
132 # Use the user_id provided to the function
133 with get_user_db_session(user_id) as db_session:
134 # Build query using ORM
135 query = db_session.query(ResearchHistory).filter(
136 ResearchHistory.status == ResearchStatus.COMPLETED
137 )
139 # Filter by subscription if provided
140 if subscription_id and subscription_id != "all":
141 # Use JSON containment for PostgreSQL or LIKE for SQLite.
142 # Note: research_meta is serialized via json.dumps which
143 # emits a space after the colon, so the LIKE pattern must
144 # include that space too — otherwise the filter silently
145 # matches zero rows. Mirrors the patterns used in
146 # get_subscriptions and get_subscription_history below.
147 query = query.filter(
148 ResearchHistory.research_meta.like(
149 f'%"subscription_id": "{escape_like(subscription_id)}"%',
150 escape="\\",
151 )
152 )
154 # Order by creation date and limit
155 results = (
156 query.order_by(ResearchHistory.created_at.desc())
157 .limit(remaining_limit * 2)
158 .all()
159 )
161 # Convert ORM objects to dictionaries for compatibility
162 results = [
163 {
164 "id": r.id,
165 "uuid_id": r.id, # In ResearchHistory, id is the UUID
166 "query": r.query,
167 "title": r.title
168 if hasattr(r, "title")
169 else None, # Include title field if exists
170 # created_at is NOT NULL (set to isoformat() on every
171 # insert), so it's always a usable timestamp string.
172 "created_at": r.created_at,
173 "completed_at": r.completed_at
174 if r.completed_at
175 else None,
176 "duration_seconds": r.duration_seconds
177 if hasattr(r, "duration_seconds")
178 else None,
179 "report_path": r.report_path
180 if hasattr(r, "report_path")
181 else None,
182 "report_content": r.report_content
183 if hasattr(r, "report_content")
184 else None, # Include database content
185 "research_meta": r.research_meta,
186 "status": r.status,
187 }
188 for r in results
189 ]
191 # Source links used to be parsed out of report_content via
192 # regex over `URL:` lines (when report_content held the
193 # inline ## Sources block). Now sources live in the
194 # research_resources table — fetch top-N for every row in
195 # ONE batched query (avoids N+1 in the loop below).
196 from ..web.services.report_assembly_service import (
197 get_research_source_links_batch,
198 )
200 research_ids_for_links = [r["id"] for r in results]
201 links_by_research_id = get_research_source_links_batch(
202 research_ids_for_links, db_session, limit=3
203 )
205 logger.info(f"Database returned {len(results)} research items")
206 if results and len(results) > 0:
207 logger.info(f"First row keys: {list(results[0].keys())}")
208 # Log first few items' metadata
209 for i, row in enumerate(results[:3]):
210 logger.info(
211 f"Item {i}: query='{row['query'][:50]}...', has meta={bool(row.get('research_meta'))}"
212 )
214 # Process results to find news items
215 processed_count = 0
216 error_count = 0
218 for row in results:
219 try:
220 # Parse metadata
221 metadata = {}
222 if row.get("research_meta"):
223 try:
224 # Handle both dict and string formats
225 if isinstance(row["research_meta"], dict):
226 metadata = row["research_meta"]
227 else:
228 metadata = json.loads(row["research_meta"])
229 except (json.JSONDecodeError, TypeError):
230 logger.exception("Error parsing metadata")
231 metadata = {}
233 # Check if this has news metadata (generated_headline or generated_topics)
234 # or if it's a news-related query
235 has_news_metadata = (
236 metadata.get("generated_headline") is not None
237 or metadata.get("generated_topics") is not None
238 )
240 query_lower = row["query"].lower()
241 is_news_query = (
242 has_news_metadata
243 or metadata.get("is_news_search")
244 or metadata.get("search_type") == "news_analysis"
245 or "breaking news" in query_lower
246 or "news stories" in query_lower
247 or (
248 "today" in query_lower
249 and (
250 "news" in query_lower
251 or "breaking" in query_lower
252 )
253 )
254 or "latest news" in query_lower
255 )
257 # Log the decision for first few items
258 if processed_count < 3 or error_count < 3: 258 ↛ 265line 258 didn't jump to line 265 because the condition on line 258 was always true
259 logger.info(
260 f"Item check: query='{row['query'][:30]}...', is_news_search={metadata.get('is_news_search')}, "
261 f"has_news_metadata={has_news_metadata}, is_news_query={is_news_query}"
262 )
264 # Only show items that have news metadata or are news queries
265 if is_news_query:
266 processed_count += 1
267 logger.info(
268 f"Processing research item #{processed_count}: {row['query'][:50]}..."
269 )
271 # Always use database content
272 findings = ""
273 summary = ""
274 report_content_db = row.get(
275 "report_content"
276 ) # Get database content
278 # Use database content
279 content = report_content_db
280 if content:
281 logger.debug(
282 f"Using database content for research {row['id']}"
283 )
285 # Process database content
286 lines = content.split("\n") if content else []
287 # `findings` is the answer-only report_content
288 # after the chat-mode-v2 refactor (#3665 Fix B,
289 # intentional). The legacy answer + ## Sources
290 # blob is gone: structured top-N source URLs live
291 # in the separate `links` array below, so snippet
292 # extraction is cleaner without Sources headers in
293 # the substrate.
294 findings = content
295 # Extract summary from first non-empty line
296 for line in lines: 296 ↛ 306line 296 didn't jump to line 306 because the loop on line 296 didn't complete
297 if line.strip() and not line.startswith("#"):
298 summary = line.strip()
299 break
300 else:
301 logger.debug(
302 f"No database content for research {row['id']}"
303 )
305 # Use stored headline/topics if available, otherwise generate
306 original_query = row["query"]
308 # Check for headline - first try database title, then metadata
309 headline = row.get("title") or metadata.get(
310 "generated_headline"
311 )
313 # For subscription results, generate headline from query if needed
314 if not headline and metadata.get("is_news_search"):
315 # Use subscription name or query as headline
316 subscription_name = metadata.get(
317 "subscription_name"
318 )
319 if subscription_name:
320 headline = f"News Update: {subscription_name}"
321 else:
322 # Generate headline from query
323 headline = f"News: {row['query'][:60]}..."
325 # Skip items without meaningful headlines or that are incomplete
326 if (
327 not headline
328 or headline == "[No headline available]"
329 ):
330 logger.debug(
331 f"Skipping item without headline: {row['id']}"
332 )
333 continue
335 # Skip items that are still in progress or suspended
336 if row["status"] in (
337 ResearchStatus.IN_PROGRESS,
338 ResearchStatus.SUSPENDED,
339 ):
340 logger.debug(
341 f"Skipping incomplete item: {row['id']} (status: {row['status']})"
342 )
343 continue
345 # Skip items without content (neither file nor database)
346 if not content:
347 logger.debug(
348 f"Skipping item without content: {row['id']}"
349 )
350 continue
352 # Use ID properly, preferring uuid_id
353 research_id = row.get("uuid_id") or str(row["id"])
355 # Use stored category and topics - no defaults
356 category = metadata.get("category")
357 if not category: 357 ↛ 360line 357 didn't jump to line 360 because the condition on line 357 was always true
358 category = "[Uncategorized]"
360 topics = metadata.get("generated_topics")
361 if not topics: 361 ↛ 367line 361 didn't jump to line 367 because the condition on line 361 was always true
362 topics = ["[No topics]"]
364 # Top-N links pulled from research_resources via
365 # the batch fetch above (no per-row DB query, no
366 # text parsing of report_content).
367 links = links_by_research_id.get(row["id"], [])
369 # Create news item from research
370 news_item = {
371 "id": f"news-{research_id}",
372 "headline": headline,
373 "category": category,
374 "summary": summary
375 or f"Research analysis for: {headline[:100]}",
376 "findings": findings,
377 "impact_score": metadata.get(
378 "impact_score", 0
379 ), # 0 indicates missing
380 "time_ago": _format_time_ago(row["created_at"]),
381 "upvotes": metadata.get("upvotes", 0),
382 "downvotes": metadata.get("downvotes", 0),
383 "source_url": f"/results/{research_id}",
384 "topics": topics, # Use generated topics
385 "links": links, # Add extracted links
386 "research_id": research_id,
387 "created_at": row["created_at"],
388 "duration_seconds": row.get("duration_seconds", 0),
389 "original_query": original_query, # Keep original query for reference
390 "is_news": metadata.get(
391 "is_news_search", False
392 ), # Flag for news searches
393 "news_date": metadata.get(
394 "news_date"
395 ), # If specific date for news
396 "news_source": metadata.get(
397 "news_source"
398 ), # If from specific source
399 "priority": metadata.get(
400 "priority", "normal"
401 ), # Priority level
402 }
404 news_items.append(news_item)
405 logger.info(f"Added news item: {headline[:50]}...")
407 if len(news_items) >= limit: 407 ↛ 408line 407 didn't jump to line 408 because the condition on line 407 was never true
408 break
410 except Exception:
411 error_count += 1
412 logger.exception(
413 f"Error processing research item with query: {row.get('query', 'UNKNOWN')[:100]}"
414 )
415 continue
417 logger.info(
418 f"Processing summary: total_results={len(results)}, processed={processed_count}, "
419 f"errors={error_count}, added={len(news_items)}"
420 )
422 # Log subscription-specific items if we were filtering
423 if subscription_id and subscription_id != "all":
424 sub_items = [
425 item for item in news_items if item.get("is_news", False)
426 ]
427 logger.info(
428 f"Subscription {subscription_id}: found {len(sub_items)} items"
429 )
431 except Exception as db_error:
432 logger.exception(f"Database error in research history: {db_error}")
433 raise DatabaseAccessException(
434 "research_history_query", _GENERIC_ERROR_DETAIL
435 )
437 # If no news items found, return empty list
438 if not news_items:
439 logger.info("No news items found, returning empty list")
440 news_items = []
442 logger.info(f"Returning {len(news_items)} news items to client")
444 # Determine the source
445 source = (
446 "news_items"
447 if any(item.get("is_news", False) for item in news_items)
448 else "research_history"
449 )
451 return {
452 "news_items": news_items[:limit],
453 "generated_at": datetime.now(timezone.utc).isoformat(),
454 "focus": focus,
455 "search_strategy": search_strategy or "default",
456 "total_items": len(news_items),
457 "source": source,
458 }
460 except NewsAPIException:
461 # Re-raise our custom exceptions
462 raise
463 except Exception:
464 logger.exception("Error getting news feed")
465 raise NewsFeedGenerationException(
466 _GENERIC_ERROR_DETAIL, user_id=user_id
467 )
470def get_subscription_history(
471 subscription_id: str, limit: int = 20
472) -> Dict[str, Any]:
473 """
474 Get research history for a specific subscription.
476 Args:
477 subscription_id: The subscription UUID
478 limit: Maximum number of history items to return
480 Returns:
481 Dict containing subscription info and its research history
482 """
483 try:
484 from ..database.session_context import get_user_db_session
485 from ..database.models import ResearchHistory
486 from ..database.models.news import NewsSubscription
488 # Get subscription details using ORM from user's encrypted database
489 with get_user_db_session() as session:
490 subscription = (
491 session.query(NewsSubscription)
492 .filter_by(id=subscription_id)
493 .first()
494 )
496 if not subscription:
497 raise SubscriptionNotFoundException(subscription_id) # noqa: TRY301 — re-raised by except NewsAPIException
499 # Convert to dict for response
500 subscription_dict = {
501 "id": subscription.id,
502 "query_or_topic": subscription.query_or_topic,
503 "subscription_type": subscription.subscription_type,
504 "refresh_interval_minutes": subscription.refresh_interval_minutes,
505 # refresh_count is derived from the actual research-run history
506 # below (NewsSubscription has no refresh_count column — reading
507 # it raised AttributeError and 500'd this endpoint).
508 "created_at": subscription.created_at.isoformat()
509 if subscription.created_at
510 else None,
511 "next_refresh": subscription.next_refresh.isoformat()
512 if subscription.next_refresh
513 else None,
514 }
516 # Now get research history from the research database. The
517 # NewsSubscription model has no user_id column — this codebase uses
518 # per-user encrypted databases, so "the subscription's user" is just
519 # whichever user's DB we found the subscription in. Reuse the Flask
520 # session username (same source the first get_user_db_session()
521 # call resolved). The previous version of this code did
522 # ``subscription_dict.get("user_id", "anonymous")`` against a dict
523 # that never carried a "user_id" key, so it always opened the
524 # "anonymous" user's database and silently returned an empty
525 # history for every real multi-user deployment.
526 with get_user_db_session() as db_session:
527 # Get all research runs that were triggered by this subscription
528 # Look for subscription_id in the research_meta JSON
529 # Note: JSON format has space after colon
530 like_pattern = (
531 f'%"subscription_id": "{escape_like(subscription_id)}"%'
532 )
533 logger.info(
534 f"Searching for research history with pattern: {like_pattern}"
535 )
537 history_items = (
538 db_session.query(ResearchHistory)
539 .filter(
540 ResearchHistory.research_meta.like(
541 like_pattern, escape="\\"
542 )
543 )
544 .order_by(ResearchHistory.created_at.desc())
545 .limit(limit)
546 .all()
547 )
549 # Convert to dict format for compatibility.
550 # ResearchHistory.id is the UUID PK (see comment on line 151);
551 # there is no separate uuid_id column, so populate both keys
552 # from h.id to preserve the downstream contract used by the
553 # processed_item['research_id']/url builders below.
554 history_items = [
555 {
556 "id": h.id,
557 "uuid_id": h.id,
558 "query": h.query,
559 "status": h.status,
560 # created_at/completed_at are Text columns already holding
561 # isoformat strings (see get_news_feed above), so use them
562 # directly — calling .isoformat() on a str raises
563 # AttributeError and previously broke this endpoint for any
564 # subscription that had research history.
565 "created_at": h.created_at,
566 "completed_at": h.completed_at if h.completed_at else None,
567 "duration_seconds": h.duration_seconds,
568 "research_meta": h.research_meta,
569 "report_path": h.report_path,
570 }
571 for h in history_items
572 ]
574 # Process history items
575 processed_history = []
576 for item in history_items:
577 processed_item = {
578 "research_id": item.get("uuid_id") or str(item.get("id")),
579 "query": item["query"],
580 "status": item["status"],
581 "created_at": item["created_at"],
582 "completed_at": item.get("completed_at"),
583 "duration_seconds": item.get("duration_seconds", 0),
584 "url": f"/progress/{item.get('uuid_id') or item.get('id')}",
585 }
587 # Parse metadata if available to get headline and topics
588 if item.get("research_meta"):
589 try:
590 # research_meta is a JSON column, so SQLAlchemy already
591 # deserializes it to a dict on read; only legacy/text rows
592 # arrive as a JSON string. Calling json.loads() on the dict
593 # raised TypeError that the bare except below swallowed,
594 # silently blanking the headline/topics for every item.
595 # Mirror the dict/str handling already used at the top of
596 # this module (see get_news_feed).
597 raw_meta = item["research_meta"]
598 meta = (
599 json.loads(raw_meta)
600 if isinstance(raw_meta, str)
601 else raw_meta
602 )
603 processed_item["triggered_by"] = meta.get(
604 "triggered_by", "subscription"
605 )
606 # Add headline and topics from metadata
607 processed_item["headline"] = meta.get(
608 "generated_headline", "[No headline]"
609 )
610 processed_item["topics"] = meta.get("generated_topics", [])
611 except Exception:
612 processed_item["headline"] = "[No headline]"
613 processed_item["topics"] = []
614 else:
615 processed_item["headline"] = "[No headline]"
616 processed_item["topics"] = []
618 processed_history.append(processed_item)
620 # Run count comes from the research history (the source of truth);
621 # the subscription row carries no counter.
622 subscription_dict["refresh_count"] = len(processed_history)
624 return {
625 "subscription": subscription_dict,
626 "history": processed_history,
627 "total_runs": len(processed_history),
628 }
630 except NewsAPIException:
631 # Re-raise our custom exceptions
632 raise
633 except Exception:
634 logger.exception("Error getting subscription history")
635 raise DatabaseAccessException(
636 "get_subscription_history", _GENERIC_ERROR_DETAIL
637 )
640def _format_time_ago(timestamp: str) -> str:
641 """Format timestamp as 'X hours ago' string.
643 Raises on unparseable input instead of masking it with a neutral label.
644 ResearchHistory.created_at is a NOT NULL column written as
645 datetime.now(UTC).isoformat() on every insert path, so a value that won't
646 parse means the row is corrupt — not a routine edge case. The only caller
647 (the per-row loop in get_news_feed) already wraps each row in a
648 try/except that logs the failure and skips the row, so a bad timestamp
649 surfaces in the logs and drops that one card rather than rendering it with
650 a misleading "Recently".
651 """
652 from dateutil import parser
654 dt = parser.parse(timestamp)
656 # If dt is naive, assume it's in UTC
657 if dt.tzinfo is None:
658 dt = dt.replace(tzinfo=timezone.utc)
660 now = datetime.now(timezone.utc)
661 diff = now - dt
663 if diff.days > 0:
664 return f"{diff.days} day{'s' if diff.days > 1 else ''} ago"
665 if diff.seconds > 3600:
666 hours = diff.seconds // 3600
667 return f"{hours} hour{'s' if hours > 1 else ''} ago"
668 if diff.seconds > 60:
669 minutes = diff.seconds // 60
670 return f"{minutes} minute{'s' if minutes > 1 else ''} ago"
671 return "Just now"
674def get_subscription(subscription_id: str) -> Optional[Dict[str, Any]]:
675 """
676 Get a single subscription by ID.
678 Args:
679 subscription_id: Subscription identifier
681 Returns:
682 Dictionary with subscription data or None if not found
683 """
684 try:
685 # Get subscription directly from user's encrypted database
686 from ..database.session_context import get_user_db_session
687 from ..database.models.news import NewsSubscription
689 with get_user_db_session() as db_session:
690 subscription = (
691 db_session.query(NewsSubscription)
692 .filter_by(id=subscription_id)
693 .first()
694 )
696 if not subscription:
697 raise SubscriptionNotFoundException(subscription_id) # noqa: TRY301 — re-raised by except NewsAPIException
699 # Convert to API format matching the template expectations
700 return {
701 "id": subscription.id,
702 "name": subscription.name or "",
703 "query_or_topic": subscription.query_or_topic,
704 "subscription_type": subscription.subscription_type,
705 "refresh_interval_minutes": subscription.refresh_interval_minutes,
706 "is_active": subscription.status == "active",
707 "status": subscription.status,
708 "folder_id": subscription.folder_id,
709 "model_provider": subscription.model_provider,
710 "model": subscription.model,
711 "search_strategy": subscription.search_strategy,
712 "custom_endpoint": subscription.custom_endpoint,
713 "search_engine": subscription.search_engine,
714 "search_iterations": subscription.search_iterations or 3,
715 "questions_per_iteration": subscription.questions_per_iteration
716 or 5,
717 "created_at": subscription.created_at.isoformat()
718 if subscription.created_at
719 else None,
720 "updated_at": subscription.updated_at.isoformat()
721 if subscription.updated_at
722 else None,
723 }
725 except NewsAPIException:
726 # Re-raise our custom exceptions
727 raise
728 except Exception:
729 logger.exception(f"Error getting subscription {subscription_id}")
730 raise DatabaseAccessException("get_subscription", _GENERIC_ERROR_DETAIL)
733def get_subscriptions(user_id: str) -> Dict[str, Any]:
734 """
735 Get all subscriptions for a user.
737 Args:
738 user_id: User identifier
740 Returns:
741 Dictionary with subscriptions list
742 """
743 try:
744 # Get subscriptions directly from user's encrypted database
745 from ..database.session_context import get_user_db_session
746 from ..database.models import ResearchHistory
747 from ..database.models.news import NewsSubscription
748 from sqlalchemy import func
750 sub_list = []
752 with get_user_db_session(user_id) as db_session:
753 # Query all subscriptions for this user
754 subscriptions = db_session.query(NewsSubscription).all()
756 for sub in subscriptions:
757 # Count actual research runs for this subscription.
758 # sub.id is an internal PK (not user input), but escape it
759 # too so every subscription_id LIKE in this module is
760 # uniformly wildcard-safe — a no-op for UUID-shaped ids.
761 like_pattern = f'%"subscription_id": "{escape_like(sub.id)}"%'
762 total_runs = (
763 db_session.query(func.count(ResearchHistory.id))
764 .filter(
765 ResearchHistory.research_meta.like(
766 like_pattern, escape="\\"
767 )
768 )
769 .scalar()
770 or 0
771 )
773 # Convert ORM object to API format
774 sub_dict = {
775 "id": sub.id,
776 "query": sub.query_or_topic,
777 "type": sub.subscription_type,
778 "refresh_minutes": sub.refresh_interval_minutes,
779 "created_at": sub.created_at.isoformat()
780 if sub.created_at
781 else None,
782 "next_refresh": sub.next_refresh.isoformat()
783 if sub.next_refresh
784 else None,
785 "last_refreshed": sub.last_refresh.isoformat()
786 if sub.last_refresh
787 else None,
788 "is_active": sub.status == "active",
789 "status": sub.status,
790 "total_runs": total_runs, # Use actual count from research_history
791 "name": sub.name or "",
792 "folder_id": sub.folder_id,
793 # Model/search config: included so consumers (the
794 # subscriptions UI, run-now callers) see the subscription's
795 # saved configuration instead of silently getting None and
796 # falling back to global defaults.
797 "model_provider": sub.model_provider,
798 "model": sub.model,
799 "search_strategy": sub.search_strategy,
800 "search_engine": sub.search_engine,
801 "custom_endpoint": sub.custom_endpoint,
802 "search_iterations": sub.search_iterations,
803 "questions_per_iteration": sub.questions_per_iteration,
804 }
805 sub_list.append(sub_dict)
807 return {"subscriptions": sub_list, "total": len(sub_list)}
809 except Exception:
810 logger.exception("Error getting subscriptions")
811 raise DatabaseAccessException(
812 "get_subscriptions", _GENERIC_ERROR_DETAIL
813 )
816def update_subscription(
817 subscription_id: str, data: Dict[str, Any]
818) -> Dict[str, Any]:
819 """
820 Update an existing subscription.
822 Args:
823 subscription_id: Subscription identifier
824 data: Dictionary with fields to update
826 Returns:
827 Dictionary with updated subscription data
828 """
829 try:
830 from ..database.session_context import get_user_db_session
831 from ..database.models.news import NewsSubscription
832 from datetime import datetime, timedelta
834 with get_user_db_session() as db_session:
835 # Get existing subscription
836 subscription = (
837 db_session.query(NewsSubscription)
838 .filter_by(id=subscription_id)
839 .first()
840 )
841 if not subscription:
842 raise SubscriptionNotFoundException(subscription_id) # noqa: TRY301 — re-raised by except NewsAPIException
844 # Update fields
845 if "name" in data:
846 subscription.name = data["name"]
847 if "query_or_topic" in data:
848 subscription.query_or_topic = data["query_or_topic"]
849 if "subscription_type" in data: 849 ↛ 850line 849 didn't jump to line 850 because the condition on line 849 was never true
850 subscription.subscription_type = data["subscription_type"]
851 if "refresh_interval_minutes" in data:
852 old_interval = subscription.refresh_interval_minutes
853 subscription.refresh_interval_minutes = data[
854 "refresh_interval_minutes"
855 ]
856 # Recalculate next_refresh if interval changed
857 if old_interval != subscription.refresh_interval_minutes: 857 ↛ 861line 857 didn't jump to line 861 because the condition on line 857 was always true
858 subscription.next_refresh = datetime.now(UTC) + timedelta(
859 minutes=subscription.refresh_interval_minutes
860 )
861 if "is_active" in data:
862 subscription.status = (
863 "active" if data["is_active"] else "paused"
864 )
865 if "status" in data: 865 ↛ 866line 865 didn't jump to line 866 because the condition on line 865 was never true
866 subscription.status = data["status"]
867 if "folder_id" in data:
868 subscription.folder_id = data["folder_id"]
869 if "model_provider" in data:
870 subscription.model_provider = normalize_provider(
871 data["model_provider"]
872 )
873 if "model" in data:
874 subscription.model = data["model"]
875 if "search_strategy" in data: 875 ↛ 876line 875 didn't jump to line 876 because the condition on line 875 was never true
876 subscription.search_strategy = data["search_strategy"]
877 if "custom_endpoint" in data: 877 ↛ 878line 877 didn't jump to line 878 because the condition on line 877 was never true
878 subscription.custom_endpoint = data["custom_endpoint"]
879 if "search_engine" in data:
880 subscription.search_engine = data["search_engine"]
881 if "search_iterations" in data:
882 subscription.search_iterations = data["search_iterations"]
883 if "questions_per_iteration" in data:
884 subscription.questions_per_iteration = data[
885 "questions_per_iteration"
886 ]
888 # N14: if this update touched the engine or provider, validate
889 # the resulting (effective) values against the user's egress
890 # policy before committing. Best-effort — needs a request
891 # context to resolve the current user's settings DB.
892 if "search_engine" in data or "model_provider" in data:
893 from flask import (
894 has_request_context,
895 session as flask_session,
896 )
898 _uid = (
899 flask_session.get("username")
900 if has_request_context()
901 else None
902 )
903 if _uid:
904 policy_reason = _validate_subscription_policy(
905 db_session,
906 _uid,
907 subscription.search_engine,
908 subscription.model_provider,
909 )
910 if policy_reason is not None:
911 raise SubscriptionUpdateException( # noqa: TRY301 — re-raised by except NewsAPIException
912 subscription_id, policy_reason
913 )
915 # Update timestamp
916 subscription.updated_at = datetime.now(UTC)
918 # Commit changes
919 db_session.commit()
921 # Notify scheduler about updated subscription
922 _notify_scheduler_about_subscription_change("updated")
924 # Convert to API format
925 return {
926 "status": "success",
927 "subscription": {
928 "id": subscription.id,
929 "name": subscription.name or "",
930 "query_or_topic": subscription.query_or_topic,
931 "subscription_type": subscription.subscription_type,
932 "refresh_interval_minutes": subscription.refresh_interval_minutes,
933 "is_active": subscription.status == "active",
934 "status": subscription.status,
935 "folder_id": subscription.folder_id,
936 "model_provider": subscription.model_provider,
937 "model": subscription.model,
938 "search_strategy": subscription.search_strategy,
939 "custom_endpoint": subscription.custom_endpoint,
940 "search_engine": subscription.search_engine,
941 "search_iterations": subscription.search_iterations or 3,
942 "questions_per_iteration": subscription.questions_per_iteration
943 or 5,
944 },
945 }
947 except NewsAPIException:
948 # Re-raise our custom exceptions
949 raise
950 except Exception:
951 logger.exception("Error updating subscription")
952 raise SubscriptionUpdateException(
953 subscription_id, _GENERIC_ERROR_DETAIL
954 )
957def _validate_subscription_policy(
958 db_session: Session, user_id, search_engine, model_provider
959) -> Optional[str]:
960 """Validate a subscription's search engine + LLM provider against the
961 user's current egress policy (N14).
963 News subscriptions store a fixed engine/provider that runs on a
964 schedule. The factory PEP catches a forbidden engine at execution
965 time, but validating at create/update time stops a forbidden config
966 from being persisted in the first place (and gives the user
967 immediate feedback). Returns a human-readable reason string when the
968 subscription should be rejected, or None when it's allowed.
970 Best-effort: if the settings backend or policy module is unavailable
971 (e.g. programmatic API use without a settings DB), validation is
972 skipped — the execution-time factory PEP remains the backstop.
973 """
974 try:
975 from ..utilities.db_utils import get_settings_manager
976 from ..security.egress.policy import (
977 PolicyDeniedError,
978 context_from_snapshot,
979 evaluate_engine,
980 evaluate_llm_endpoint,
981 )
983 settings_manager = get_settings_manager(db_session, user_id)
984 snapshot = settings_manager.get_settings_snapshot()
985 if not isinstance(snapshot, dict):
986 return None
987 primary = settings_manager.get_setting(
988 "search.tool", DEFAULT_SEARCH_TOOL
989 )
990 try:
991 ctx = context_from_snapshot(
992 snapshot, primary or DEFAULT_SEARCH_TOOL, username=user_id
993 )
994 except PolicyDeniedError as exc:
995 return f"egress policy refused: {exc.decision.reason}"
996 except ValueError as exc:
997 # An incoherent egress config makes context_from_snapshot
998 # raise ValueError. Surface it as a validation failure at
999 # subscription create/update time instead of letting the outer
1000 # except silently skip the check — otherwise the subscription
1001 # persists and only fails at execution time. Do not echo the raw
1002 # ValueError to the client (CWE-209): it can carry policy-config
1003 # internals; log it server-side instead. Mirrors the same fix in
1004 # web/routes/research_routes.py's start-research precheck.
1005 logger.bind(policy_audit=True).warning(
1006 "Subscription egress policy misconfigured",
1007 reason=str(exc),
1008 )
1009 return "egress policy is misconfigured"
1011 if search_engine:
1012 decision = evaluate_engine(
1013 search_engine, ctx, settings_snapshot=snapshot
1014 )
1015 if not decision.allowed:
1016 return (
1017 f"search engine '{search_engine}' is not permitted "
1018 f"under the current egress policy ({decision.reason})"
1019 )
1020 if model_provider:
1021 decision = evaluate_llm_endpoint(
1022 normalize_provider(model_provider),
1023 ctx,
1024 settings_snapshot=snapshot,
1025 )
1026 if not decision.allowed:
1027 return (
1028 f"LLM provider '{model_provider}' is not permitted "
1029 f"under the current egress policy ({decision.reason})"
1030 )
1031 return None
1032 except Exception:
1033 # Settings/policy unavailable → skip; execution-time PEP backstops.
1034 logger.debug("subscription policy pre-check skipped", exc_info=True)
1035 return None
1038def create_subscription(
1039 user_id: str,
1040 query: str,
1041 subscription_type: str = "search",
1042 refresh_minutes: Optional[int] = None,
1043 source_research_id: Optional[str] = None,
1044 model_provider: Optional[str] = None,
1045 model: Optional[str] = None,
1046 search_strategy: Optional[str] = None,
1047 custom_endpoint: Optional[str] = None,
1048 name: Optional[str] = None,
1049 folder_id: Optional[str] = None,
1050 is_active: bool = True,
1051 search_engine: Optional[str] = None,
1052 search_iterations: Optional[int] = None,
1053 questions_per_iteration: Optional[int] = None,
1054) -> Dict[str, Any]:
1055 """
1056 Create a new subscription for user.
1058 Args:
1059 user_id: User identifier
1060 query: Search query or topic
1061 subscription_type: "search" or "topic"
1062 refresh_minutes: Refresh interval in minutes
1064 Returns:
1065 Dictionary with subscription details
1066 """
1067 try:
1068 from ..database.session_context import get_user_db_session
1069 from ..database.models.news import NewsSubscription
1070 from datetime import datetime, timedelta
1071 import uuid
1073 # Get default refresh interval from settings if not provided
1074 # NOTE: This API function accesses the settings DB for convenience when used
1075 # within the Flask application context. For programmatic API access outside
1076 # the web context, callers should provide refresh_minutes explicitly to avoid
1077 # dependency on the settings database being initialized.
1079 with get_user_db_session(user_id) as db_session:
1080 # N14: reject a subscription whose engine/provider violates
1081 # the user's current egress policy before persisting it.
1082 policy_reason = _validate_subscription_policy(
1083 db_session, user_id, search_engine, model_provider
1084 )
1085 if policy_reason is not None:
1086 raise SubscriptionCreationException( # noqa: TRY301 — re-raised as-is by except SubscriptionCreationException
1087 policy_reason,
1088 {"query": query, "type": subscription_type},
1089 )
1091 if refresh_minutes is None:
1092 try:
1093 from ..utilities.db_utils import get_settings_manager
1095 settings_manager = get_settings_manager(db_session, user_id)
1096 refresh_minutes = settings_manager.get_setting(
1097 "news.subscription.refresh_minutes", 240
1098 )
1099 except (ImportError, AttributeError, TypeError):
1100 # Fallback for when settings DB is not available (e.g., programmatic API usage)
1101 logger.debug(
1102 "Settings manager not available, using default refresh_minutes"
1103 )
1104 refresh_minutes = 240 # Default to 4 hours
1105 # Create new subscription
1106 subscription = NewsSubscription(
1107 id=str(uuid.uuid4()),
1108 name=name,
1109 query_or_topic=query,
1110 subscription_type=subscription_type,
1111 refresh_interval_minutes=refresh_minutes,
1112 status="active" if is_active else "paused",
1113 model_provider=normalize_provider(model_provider),
1114 model=model,
1115 search_strategy=search_strategy or "news_aggregation",
1116 custom_endpoint=custom_endpoint,
1117 folder_id=folder_id,
1118 search_engine=search_engine,
1119 search_iterations=search_iterations,
1120 questions_per_iteration=questions_per_iteration,
1121 created_at=datetime.now(UTC),
1122 updated_at=datetime.now(UTC),
1123 last_refresh=None,
1124 next_refresh=datetime.now(UTC)
1125 + timedelta(minutes=refresh_minutes),
1126 source_id=source_research_id,
1127 )
1129 # Add to database
1130 db_session.add(subscription)
1131 db_session.commit()
1133 # Notify scheduler about new subscription
1134 _notify_scheduler_about_subscription_change("created", user_id)
1136 return {
1137 "status": "success",
1138 "subscription_id": subscription.id,
1139 "type": subscription_type,
1140 "query": query,
1141 "refresh_minutes": refresh_minutes,
1142 }
1144 except SubscriptionCreationException:
1145 # Already a structured creation error (e.g. the N14 policy
1146 # rejection) — propagate as-is instead of re-wrapping.
1147 raise
1148 except Exception:
1149 logger.exception("Error creating subscription")
1150 raise SubscriptionCreationException(
1151 _GENERIC_ERROR_DETAIL,
1152 {"query": query, "type": subscription_type},
1153 )
1156def delete_subscription(subscription_id: str) -> Dict[str, Any]:
1157 """
1158 Delete a subscription.
1160 Args:
1161 subscription_id: ID of subscription to delete
1163 Returns:
1164 Dictionary with status
1165 """
1166 try:
1167 from ..database.session_context import get_user_db_session
1168 from ..database.models.news import NewsSubscription
1170 with get_user_db_session() as db_session:
1171 subscription = (
1172 db_session.query(NewsSubscription)
1173 .filter_by(id=subscription_id)
1174 .first()
1175 )
1176 if subscription:
1177 db_session.delete(subscription)
1178 db_session.commit()
1180 # Notify scheduler about deleted subscription
1181 _notify_scheduler_about_subscription_change("deleted")
1183 return {"status": "success", "deleted": subscription_id}
1184 raise SubscriptionNotFoundException(subscription_id) # noqa: TRY301 — re-raised by except NewsAPIException
1185 except NewsAPIException:
1186 # Re-raise our custom exceptions
1187 raise
1188 except Exception:
1189 logger.exception("Error deleting subscription")
1190 raise SubscriptionDeletionException(
1191 subscription_id, _GENERIC_ERROR_DETAIL
1192 )
1195def get_votes_for_cards(card_ids: list, user_id: str) -> Dict[str, Any]:
1196 """
1197 Get vote counts and user's votes for multiple news cards.
1199 Args:
1200 card_ids: List of card IDs to get votes for
1201 user_id: User identifier (not used - per-user database)
1203 Returns:
1204 Dictionary with vote information for each card
1205 """
1206 from flask import session as flask_session, has_request_context
1207 from ..database.models.news import UserRating, RatingType
1208 from ..database.session_context import get_user_db_session
1210 # Resolve username before try block
1211 if not has_request_context(): 1211 ↛ 1218line 1211 didn't jump to line 1218 because the condition on line 1211 was always true
1212 # If called outside of request context (e.g., in tests), use user_id directly
1213 username = user_id if user_id else None
1214 if not username:
1215 raise ValueError("No username provided and no request context")
1216 else:
1217 # Get username from session
1218 username = flask_session.get("username")
1219 if not username:
1220 raise ValueError("No username in session")
1222 try:
1223 # Get database session
1224 with get_user_db_session(username) as db:
1225 results = {}
1227 for card_id in card_ids:
1228 # Get user's vote for this card
1229 user_vote = (
1230 db.query(UserRating)
1231 .filter_by(
1232 card_id=card_id, rating_type=RatingType.RELEVANCE
1233 )
1234 .first()
1235 )
1237 # Count total votes for this card
1238 upvotes = (
1239 db.query(UserRating)
1240 .filter_by(
1241 card_id=card_id,
1242 rating_type=RatingType.RELEVANCE,
1243 rating_value="up",
1244 )
1245 .count()
1246 )
1248 downvotes = (
1249 db.query(UserRating)
1250 .filter_by(
1251 card_id=card_id,
1252 rating_type=RatingType.RELEVANCE,
1253 rating_value="down",
1254 )
1255 .count()
1256 )
1258 results[card_id] = {
1259 "upvotes": upvotes,
1260 "downvotes": downvotes,
1261 "user_vote": user_vote.rating_value if user_vote else None,
1262 }
1264 return {"success": True, "votes": results}
1266 except Exception:
1267 logger.exception("Error getting votes for cards")
1268 raise
1271def submit_feedback(card_id: str, user_id: str, vote: str) -> Dict[str, Any]:
1272 """
1273 Submit feedback (vote) for a news card.
1275 Args:
1276 card_id: ID of the news card
1277 user_id: User identifier (not used - per-user database)
1278 vote: "up" or "down"
1280 Returns:
1281 Dictionary with updated vote counts
1282 """
1283 from flask import session as flask_session, has_request_context
1284 from sqlalchemy_utc import utcnow
1285 from ..database.models.news import UserRating, RatingType
1286 from ..database.session_context import get_user_db_session
1288 # Validate vote value
1289 if vote not in ["up", "down"]:
1290 raise ValueError(f"Invalid vote type: {vote}")
1292 # Resolve username before try block
1293 if not has_request_context(): 1293 ↛ 1300line 1293 didn't jump to line 1300 because the condition on line 1293 was always true
1294 # If called outside of request context (e.g., in tests), use user_id directly
1295 username = user_id if user_id else None
1296 if not username:
1297 raise ValueError("No username provided and no request context")
1298 else:
1299 # Get username from session
1300 username = flask_session.get("username")
1301 if not username:
1302 raise ValueError("No username in session")
1304 try:
1305 # Get database session
1306 with get_user_db_session(username) as db:
1307 # We don't check if the card exists in the database since news items
1308 # are generated dynamically and may not be stored as NewsCard entries
1310 # Check if user already voted on this card
1311 existing_rating = (
1312 db.query(UserRating)
1313 .filter_by(card_id=card_id, rating_type=RatingType.RELEVANCE)
1314 .first()
1315 )
1317 if existing_rating:
1318 # Update existing vote
1319 existing_rating.rating_value = vote
1320 existing_rating.created_at = utcnow()
1321 else:
1322 # Create new rating
1323 new_rating = UserRating(
1324 card_id=card_id,
1325 rating_type=RatingType.RELEVANCE,
1326 rating_value=vote,
1327 )
1328 db.add(new_rating)
1330 db.commit()
1332 # Count total votes for this card
1333 upvotes = (
1334 db.query(UserRating)
1335 .filter_by(
1336 card_id=card_id,
1337 rating_type=RatingType.RELEVANCE,
1338 rating_value="up",
1339 )
1340 .count()
1341 )
1343 downvotes = (
1344 db.query(UserRating)
1345 .filter_by(
1346 card_id=card_id,
1347 rating_type=RatingType.RELEVANCE,
1348 rating_value="down",
1349 )
1350 .count()
1351 )
1353 logger.info(
1354 f"Feedback submitted for card {card_id}: {vote} (up: {upvotes}, down: {downvotes})"
1355 )
1357 return {
1358 "success": True,
1359 "card_id": card_id,
1360 "vote": vote,
1361 "upvotes": upvotes,
1362 "downvotes": downvotes,
1363 }
1365 except Exception:
1366 logger.exception(f"Error submitting feedback for card {card_id}")
1367 raise
1370def research_news_item(card_id: str, depth: str = "quick") -> Dict[str, Any]:
1371 """
1372 Perform deeper research on a news item.
1374 Args:
1375 card_id: ID of the news card to research
1376 depth: Research depth - "quick", "detailed", or "report"
1378 Returns:
1379 Dictionary with research results
1380 """
1381 # TODO: Implement with per-user database for cards
1382 logger.warning(
1383 "research_news_item not yet implemented with per-user databases"
1384 )
1385 raise NotImplementedException("research_news_item")
1388def save_news_preferences(
1389 user_id: str, preferences: Dict[str, Any]
1390) -> Dict[str, Any]:
1391 """
1392 Save user preferences for news.
1394 Args:
1395 user_id: User identifier
1396 preferences: Dictionary of preferences to save
1398 Returns:
1399 Dictionary with status and message
1400 """
1401 # TODO: Implement with per-user database for preferences
1402 logger.warning(
1403 "save_news_preferences not yet implemented with per-user databases"
1404 )
1405 raise NotImplementedException("save_news_preferences")
1408def get_news_categories() -> Dict[str, Any]:
1409 """
1410 Get available news categories with counts.
1412 Returns:
1413 Dictionary with categories and statistics
1414 """
1415 # TODO: Implement with per-user database for categories
1416 logger.warning(
1417 "get_news_categories not yet implemented with per-user databases"
1418 )
1419 raise NotImplementedException("get_news_categories")