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