Coverage for src/local_deep_research/web/routers/news_flask_api.py: 93%
733 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2News API endpoints (FastAPI ``APIRouter``, mounted at ``/news/api``).
4The module name is a leftover: this code was FastAPI, was converted to a
5Flask blueprint to match LDR's then-Flask architecture, and the ASGI port
6converted it back. The name was kept across the port so importers and
7patch targets stay stable.
8"""
10import asyncio
12from fastapi import APIRouter, Depends, HTTPException, Request
13from fastapi.responses import JSONResponse
14from ..dependencies.auth import require_auth
15from ..dependencies.rate_limit import limiter
16from ..dependencies.threadpool import run_db_sync
18import uuid
19from typing import Any, Annotated
21from loguru import logger
23from ...llm.providers.base import normalize_provider
24from ...news import api
25from ...news.exceptions import NewsAPIException
26from ...news.folder_manager import FolderManager
27from ...database.models import SubscriptionFolder
29from ...database.session_context import get_user_db_session
30from ...settings.env_registry import get_env_setting
31from ...utilities.db_utils import get_settings_manager
32from ...utilities.url_utils import is_safe_custom_llm_endpoint
33from ..dependencies.json_body import json_body_error, read_json_dict
35# Hard ceiling for user-supplied ``limit`` query params on the news
36# endpoints in this module. Matches the ``max_value`` of the
37# ``news.feed.default_limit`` setting so a direct API caller cannot request a
38# larger page than the UI can configure. Shared by the feed and
39# subscription-history endpoints here so their caps stay in lockstep.
40NEWS_FEED_MAX_LIMIT = 100
41# The service currently performs three queries per card ID. Keep the batch
42# endpoint bounded to one feed page so a malformed or oversized request cannot
43# amplify database work independently of the feed's own pagination ceiling.
44NEWS_BATCH_FEEDBACK_MAX_CARD_IDS = NEWS_FEED_MAX_LIMIT
46# Subscription scheduling and search controls arrive as untyped JSON. Keep
47# their request-boundary limits in one place so the create route and both
48# update routes cannot drift apart. One minute and one week are both offered
49# by the current UI, so those are the inclusive refresh bounds.
50NEWS_SUBSCRIPTION_MIN_REFRESH_MINUTES = 1
51NEWS_SUBSCRIPTION_MAX_REFRESH_MINUTES = 10_080
52NEWS_SUBSCRIPTION_MAX_QUERY_LENGTH = 10_000
53NEWS_SUBSCRIPTION_MAX_SEARCH_ITERATIONS = 20
54NEWS_SUBSCRIPTION_MAX_QUESTIONS_PER_ITERATION = 10
57def _subscription_validation_error(message: str) -> JSONResponse:
58 """Return the stable simple-error shape used by the news API."""
59 return JSONResponse({"error": message}, status_code=400)
62def _validate_subscription_payload(
63 data: dict[str, Any],
64 *,
65 require_query: bool = False,
66 query_field: str = "query",
67 refresh_field: str = "refresh_minutes",
68) -> JSONResponse | None:
69 """Validate fields shared by subscription create/update payloads.
71 ``bool`` is a subclass of ``int`` in Python, so the numeric checks use an
72 exact type comparison. This helper intentionally runs before any DB,
73 scheduler, endpoint-resolution, or service work at each call site.
74 """
75 has_query = query_field in data
76 if require_query and not has_query:
77 return _subscription_validation_error("query is required")
78 if has_query:
79 query = data[query_field]
80 if query is None and require_query:
81 return _subscription_validation_error("query is required")
82 if not isinstance(query, str):
83 return _subscription_validation_error("query must be a string")
84 if not query.strip():
85 return _subscription_validation_error("query is required")
86 if len(query) > NEWS_SUBSCRIPTION_MAX_QUERY_LENGTH:
87 return _subscription_validation_error(
88 "query exceeds maximum length of "
89 f"{NEWS_SUBSCRIPTION_MAX_QUERY_LENGTH} characters"
90 )
92 if "is_active" in data and not isinstance(data["is_active"], bool):
93 return _subscription_validation_error("is_active must be a boolean")
95 for field in ("name", "folder_id", "model_provider"):
96 if (
97 field in data
98 and data[field] is not None
99 and not isinstance(data[field], str)
100 ):
101 return _subscription_validation_error(
102 f"{field} must be a string or null"
103 )
105 integer_ranges = (
106 (
107 refresh_field,
108 NEWS_SUBSCRIPTION_MIN_REFRESH_MINUTES,
109 NEWS_SUBSCRIPTION_MAX_REFRESH_MINUTES,
110 ),
111 ("search_iterations", 1, NEWS_SUBSCRIPTION_MAX_SEARCH_ITERATIONS),
112 (
113 "questions_per_iteration",
114 1,
115 NEWS_SUBSCRIPTION_MAX_QUESTIONS_PER_ITERATION,
116 ),
117 )
118 for field, minimum, maximum in integer_ranges:
119 if field not in data:
120 continue
121 value = data[field]
122 if type(value) is not int or not minimum <= value <= maximum:
123 return _subscription_validation_error(
124 f"{field} must be an integer between {minimum} and {maximum}"
125 )
127 return None
130def require_scheduler_control(request: Request) -> None:
131 """Dependency that gates global scheduler control behind a setting.
133 The news scheduler is a global singleton — starting, stopping, or
134 triggering it affects all users. This checks the
135 ``news.scheduler.allow_api_control`` setting (env var
136 ``LDR_NEWS_SCHEDULER_ALLOW_API_CONTROL``, default ``false``) and
137 raises HTTPException 403 when disabled.
138 """
139 if not get_env_setting("news.scheduler.allow_api_control", False):
140 # Every route using this dependency declares Depends(require_auth)
141 # ahead of it, and FastAPI resolves declared dependencies in order,
142 # so session["username"] is guaranteed here. Index rather than
143 # .get(..., "unknown"): this is an audit line for a blocked
144 # privileged action, and a shared "unknown" placeholder silently
145 # merges every such attempt into one unattributable entry if that
146 # ordering invariant is ever broken. Failing fast surfaces the
147 # broken invariant instead of quietly degrading the audit trail.
148 # Ported from #5549, which made the same change on main.
149 username = request.session["username"]
150 remote_addr = request.client.host if request.client else "unknown"
151 logger.warning(
152 "Scheduler API control blocked (user={}, ip={})",
153 username,
154 remote_addr,
155 )
156 raise HTTPException(
157 status_code=403,
158 detail=(
159 "Scheduler API control is disabled. "
160 "Contact your administrator to enable it."
161 ),
162 )
165def safe_error_message(e: Exception, context: str = "") -> str:
166 """
167 Return a safe error message that doesn't expose internal details.
169 Args:
170 e: The exception
171 context: Optional context about what was being attempted
173 Returns:
174 A generic error message safe for external users
175 """
176 # Log the actual error for debugging
177 logger.exception(f"Error in {context}")
179 # Return generic messages based on exception type
180 if isinstance(e, ValueError):
181 return "Invalid input provided"
182 if isinstance(e, KeyError):
183 return "Required data missing"
184 if isinstance(e, TypeError):
185 return "Invalid data format"
186 # Generic message for production
187 return f"An error occurred{f' while {context}' if context else ''}"
190def _is_job_owned_by_user(job, username, scheduler):
191 """Check if an APScheduler job belongs to a specific user."""
192 # Primary: all news scheduler jobs pass username as first arg
193 if hasattr(job, "args") and job.args and job.args[0] == username:
194 return True
195 # Fallback: check the tracked scheduled_jobs set
196 if hasattr(scheduler, "user_sessions"):
197 session_info = scheduler.user_sessions.get(username, {})
198 if job.id in session_info.get("scheduled_jobs", set()):
199 return True
200 return False
203# In Flask, this Blueprint had no prefix but was nested under /news.
204# In FastAPI, routers are flat, so we add the full prefix.
205router = APIRouter(prefix="/news/api", tags=["news_api"])
207# Shared rate limits for POST endpoints (port of main's news_routes
208# Flask-Limiter shared limits — one bucket per scope).
209_news_create_limit = limiter.shared_limit("10 per minute", scope="news_create")
210_news_research_limit = limiter.shared_limit(
211 "5 per minute", scope="news_research"
212)
213_news_feedback_limit = limiter.shared_limit(
214 "30 per minute", scope="news_feedback"
215)
216_news_preferences_limit = limiter.shared_limit(
217 "10 per minute", scope="news_preferences"
218)
219# NOTE: Routes use username (not .get()) intentionally.
220# Depends(require_auth) guarantees the key exists; direct access fails
221# fast if the dependency is ever removed.
223# Components are initialized in api.py
226def get_user_id(request=None):
227 """Get current user ID from the FastAPI request session.
229 Pass `request` to read it from the active session. If omitted (legacy
230 callers), returns None — these routes should be using
231 `Depends(require_auth)` directly instead of this helper.
232 """
233 if request is None:
234 return None
235 username = (
236 request.session.get("username") if hasattr(request, "session") else None
237 )
238 if not username:
239 # For news, we need authenticated users
240 return None
242 return username
245def _start_research_in_process(request, request_data: dict, username: str):
246 """Start a research run by calling the research router's sync entrypoint
247 directly, instead of a loopback HTTP POST.
249 The two news "run subscription now" / "check overdue" paths used to POST
250 to ``<request.base_url>/research/api/start`` via ``safe_post``. Because
251 ``request.base_url`` is derived from the client-controlled ``Host``
252 header and ``safe_post`` was called with ``allow_localhost=True,
253 allow_private_ips=True``, a spoofed ``Host`` turned this into an SSRF: the
254 server would issue a request carrying the caller's own session cookie +
255 CSRF token to an attacker-named internal host, and ``check-overdue``
256 reflected the raw response body back to the client. Calling the handler
257 in-process removes the network hop (and the SSRF) entirely — the same
258 move main made in #4834.
260 Returns the research-start result as a plain dict (``status``,
261 ``research_id``, …), normalizing ``_start_research_sync``'s JSONResponse
262 error returns into the same dict shape the callers expect.
263 """
264 import json as _json
266 from .research import _start_research_sync
268 # base_url is only embedded cosmetically (server_url for link
269 # generation) inside _start_research_sync — it drives no network call,
270 # so passing the Host-derived value here is inert (matches what the
271 # direct /research/api/start endpoint passes).
272 base_url = str(request.base_url).rstrip("/")
273 session_id = request.session.get("session_id")
275 result = _start_research_sync(request_data, username, base_url, session_id)
276 if isinstance(result, JSONResponse): 276 ↛ 282line 276 didn't jump to line 282 because the condition on line 276 was never true
277 # _start_research_sync returns a JSONResponse on failure with a
278 # meaningful status_code (400 validation, 409 duplicate, 429
279 # capacity). Preserve it under `_http_status` so the caller can
280 # propagate the real code instead of collapsing everything to 500
281 # (the old loopback forwarded response.status_code).
282 try:
283 body = _json.loads(bytes(result.body).decode("utf-8"))
284 except Exception:
285 body = {"status": "error", "message": "Failed to start research"}
286 body["_http_status"] = result.status_code
287 return body
288 if isinstance(result, dict): 288 ↛ 293line 288 didn't jump to line 293 because the condition on line 288 was always true
289 # Normalize a ResearchStatus enum status to its str value so the
290 # callers' `in ("success", "queued")` check is robust.
291 status = result.get("status")
292 return {**result, "status": getattr(status, "value", status)}
293 return {"status": "error", "message": "Failed to start research"}
296async def _reject_custom_endpoint_async(custom_endpoint):
297 """Off-loop wrapper for :func:`_reject_custom_endpoint`.
299 SSRF validation ends in ``socket.getaddrinfo`` (see
300 ``security/ssrf_validator.py``), which is a blocking OS resolver call
301 with no timeout parameter — it cannot be bounded in place. Called
302 directly from an ``async def`` handler it stalls the event loop, and
303 the server runs single-worker, so every other in-flight request stalls
304 with it. An authenticated caller supplying a hostname that resolves
305 slowly or black-holes would hold the whole process for the resolver's
306 full retry budget. Hand it to a worker thread instead.
307 """
308 return await asyncio.to_thread(_reject_custom_endpoint, custom_endpoint)
311def _reject_custom_endpoint(custom_endpoint):
312 """Return a 400 JSONResponse if ``custom_endpoint`` fails SSRF
313 validation, else None.
315 Blocking (DNS). From an ``async def`` handler call
316 :func:`_reject_custom_endpoint_async` instead of this directly.
318 Rejects cloud-metadata / link-local targets at the request boundary as
319 fail-fast defense-in-depth (the OpenAI-compatible provider re-validates
320 the same URL via assert_base_url_safe before the client is built).
321 Private IPs and localhost are allowed because local LLMs live there;
322 scheme-less endpoints are normalized exactly as the provider does.
323 """
324 if is_safe_custom_llm_endpoint(custom_endpoint):
325 return None
326 return JSONResponse(
327 {"success": False, "error": "Invalid custom endpoint URL"},
328 status_code=400,
329 )
332# The documented sentinel meaning "no subscription filter". Upstream
333# (#5603) exempts it from the UUID check on the feed route; without
334# this, ?subscription_id=all 400s even though news/api.py still
335# implements the "all" branch.
336NO_SUBSCRIPTION_FILTER = "all"
339def _is_valid_uuid(value: str) -> bool:
340 """Return True if ``value`` parses as a UUID, False otherwise.
342 Used to validate subscription_id params before they reach the
343 LIKE-pattern queries in ``news/api.py``. Without this check, a request
344 like ``?subscription_id=%`` would expand the LIKE filter and match
345 arbitrary subscriptions (enumeration vector).
346 """
347 try:
348 uuid.UUID(str(value))
349 except (ValueError, AttributeError, TypeError):
350 return False
351 return True
354@router.get("/feed")
355def get_news_feed(
356 request: Request, username: Annotated[str, Depends(require_auth)]
357) -> Any:
358 """
359 Get personalized news feed for user.
361 Query params:
362 user_id: User identifier (default: anonymous)
363 limit: Maximum number of cards to return (default: 20)
364 use_cache: Whether to use cached news (default: true)
365 strategy: Override default recommendation strategy
366 focus: Optional focus area for news
367 """
368 try:
369 # Get current user (require_auth ensures we have one)
370 user_id = username
371 logger.info(f"News feed requested by user: {user_id}")
373 # Get query parameters
374 with get_user_db_session(username) as _db:
375 settings_manager = get_settings_manager(_db, username)
376 default_limit = settings_manager.get_setting(
377 "news.feed.default_limit"
378 )
379 try:
380 limit = int(request.query_params.get("limit", default_limit))
381 except (TypeError, ValueError):
382 limit = int(default_limit)
383 limit = max(1, min(limit, NEWS_FEED_MAX_LIMIT))
384 use_cache = (
385 request.query_params.get("use_cache", "true").lower() == "true"
386 )
387 strategy = request.query_params.get("strategy")
388 focus = request.query_params.get("focus")
389 subscription_id = request.query_params.get("subscription_id")
390 if (
391 subscription_id
392 and subscription_id != NO_SUBSCRIPTION_FILTER
393 and not _is_valid_uuid(subscription_id)
394 ):
395 return JSONResponse(
396 {"success": False, "error": "Invalid subscription_id"},
397 status_code=400,
398 )
400 logger.info(
401 f"News feed params: limit={limit}, subscription_id={subscription_id}, focus={focus}"
402 )
404 # Call the direct API function (now synchronous)
405 result = api.get_news_feed(
406 user_id=user_id,
407 limit=limit,
408 use_cache=use_cache,
409 focus=focus,
410 search_strategy=strategy,
411 subscription_id=subscription_id,
412 )
414 # Check for errors in result
415 if "error" in result and result.get("news_items") == []:
416 # Sanitize error message before returning to client
417 safe_msg = safe_error_message(
418 Exception(result["error"]), context="get_news_feed"
419 )
420 status = 400 if "must be between" in result["error"] else 500
421 return JSONResponse(
422 {"error": safe_msg, "news_items": []}, status_code=status
423 )
425 # Debug: Log the result before returning
426 logger.info(
427 f"API returning {len(result.get('news_items', []))} news items"
428 )
429 if result.get("news_items"):
430 logger.info(
431 f"First item ID: {result['news_items'][0].get('id', 'NO_ID')}"
432 )
434 return result
436 except NewsAPIException:
437 # api.get_news_feed() raises DatabaseAccessException /
438 # NewsFeedGenerationException on failure — let the NewsAPIException
439 # handler in fastapi_app.py render exc.to_dict() (error_code,
440 # details) instead of downgrading it to a generic {"error": ...}
441 # below.
442 raise
443 except Exception as e:
444 return JSONResponse(
445 {
446 "error": safe_error_message(e, "getting news feed"),
447 "news_items": [],
448 },
449 status_code=500,
450 )
453@router.post("/subscribe")
454@_news_create_limit
455async def create_subscription(
456 request: Request, username: Annotated[str, Depends(require_auth)]
457) -> Any:
458 """
459 Create a new subscription for user.
461 JSON body:
462 query: Search query or topic
463 subscription_type: "search" or "topic" (default: "search")
464 refresh_minutes: Refresh interval in minutes (default: from settings)
465 """
466 try:
467 data, _body_err = await read_json_dict(
468 request, "simple", "No JSON data provided"
469 )
470 if _body_err is not None:
471 return _body_err
472 except Exception:
473 # Handle invalid JSON
474 return JSONResponse({"error": "Invalid JSON data"}, status_code=400)
476 try:
477 validation_error = _validate_subscription_payload(
478 data, require_query=True
479 )
480 if validation_error is not None:
481 return validation_error
483 # Get current user
484 user_id = username
486 # Extract parameters
487 query = data.get("query")
488 subscription_type = data.get("subscription_type", "search")
489 refresh_minutes = data.get(
490 "refresh_minutes"
491 ) # Will use default from api.py
493 # Extract model configuration (optional)
494 # Normalised to the lowercase canonical form the rest of the
495 # route/service layer compares against (and falsy -> None), so a
496 # subscription created with "OLLAMA" still matches later lookups.
497 model_provider = normalize_provider(data.get("model_provider"))
498 model = data.get("model")
499 search_strategy = data.get("search_strategy", "news_aggregation")
500 custom_endpoint = data.get("custom_endpoint")
502 # Extract additional fields
503 name = data.get("name")
504 folder_id = data.get("folder_id")
505 is_active = data.get("is_active", True)
506 search_engine = data.get("search_engine")
507 search_iterations = data.get("search_iterations")
508 questions_per_iteration = data.get("questions_per_iteration")
510 bad_endpoint = await _reject_custom_endpoint_async(custom_endpoint)
511 if bad_endpoint is not None:
512 return bad_endpoint
514 # Sync DB work (SQLCipher session + commit) — threadpool, not loop.
515 return await run_db_sync(
516 api.create_subscription,
517 user_id=user_id,
518 query=query,
519 subscription_type=subscription_type,
520 refresh_minutes=refresh_minutes,
521 model_provider=model_provider,
522 model=model,
523 search_strategy=search_strategy,
524 custom_endpoint=custom_endpoint,
525 name=name,
526 folder_id=folder_id,
527 is_active=is_active,
528 search_engine=search_engine,
529 search_iterations=search_iterations,
530 questions_per_iteration=questions_per_iteration,
531 )
533 except ValueError as e:
534 return JSONResponse(
535 {"error": safe_error_message(e, "creating subscription")},
536 status_code=400,
537 )
538 except NewsAPIException:
539 # api.create_subscription() always raises SubscriptionCreationException
540 # on failure (e.g. the N14 egress-policy rejection, or a wrapped DB
541 # error) — let the NewsAPIException handler in fastapi_app.py render
542 # exc.to_dict() (error_code, details) instead of downgrading it to a
543 # generic {"error": ...} below.
544 raise
545 except Exception as e:
546 return JSONResponse(
547 {"error": safe_error_message(e, "creating subscription")},
548 status_code=500,
549 )
552@router.post("/vote")
553async def vote_on_news(
554 request: Request, username: Annotated[str, Depends(require_auth)]
555) -> Any:
556 """
557 Submit vote on a news item.
559 JSON body:
560 card_id: ID of the news card
561 vote: "up" or "down"
562 """
563 try:
564 data, _body_err = await read_json_dict(
565 request, "simple", "No JSON data provided"
566 )
567 if _body_err is not None:
568 return _body_err
570 # Get current user
571 user_id = username
573 card_id = data.get("card_id")
574 vote = data.get("vote")
576 # Validate
577 if not all([card_id, vote]):
578 return JSONResponse(
579 {"error": "card_id and vote are required"}, status_code=400
580 )
582 # Sync DB work (SQLCipher session + commit) — threadpool, not loop.
583 return await run_db_sync(
584 api.submit_feedback, card_id=card_id, user_id=user_id, vote=vote
585 )
587 except ValueError as e:
588 error_msg = str(e)
589 if "not found" in error_msg.lower():
590 return JSONResponse(
591 {"error": "Resource not found"}, status_code=404
592 )
593 return JSONResponse(
594 {"error": safe_error_message(e, "submitting vote")},
595 status_code=400,
596 )
597 except Exception as e:
598 return JSONResponse(
599 {"error": safe_error_message(e, "submitting vote")},
600 status_code=500,
601 )
604@router.post("/feedback/batch")
605async def get_batch_feedback(
606 request: Request, username: Annotated[str, Depends(require_auth)]
607) -> Any:
608 """
609 Get feedback (votes) for multiple news cards.
610 JSON body:
611 card_ids: List of card IDs
612 """
613 try:
614 data, _body_err = await read_json_dict(
615 request, "simple", "No JSON data provided"
616 )
617 if _body_err is not None:
618 return _body_err
619 card_ids = data.get("card_ids", [])
620 if not isinstance(card_ids, list) or any(
621 not isinstance(card_id, str) or not card_id.strip()
622 for card_id in card_ids
623 ):
624 return JSONResponse(
625 {"error": "card_ids must be a list of non-empty strings"},
626 status_code=400,
627 )
628 if len(card_ids) > NEWS_BATCH_FEEDBACK_MAX_CARD_IDS:
629 return JSONResponse(
630 {
631 "error": (
632 "card_ids exceeds maximum "
633 f"({NEWS_BATCH_FEEDBACK_MAX_CARD_IDS})"
634 )
635 },
636 status_code=400,
637 )
638 if not card_ids:
639 return {"votes": {}}
641 # Get current user
642 user_id = username
644 # Sync DB work (SQLCipher session) — threadpool, not loop.
645 return await run_db_sync(
646 api.get_votes_for_cards, card_ids=card_ids, user_id=user_id
647 )
649 except ValueError as e:
650 error_msg = str(e)
651 if "not found" in error_msg.lower(): 651 ↛ 655line 651 didn't jump to line 655 because the condition on line 651 was always true
652 return JSONResponse(
653 {"error": "Resource not found"}, status_code=404
654 )
655 return JSONResponse(
656 {"error": safe_error_message(e, "getting votes")},
657 status_code=400,
658 )
659 except Exception as e:
660 logger.exception("Error getting batch feedback")
661 return JSONResponse(
662 {"error": safe_error_message(e, "getting votes")},
663 status_code=500,
664 )
667@router.post("/feedback/{card_id}")
668@_news_feedback_limit
669async def submit_feedback(
670 card_id: str,
671 request: Request,
672 username: Annotated[str, Depends(require_auth)],
673) -> Any:
674 """
675 Submit feedback (vote) for a news card.
677 JSON body:
678 vote: "up" or "down"
679 """
680 try:
681 data, _body_err = await read_json_dict(
682 request, "simple", "No JSON data provided"
683 )
684 if _body_err is not None:
685 return _body_err
687 # Get current user
688 user_id = username
689 vote = data.get("vote")
691 # Validate
692 if not vote:
693 return JSONResponse({"error": "vote is required"}, status_code=400)
695 # Sync DB work (SQLCipher session + commit) — threadpool, not loop.
696 return await run_db_sync(
697 api.submit_feedback, card_id=card_id, user_id=user_id, vote=vote
698 )
700 except ValueError as e:
701 error_msg = str(e)
702 if "not found" in error_msg.lower():
703 return JSONResponse(
704 {"error": "Resource not found"}, status_code=404
705 )
706 if "must be" in error_msg.lower():
707 return JSONResponse(
708 {"error": "Invalid input value"}, status_code=400
709 )
710 return JSONResponse(
711 {"error": safe_error_message(e, "submitting feedback")},
712 status_code=400,
713 )
714 except Exception as e:
715 return JSONResponse(
716 {"error": safe_error_message(e, "submitting feedback")},
717 status_code=500,
718 )
721@router.post("/research/{card_id}")
722@_news_research_limit
723async def research_news_item(
724 card_id: str,
725 request: Request,
726 username: Annotated[str, Depends(require_auth)],
727) -> Any:
728 """
729 Perform deeper research on a news item.
731 JSON body:
732 depth: "quick", "detailed", or "report" (default: "quick")
733 """
734 try:
735 # Body is OPTIONAL here (`or {}`), so an absent/empty body is valid
736 # and must stay valid. Only a body that will not PARSE is a 400 --
737 # previously it escaped to the generic handler as a 500.
738 try:
739 data = await request.json() or {}
740 except Exception:
741 return json_body_error("simple", "Request body must be valid JSON")
742 if not isinstance(data, dict): 742 ↛ 743line 742 didn't jump to line 743 because the condition on line 742 was never true
743 return json_body_error("simple", "Request body must be valid JSON")
744 depth = data.get("depth", "quick")
746 # Sync news-api work — threadpool, not loop.
747 return await run_db_sync(api.research_news_item, card_id, depth)
749 except NewsAPIException:
750 # api.research_news_item() currently always raises
751 # NotImplementedException (status_code=501) — let it carry that
752 # status through the registered handler instead of downgrading it
753 # to a generic 500 below.
754 raise
755 except Exception as e:
756 return JSONResponse(
757 {"error": safe_error_message(e, "researching news item")},
758 status_code=500,
759 )
762@router.get("/subscriptions/current")
763def get_current_user_subscriptions(
764 request: Request, username: Annotated[str, Depends(require_auth)]
765) -> Any:
766 """Get all subscriptions for current user."""
767 try:
768 # Get current user
769 user_id = username
771 # Ensure we have a database session for the user
772 # This will trigger register_activity
773 logger.debug(f"Getting news feed for user {user_id}")
775 # Use the API function
776 result = api.get_subscriptions(user_id)
777 if "error" in result:
778 logger.error(
779 f"Error getting subscriptions for user {user_id}: {result['error']}"
780 )
781 return JSONResponse(
782 {"error": "Failed to retrieve subscriptions"}, status_code=500
783 )
784 return result
786 except NewsAPIException:
787 # api.get_subscriptions() raises DatabaseAccessException on failure
788 # — let the NewsAPIException handler in fastapi_app.py render
789 # exc.to_dict() (error_code, details) instead of downgrading it to
790 # a generic {"error": ...} below.
791 raise
792 except Exception as e:
793 return JSONResponse(
794 {"error": safe_error_message(e, "getting subscriptions")},
795 status_code=500,
796 )
799@router.get("/subscriptions/{subscription_id}")
800def get_subscription(
801 subscription_id: str,
802 request: Request,
803 username: Annotated[str, Depends(require_auth)],
804) -> Any:
805 """Get a single subscription by ID."""
806 try:
807 # Handle null or invalid subscription IDs
808 if (
809 subscription_id == "null"
810 or subscription_id == "undefined"
811 or not subscription_id
812 ):
813 return JSONResponse(
814 {"error": "Invalid subscription ID"}, status_code=400
815 )
817 # api.get_subscription() always returns a subscription dict or
818 # raises SubscriptionNotFoundException — never a falsy value — so
819 # there is no "not found" case to handle here beyond the
820 # except NewsAPIException re-raise below.
821 return api.get_subscription(subscription_id, username=username)
823 except NewsAPIException:
824 # Carries its own status (e.g. 404 for not-found) — let the
825 # NewsAPIException handler in fastapi_app.py render it instead of
826 # downgrading it to a generic 500 below.
827 raise
828 except Exception as e:
829 return JSONResponse(
830 {"error": safe_error_message(e, "getting subscription")},
831 status_code=500,
832 )
835@router.put(
836 "/subscriptions/{subscription_id}",
837 operation_id="news_update_subscription",
838)
839async def update_subscription(
840 subscription_id: str,
841 request: Request,
842 username: Annotated[str, Depends(require_auth)],
843) -> Any:
844 """Update a subscription."""
845 try:
846 data, _body_err = await read_json_dict(
847 request, "simple", "No JSON data provided"
848 )
849 if _body_err is not None:
850 return _body_err
851 except Exception:
852 return JSONResponse({"error": "Invalid JSON data"}, status_code=400)
854 try:
855 validation_error = _validate_subscription_payload(data)
856 if validation_error is not None:
857 return validation_error
859 bad_endpoint = await _reject_custom_endpoint_async(
860 data.get("custom_endpoint")
861 )
862 if bad_endpoint is not None:
863 return bad_endpoint
865 # Prepare update data
866 update_data = {}
868 # Map fields from request to storage format
869 field_mapping = {
870 "query": "query_or_topic",
871 "name": "name",
872 "refresh_minutes": "refresh_interval_minutes",
873 "is_active": "is_active",
874 "folder_id": "folder_id",
875 "model_provider": "model_provider",
876 "model": "model",
877 "search_strategy": "search_strategy",
878 "custom_endpoint": "custom_endpoint",
879 "search_engine": "search_engine",
880 "search_iterations": "search_iterations",
881 "questions_per_iteration": "questions_per_iteration",
882 }
884 for request_field, storage_field in field_mapping.items():
885 if request_field in data:
886 update_data[storage_field] = data[request_field]
888 # Update subscription — sync DB work goes to the threadpool.
889 # api.update_subscription() always returns a {"status": "success",
890 # "subscription": {...}} dict or raises a NewsAPIException — never a
891 # dict with an "error" key — so there is no such case to handle
892 # here beyond the except NewsAPIException re-raise below.
893 return await run_db_sync(
894 api.update_subscription,
895 subscription_id,
896 update_data,
897 username=username,
898 )
900 except NewsAPIException:
901 # Carries its own status (e.g. 404 for not-found) — let the
902 # NewsAPIException handler in fastapi_app.py render it instead of
903 # downgrading it to a generic 500 below.
904 raise
905 except Exception as e:
906 return JSONResponse(
907 {"error": safe_error_message(e, "updating subscription")},
908 status_code=500,
909 )
912@router.delete("/subscriptions/{subscription_id}")
913def delete_subscription(
914 subscription_id: str,
915 request: Request,
916 username: Annotated[str, Depends(require_auth)],
917) -> Any:
918 """Delete a subscription."""
919 try:
920 # api.delete_subscription() always returns a truthy status dict or
921 # raises SubscriptionNotFoundException — never a falsy value — so
922 # there is no "not found" case to handle here beyond the
923 # except NewsAPIException re-raise below.
924 api.delete_subscription(subscription_id, username=username)
925 return {
926 "status": "success",
927 "message": f"Subscription {subscription_id} deleted",
928 }
930 except NewsAPIException:
931 # Carries its own status (e.g. 404 for not-found) — let the
932 # NewsAPIException handler in fastapi_app.py render it instead of
933 # downgrading it to a generic 500 below.
934 raise
935 except Exception as e:
936 return JSONResponse(
937 {"error": safe_error_message(e, "deleting subscription")},
938 status_code=500,
939 )
942@router.post("/subscriptions/{subscription_id}/run")
943def run_subscription_now(
944 subscription_id: str,
945 request: Request,
946 username: Annotated[str, Depends(require_auth)],
947) -> Any:
948 """Manually trigger a subscription to run now."""
949 try:
950 from ...news.core.utils import get_local_date_string
951 from ...news.subscription_runner import (
952 advance_refresh_schedule,
953 build_subscription_request_data,
954 )
955 from ...database.session_context import get_user_db_session
956 from ...database.models.news import NewsSubscription
957 from ...settings.manager import SettingsManager
958 from datetime import datetime, timezone
960 # Load the subscription from the user's database. Reading the ORM row
961 # directly (rather than the trimmed api.get_subscriptions() dict, which
962 # drops model_provider/model/search_strategy/search_engine) ensures the
963 # manual run honors the subscription's saved model config, matching the
964 # overdue sweep and the scheduler.
965 # Read the subscription + build the payload, then release the read
966 # transaction before the blocking POST below. The per-user encrypted DB
967 # uses deferred isolation, so a SELECT holds a SHARED lock on the
968 # SQLCipher file until the transaction ends — and the session is
969 # request-cached, so it is NOT closed on `with` exit. We therefore
970 # rollback() explicitly to drop the read lock before the (up to 30s)
971 # HTTP call, then reopen a short session to advance.
972 with get_user_db_session(username) as db:
973 sub = (
974 db.query(NewsSubscription)
975 .filter(NewsSubscription.id == subscription_id)
976 .first()
977 )
978 if not sub:
979 return JSONResponse(
980 {"error": "Subscription not found"}, status_code=404
981 )
983 subscription_pk = sub.id
984 # Snapshot next_refresh for the post-POST compare-and-set (below):
985 # a fast-failing run's failure handler may reset next_refresh on the
986 # worker thread, and the advance must not clobber that.
987 prev_next_refresh = sub.next_refresh
988 settings_manager = SettingsManager(db)
989 current_date = get_local_date_string(settings_manager)
991 request_data = build_subscription_request_data(
992 query_template=sub.query_or_topic,
993 current_date=current_date,
994 triggered_by="manual",
995 subscription_id=sub.id,
996 model_provider=sub.model_provider,
997 model=sub.model,
998 search_strategy=sub.search_strategy,
999 search_engine=sub.search_engine,
1000 custom_endpoint=sub.custom_endpoint,
1001 title=sub.name,
1002 )
1003 # End the read transaction so the SHARED lock is released before the
1004 # blocking POST (exiting the `with` does not — the session is
1005 # request-cached and not closed on exit).
1006 db.rollback()
1008 # Start the research IN-PROCESS (no loopback HTTP POST). The old
1009 # loopback built its target URL from request.base_url — i.e. the
1010 # client-controlled Host header — and sent the caller's own session
1011 # cookie + CSRF token to it via safe_post(allow_localhost=True,
1012 # allow_private_ips=True). A spoofed Host turned this into an SSRF /
1013 # internal-network request oracle. Calling the handler directly
1014 # removes the network hop (and the SSRF) entirely and matches main's
1015 # #4834, which replaced the same loopback with an in-process call.
1016 data = _start_research_in_process(request, request_data, username)
1018 if data.get("status") in ("success", "queued"):
1019 # Advance the schedule so a subscription that was also overdue
1020 # is not immediately re-run by the scheduler while this run is
1021 # in flight. If the run later fails, the research failure
1022 # handler resets next_refresh so the scheduler retries it.
1023 # Compare-and-set: only advance if next_refresh is unchanged
1024 # since we read it pre-spawn. A fast-failing run can reset it
1025 # (worker thread) before we get here; in that case leave the
1026 # reset in place rather than clobbering it and re-hiding the
1027 # failed subscription for a full interval.
1028 with get_user_db_session(username) as db:
1029 sub = (
1030 db.query(NewsSubscription)
1031 .filter(NewsSubscription.id == subscription_pk)
1032 .first()
1033 )
1034 if sub and sub.next_refresh == prev_next_refresh:
1035 advance_refresh_schedule(sub, datetime.now(timezone.utc))
1036 db.commit()
1037 return {
1038 "status": "success",
1039 "message": "Research started",
1040 "research_id": data.get("research_id"),
1041 "url": f"/progress/{data.get('research_id')}",
1042 }
1043 # start_research reports failures under "message"; keep "error"
1044 # as a fallback for any other shape. Propagate the real status code
1045 # (400/409/429) the in-process call surfaced, not a flat 500.
1046 return JSONResponse(
1047 {
1048 "error": data.get(
1049 "message",
1050 data.get("error", "Failed to start research"),
1051 )
1052 },
1053 status_code=data.get("_http_status", 500),
1054 )
1056 except Exception as e:
1057 return JSONResponse(
1058 {"error": safe_error_message(e, "running subscription")},
1059 status_code=500,
1060 )
1063@router.get("/subscriptions/{subscription_id}/history")
1064def get_subscription_history(
1065 subscription_id: str,
1066 request: Request,
1067 username: Annotated[str, Depends(require_auth)],
1068) -> Any:
1069 """Get research history for a subscription."""
1070 try:
1071 if not _is_valid_uuid(subscription_id):
1072 return JSONResponse(
1073 {"success": False, "error": "Invalid subscription_id"},
1074 status_code=400,
1075 )
1076 with get_user_db_session(username) as _db:
1077 settings_manager = get_settings_manager(_db, username)
1078 default_limit = settings_manager.get_setting(
1079 "news.feed.default_limit"
1080 )
1081 try:
1082 limit = int(request.query_params.get("limit", default_limit))
1083 except (TypeError, ValueError):
1084 limit = int(default_limit)
1085 limit = max(1, min(limit, NEWS_FEED_MAX_LIMIT))
1086 # api.get_subscription_history() always returns a dict with
1087 # subscription/history/total_runs, or raises SubscriptionNotFoundException
1088 # — never a dict with an "error" key — so there is no such case to
1089 # handle here beyond the except NewsAPIException re-raise below.
1090 return api.get_subscription_history(
1091 subscription_id, limit, username=username
1092 )
1093 except NewsAPIException:
1094 # Carries its own status (e.g. 404 for not-found) — let the
1095 # NewsAPIException handler in fastapi_app.py render it instead of
1096 # downgrading it to a generic 500 below.
1097 raise
1098 except Exception as e:
1099 return JSONResponse(
1100 {"error": safe_error_message(e, "getting subscription history")},
1101 status_code=500,
1102 )
1105@router.post("/preferences")
1106@_news_preferences_limit
1107async def save_preferences(
1108 request: Request, username: Annotated[str, Depends(require_auth)]
1109) -> Any:
1110 """Save user preferences for news."""
1111 try:
1112 data, _body_err = await read_json_dict(
1113 request, "simple", "No JSON data provided"
1114 )
1115 if _body_err is not None:
1116 return _body_err
1118 # Get current user
1119 user_id = username
1120 preferences = data.get("preferences", {})
1122 # Sync DB work (SQLCipher session + commit) — threadpool, not loop.
1123 return await run_db_sync(
1124 api.save_news_preferences, user_id, preferences
1125 )
1127 except NewsAPIException:
1128 # api.save_news_preferences() currently always raises
1129 # NotImplementedException (status_code=501) — let it carry that
1130 # status through the registered handler instead of downgrading it
1131 # to a generic 500 below.
1132 raise
1133 except Exception as e:
1134 return JSONResponse(
1135 {"error": safe_error_message(e, "saving preferences")},
1136 status_code=500,
1137 )
1140@router.get("/categories")
1141def get_categories(
1142 request: Request, username: Annotated[str, Depends(require_auth)]
1143) -> Any:
1144 """Get news category distribution."""
1145 from ...news.exceptions import NotImplementedException
1147 try:
1148 return api.get_news_categories()
1149 except NotImplementedException:
1150 # Surface "not yet implemented" as 501 instead of generic 500. The
1151 # message is a static string — not ``str(e)`` — so this endpoint
1152 # doesn't echo exception-derived text to the client (CWE-209,
1153 # CodeQL "Information exposure through an exception"); every other
1154 # handler in this file already goes through safe_error_message()
1155 # for the same reason. The exception is still logged server-side.
1156 logger.exception("get_news_categories: feature not implemented")
1157 return JSONResponse(
1158 {
1159 "error": "This feature is not yet implemented.",
1160 "error_code": "NOT_IMPLEMENTED",
1161 },
1162 status_code=501,
1163 )
1164 except Exception as e:
1165 return JSONResponse(
1166 {"error": safe_error_message(e, "getting categories")},
1167 status_code=500,
1168 )
1171@router.get("/scheduler/status")
1172def get_scheduler_status(
1173 request: Request, username: Annotated[str, Depends(require_auth)]
1174) -> Any:
1175 """Get activity-based scheduler status."""
1176 try:
1177 logger.info("Scheduler status endpoint called")
1178 from ...scheduler.background import get_background_job_scheduler
1180 # Get scheduler instance
1181 scheduler = get_background_job_scheduler()
1182 show_all = get_env_setting("news.scheduler.allow_api_control", False)
1183 logger.info(
1184 f"Scheduler instance obtained: is_running={scheduler.is_running}"
1185 )
1187 # Build status manually to avoid potential deadlock
1188 if show_all:
1189 active_users = (
1190 len(scheduler.user_sessions)
1191 if hasattr(scheduler, "user_sessions")
1192 else 0
1193 )
1194 else:
1195 active_users = (
1196 1
1197 if hasattr(scheduler, "user_sessions")
1198 and username in scheduler.user_sessions
1199 else 0
1200 )
1202 status = {
1203 "scheduler_available": True, # APScheduler is installed and working
1204 "is_running": scheduler.is_running,
1205 "config": scheduler.config.copy()
1206 if hasattr(scheduler, "config")
1207 else {},
1208 "active_users": active_users,
1209 "total_scheduled_jobs": 0,
1210 }
1212 # Count scheduled jobs
1213 if hasattr(scheduler, "user_sessions"): 1213 ↛ 1225line 1213 didn't jump to line 1225 because the condition on line 1213 was always true
1214 if show_all:
1215 total_jobs = sum(
1216 len(sess.get("scheduled_jobs", set()))
1217 for sess in scheduler.user_sessions.values()
1218 )
1219 else:
1220 user_session = scheduler.user_sessions.get(username, {})
1221 total_jobs = len(user_session.get("scheduled_jobs", set()))
1222 status["total_scheduled_jobs"] = total_jobs
1224 # Also count actual APScheduler jobs
1225 if hasattr(scheduler, "scheduler") and scheduler.scheduler: 1225 ↛ 1251line 1225 didn't jump to line 1251 because the condition on line 1225 was always true
1226 try:
1227 apscheduler_jobs = scheduler.scheduler.get_jobs()
1228 if not show_all:
1229 apscheduler_jobs = [
1230 j
1231 for j in apscheduler_jobs
1232 if _is_job_owned_by_user(j, username, scheduler)
1233 ]
1234 status["apscheduler_job_count"] = len(apscheduler_jobs)
1235 status["apscheduler_jobs"] = [
1236 {
1237 "id": job.id,
1238 "name": job.name,
1239 "next_run": job.next_run_time.isoformat()
1240 if job.next_run_time
1241 else None,
1242 }
1243 for job in apscheduler_jobs[
1244 :10
1245 ] # Limit to first 10 for display
1246 ]
1247 except Exception:
1248 logger.exception("Error getting APScheduler jobs")
1249 status["apscheduler_job_count"] = 0
1251 logger.info(f"Status built: {list(status.keys())}")
1253 # Add scheduled_jobs field that JS expects
1254 status["scheduled_jobs"] = status.get("total_scheduled_jobs", 0)
1256 logger.info(
1257 f"Returning status: is_running={status.get('is_running')}, active_users={status.get('active_users')}"
1258 )
1259 return status
1261 except Exception as e:
1262 return JSONResponse(
1263 {"error": safe_error_message(e, "getting scheduler status")},
1264 status_code=500,
1265 )
1268@router.post("/scheduler/start")
1269def start_scheduler(
1270 request: Request,
1271 username: Annotated[str, Depends(require_auth)],
1272 _gate: Annotated[None, Depends(require_scheduler_control)],
1273) -> Any:
1274 """Start the subscription scheduler."""
1275 try:
1276 from ...scheduler.background import get_background_job_scheduler
1278 # Get scheduler instance
1279 scheduler = get_background_job_scheduler()
1281 if scheduler.is_running:
1282 return {"message": "Scheduler is already running"}
1284 # Start the scheduler
1285 scheduler.start()
1287 # Update app reference
1288 # Scheduler is a singleton — no need to store on app
1290 logger.info("News scheduler started via API")
1291 return {
1292 "status": "success",
1293 "message": "Scheduler started",
1294 "active_users": len(scheduler.user_sessions),
1295 }
1297 except Exception as e:
1298 return JSONResponse(
1299 {"error": safe_error_message(e, "starting scheduler")},
1300 status_code=500,
1301 )
1304@router.post("/scheduler/stop")
1305def stop_scheduler(
1306 request: Request,
1307 username: Annotated[str, Depends(require_auth)],
1308 _gate: Annotated[None, Depends(require_scheduler_control)],
1309) -> Any:
1310 """Stop the subscription scheduler."""
1311 try:
1312 from ...scheduler.background import get_background_job_scheduler
1314 if get_background_job_scheduler():
1315 scheduler = get_background_job_scheduler()
1316 if scheduler.is_running:
1317 scheduler.stop()
1318 logger.info("News scheduler stopped via API")
1319 return {"status": "success", "message": "Scheduler stopped"}
1321 return {"message": "Scheduler is not running"}
1322 return JSONResponse(
1323 {"message": "No scheduler instance found"}, status_code=404
1324 )
1326 except Exception as e:
1327 return JSONResponse(
1328 {"error": safe_error_message(e, "stopping scheduler")},
1329 status_code=500,
1330 )
1333@router.post("/scheduler/check-now")
1334def check_subscriptions_now(
1335 request: Request,
1336 username: Annotated[str, Depends(require_auth)],
1337 _gate: Annotated[None, Depends(require_scheduler_control)],
1338) -> Any:
1339 """Manually trigger subscription checking."""
1340 try:
1341 from ...scheduler.background import get_background_job_scheduler
1343 if not get_background_job_scheduler():
1344 return JSONResponse(
1345 {"error": "Scheduler not initialized"}, status_code=503
1346 )
1348 scheduler = get_background_job_scheduler()
1349 if not scheduler.is_running:
1350 return JSONResponse(
1351 {"error": "Scheduler is not running"}, status_code=503
1352 )
1354 # Run the check subscriptions task immediately
1355 scheduler_instance = get_background_job_scheduler()
1357 # Get count of due subscriptions
1358 from ...database.models import NewsSubscription as BaseSubscription
1359 from datetime import datetime, timedelta, timezone
1361 with get_user_db_session(username) as session:
1362 now = datetime.now(timezone.utc)
1363 count = (
1364 session.query(BaseSubscription)
1365 .filter(BaseSubscription.due_filter(now))
1366 .count()
1367 )
1369 # Trigger the check asynchronously via APScheduler, matching main:
1370 # the per-user job id + replace_existing dedups double-clicks, and
1371 # _wrap_job supplies the worker-side context handling.
1372 scheduler_instance.scheduler.add_job(
1373 func=scheduler_instance._wrap_job(
1374 scheduler_instance._check_user_overdue_subscriptions,
1375 username=username,
1376 ),
1377 args=[username],
1378 trigger="date",
1379 run_date=datetime.now(timezone.utc) + timedelta(seconds=1),
1380 id=f"manual_check_{username}",
1381 replace_existing=True,
1382 )
1384 return {
1385 "status": "success",
1386 "message": f"Checking {count} due subscriptions",
1387 "count": count,
1388 }
1390 except Exception as e:
1391 return JSONResponse(
1392 {"error": safe_error_message(e, "checking subscriptions")},
1393 status_code=500,
1394 )
1397@router.post("/scheduler/cleanup-now")
1398def trigger_cleanup(
1399 request: Request,
1400 username: Annotated[str, Depends(require_auth)],
1401 _gate: Annotated[None, Depends(require_scheduler_control)],
1402) -> Any:
1403 """Manually trigger cleanup job."""
1404 try:
1405 from ...scheduler.background import get_background_job_scheduler
1406 from datetime import datetime, UTC, timedelta
1408 scheduler = get_background_job_scheduler()
1410 if not scheduler.is_running:
1411 return JSONResponse(
1412 {"error": "Scheduler is not running"}, status_code=400
1413 )
1415 # Schedule cleanup to run in 1 second
1416 # replace_existing=True and _wrap_job both matter here, and both were
1417 # lost in the port. The job id is fixed and the run_date is 1s out, so
1418 # a second POST inside that window raises ConflictingIdError, which the
1419 # blanket except below turns into a 500 -- a double-click on the admin
1420 # button is enough. main replaced the pending job and returned 200.
1421 # _wrap_job is inert for a system job (username=None -> nullcontext)
1422 # but this was the only add_job in the tree without it; the sibling
1423 # check_subscriptions_now above keeps both and says why.
1424 scheduler.scheduler.add_job(
1425 scheduler._wrap_job(scheduler._run_cleanup_with_tracking),
1426 "date",
1427 run_date=datetime.now(UTC) + timedelta(seconds=1),
1428 id="manual_cleanup_trigger",
1429 replace_existing=True,
1430 )
1432 return {
1433 "status": "triggered",
1434 "message": "Cleanup job will run within seconds",
1435 }
1437 except Exception as e:
1438 return JSONResponse(
1439 {"error": safe_error_message(e, "triggering cleanup")},
1440 status_code=500,
1441 )
1444@router.get("/scheduler/users")
1445def get_active_users(
1446 request: Request, username: Annotated[str, Depends(require_auth)]
1447) -> Any:
1448 """Get summary of active user sessions."""
1449 try:
1450 from ...scheduler.background import get_background_job_scheduler
1452 scheduler = get_background_job_scheduler()
1453 users_summary = scheduler.get_user_sessions_summary()
1455 show_all = get_env_setting("news.scheduler.allow_api_control", False)
1456 if not show_all:
1457 users_summary = [
1458 u for u in users_summary if u.get("user_id") == username
1459 ]
1461 return {"active_users": len(users_summary), "users": users_summary}
1463 except Exception as e:
1464 return JSONResponse(
1465 {"error": safe_error_message(e, "getting active users")},
1466 status_code=500,
1467 )
1470@router.get("/scheduler/stats")
1471def scheduler_stats(
1472 request: Request, username: Annotated[str, Depends(require_auth)]
1473) -> Any:
1474 """Get scheduler statistics and state."""
1475 try:
1476 from ...scheduler.background import get_background_job_scheduler
1478 scheduler = get_background_job_scheduler()
1480 # Debug info
1481 debug_info = {
1482 "current_user": username,
1483 "scheduler_running": scheduler.is_running,
1484 "user_sessions": {},
1485 "apscheduler_jobs": [],
1486 }
1488 show_all = get_env_setting("news.scheduler.allow_api_control", False)
1490 # Get user session info
1491 if hasattr(scheduler, "user_sessions"): 1491 ↛ 1510line 1491 didn't jump to line 1510 because the condition on line 1491 was always true
1492 for user, session_info in scheduler.user_sessions.items():
1493 if not show_all and user != username:
1494 continue
1495 debug_info["user_sessions"][user] = {
1496 "has_password": bool(
1497 scheduler._credential_store.retrieve(user)
1498 ),
1499 "last_activity": session_info.get(
1500 "last_activity"
1501 ).isoformat()
1502 if session_info.get("last_activity")
1503 else None,
1504 "scheduled_jobs_count": len(
1505 session_info.get("scheduled_jobs", set())
1506 ),
1507 }
1509 # Get APScheduler jobs
1510 if hasattr(scheduler, "scheduler") and scheduler.scheduler: 1510 ↛ 1530line 1510 didn't jump to line 1530 because the condition on line 1510 was always true
1511 jobs = scheduler.scheduler.get_jobs()
1512 if not show_all:
1513 jobs = [
1514 j
1515 for j in jobs
1516 if _is_job_owned_by_user(j, username, scheduler)
1517 ]
1518 debug_info["apscheduler_jobs"] = [
1519 {
1520 "id": job.id,
1521 "name": job.name,
1522 "next_run": job.next_run_time.isoformat()
1523 if job.next_run_time
1524 else None,
1525 "trigger": str(job.trigger),
1526 }
1527 for job in jobs
1528 ]
1530 return debug_info
1532 except Exception as e:
1533 return JSONResponse(
1534 {"error": safe_error_message(e, "getting scheduler stats")},
1535 status_code=500,
1536 )
1539@router.post("/check-overdue")
1540def check_overdue_subscriptions(
1541 request: Request, username: Annotated[str, Depends(require_auth)]
1542):
1543 """Check and run all overdue subscriptions for the current user."""
1544 try:
1545 from ...news.subscription_runner import (
1546 advance_refresh_schedule,
1547 build_subscription_request_data,
1548 )
1549 from ...database.session_context import get_user_db_session
1550 from ...database.models.news import NewsSubscription
1551 from datetime import datetime, UTC
1553 # Get overdue subscriptions
1554 overdue_count = 0
1555 results = []
1556 with get_user_db_session(username) as db:
1557 now = datetime.now(UTC)
1558 overdue_subs = (
1559 db.query(NewsSubscription)
1560 .filter(NewsSubscription.due_filter(now))
1561 .all()
1562 )
1564 logger.info(
1565 f"Found {len(overdue_subs)} overdue subscriptions for {username}"
1566 )
1568 # Get timezone-aware current date using settings
1569 from ...news.core.utils import get_local_date_string
1570 from ...settings.manager import SettingsManager
1572 settings_manager = SettingsManager(db)
1573 current_date = get_local_date_string(settings_manager)
1575 for sub in overdue_subs:
1576 # Capture identity up front as plain strings. This loop shares
1577 # one DB session across every start_research call, and
1578 # start_research's error path does not roll back — so a failed
1579 # run can leave the session in a PendingRollbackError state.
1580 # Reading sub.* again in an error branch would then raise (the
1581 # row was expired by an earlier commit), collapsing the whole
1582 # sweep. Snapshotting here keeps the error branches session-free.
1583 sub_id = str(sub.id)
1584 sub_label = sub.name or sub.query_or_topic[:50]
1585 try:
1586 # Run the subscription using the same pattern as run_subscription_now
1587 logger.info(
1588 f"Running overdue subscription: {sub_label[:30]}"
1589 )
1591 # Snapshot for the post-run compare-and-set (see below).
1592 prev_next_refresh = sub.next_refresh
1594 request_data = build_subscription_request_data(
1595 query_template=sub.query_or_topic,
1596 current_date=current_date,
1597 triggered_by="overdue_check",
1598 subscription_id=sub.id,
1599 model_provider=sub.model_provider,
1600 model=sub.model,
1601 search_strategy=sub.search_strategy,
1602 search_engine=sub.search_engine,
1603 custom_endpoint=sub.custom_endpoint,
1604 title=sub.name,
1605 )
1607 # Start the research IN-PROCESS — see the run-now handler
1608 # above. The old loopback built its URL from the
1609 # client-controlled Host header and forwarded the caller's
1610 # cookie/CSRF via safe_post(allow_localhost/private=True):
1611 # an SSRF whose raw response body was reflected back to the
1612 # client in the `error` field below (an internal-response
1613 # oracle). The in-process call removes both.
1614 result = _start_research_in_process(
1615 request, request_data, username
1616 )
1618 if result.get("status") in ("success", "queued"):
1619 overdue_count += 1
1621 # Update subscription's last/next refresh times.
1622 # Compare-and-set: re-read and skip the advance if a
1623 # fast-failing run already reset next_refresh (worker
1624 # thread), so we don't clobber the reset and re-hide it.
1625 db.refresh(sub)
1626 if sub.next_refresh == prev_next_refresh:
1627 advance_refresh_schedule(sub, datetime.now(UTC))
1628 db.commit()
1630 results.append(
1631 {
1632 "id": sub_id,
1633 "name": sub_label,
1634 "research_id": result.get("research_id"),
1635 }
1636 )
1637 else:
1638 # start_research failed and may have left the shared
1639 # session dirty; reset it so the next subscription runs.
1640 db.rollback()
1641 results.append(
1642 {
1643 "id": sub_id,
1644 "name": sub_label,
1645 # start_research reports failures under
1646 # "message"; keep "error" as a fallback for
1647 # any other shape.
1648 "error": result.get(
1649 "message",
1650 result.get(
1651 "error", "Failed to start research"
1652 ),
1653 ),
1654 }
1655 )
1656 except Exception as e:
1657 # Recover the shared session (a failed start_research commit
1658 # can leave it in a PendingRollbackError state) so the
1659 # remaining overdue subscriptions in the sweep still run.
1660 db.rollback()
1661 logger.exception(f"Error running subscription {sub_id}")
1662 results.append(
1663 {
1664 "id": sub_id,
1665 "name": sub_label,
1666 "error": safe_error_message(
1667 e, "running subscription"
1668 ),
1669 }
1670 )
1672 return {
1673 "status": "success",
1674 "overdue_found": len(overdue_subs),
1675 "started": overdue_count,
1676 "results": results,
1677 }
1679 except Exception as e:
1680 return JSONResponse(
1681 {"error": safe_error_message(e, "checking overdue subscriptions")},
1682 status_code=500,
1683 )
1686# Folder and subscription management routes
1687@router.get("/subscription/folders")
1688def get_folders(
1689 request: Request, username: Annotated[str, Depends(require_auth)]
1690):
1691 """Get all folders for the current user"""
1692 try:
1693 user_id = username
1695 with get_user_db_session(username) as session:
1696 manager = FolderManager(session)
1697 folders = manager.get_user_folders(user_id)
1699 return [folder.to_dict() for folder in folders]
1701 except Exception as e:
1702 return JSONResponse(
1703 {"error": safe_error_message(e, "getting folders")},
1704 status_code=500,
1705 )
1708@router.post("/subscription/folders")
1709async def create_folder(
1710 request: Request, username: Annotated[str, Depends(require_auth)]
1711):
1712 """Create a new folder"""
1713 data, _body_err = await read_json_dict(
1714 request, "simple", "Request body must be valid JSON"
1715 )
1716 if _body_err is not None:
1717 return _body_err
1719 def _impl():
1720 try:
1721 if not data.get("name"):
1722 return JSONResponse(
1723 {"error": "Folder name is required"}, status_code=400
1724 )
1726 with get_user_db_session(username) as session:
1727 manager = FolderManager(session)
1729 # Check if folder already exists
1730 existing = (
1731 session.query(SubscriptionFolder)
1732 .filter_by(name=data["name"])
1733 .first()
1734 )
1735 if existing:
1736 return JSONResponse(
1737 {"error": "Folder already exists"}, status_code=409
1738 )
1740 folder = manager.create_folder(
1741 name=data["name"],
1742 description=data.get("description"),
1743 )
1745 return JSONResponse(folder.to_dict(), status_code=201)
1747 except Exception as e:
1748 return JSONResponse(
1749 {"error": safe_error_message(e, "creating folder")},
1750 status_code=500,
1751 )
1753 return await run_db_sync(_impl)
1756@router.put("/subscription/folders/{folder_id}")
1757async def update_folder(
1758 request: Request, folder_id, username: Annotated[str, Depends(require_auth)]
1759):
1760 """Update a folder"""
1761 data, _body_err = await read_json_dict(
1762 request, "simple", "Request body must be valid JSON"
1763 )
1764 if _body_err is not None:
1765 return _body_err
1767 def _impl():
1768 try:
1769 with get_user_db_session(username) as session:
1770 manager = FolderManager(session)
1771 folder = manager.update_folder(folder_id, **data)
1773 if not folder:
1774 return JSONResponse(
1775 {"error": "Folder not found"}, status_code=404
1776 )
1778 return folder.to_dict()
1780 except Exception as e:
1781 return JSONResponse(
1782 {"error": safe_error_message(e, "updating folder")},
1783 status_code=500,
1784 )
1786 return await run_db_sync(_impl)
1789@router.delete("/subscription/folders/{folder_id}")
1790def delete_folder(
1791 request: Request, folder_id, username: Annotated[str, Depends(require_auth)]
1792):
1793 """Delete a folder"""
1794 try:
1795 move_to = request.query_params.get("move_to")
1797 with get_user_db_session(username) as session:
1798 manager = FolderManager(session)
1799 success = manager.delete_folder(folder_id, move_to)
1801 if not success:
1802 return JSONResponse(
1803 {"error": "Folder not found"}, status_code=404
1804 )
1806 return {"status": "deleted"}
1808 except Exception as e:
1809 return JSONResponse(
1810 {"error": safe_error_message(e, "deleting folder")},
1811 status_code=500,
1812 )
1815@router.get("/subscription/subscriptions/organized")
1816def get_subscriptions_organized(
1817 request: Request, username: Annotated[str, Depends(require_auth)]
1818):
1819 """Get subscriptions organized by folder"""
1820 try:
1821 user_id = username
1823 with get_user_db_session(username) as session:
1824 manager = FolderManager(session)
1825 organized = manager.get_subscriptions_by_folder(user_id)
1827 # get_subscriptions_by_folder already returns JSON-friendly dicts
1828 # ({"folders": [{"folder": {...}, "subscriptions": [...]}, ...],
1829 # "uncategorized": [...]}). The previous code called .to_dict() on
1830 # those plain dicts (AttributeError -> HTTP 500). Flatten into the
1831 # {folder_name: [subscription, ...]} map the subscriptions UI
1832 # consumes (Object.values(...) for the "all" view, keyed lookup per
1833 # folder), with ungrouped subscriptions under "uncategorized".
1834 result = {}
1835 for entry in organized.get("folders", []):
1836 folder_name = entry["folder"].get("name") or entry[
1837 "folder"
1838 ].get("id")
1839 result[folder_name] = entry["subscriptions"]
1840 # Merge (don't overwrite) so a user folder literally named
1841 # "uncategorized" doesn't have its subscriptions dropped by the
1842 # ungrouped bucket. In the normal case this just sets the key.
1843 result.setdefault("uncategorized", []).extend(
1844 organized.get("uncategorized", [])
1845 )
1847 return result
1849 except Exception as e:
1850 return JSONResponse(
1851 {"error": safe_error_message(e, "getting organized subscriptions")},
1852 status_code=500,
1853 )
1856@router.put("/subscription/subscriptions/{subscription_id}")
1857async def update_subscription_folder(
1858 request: Request,
1859 subscription_id,
1860 username: Annotated[str, Depends(require_auth)],
1861):
1862 """Update a subscription (mainly for folder assignment)"""
1863 data, _body_err = await read_json_dict(
1864 request, "simple", "Request body must be valid JSON"
1865 )
1866 if _body_err is not None:
1867 return _body_err
1868 validation_error = _validate_subscription_payload(
1869 data,
1870 query_field="query_or_topic",
1871 refresh_field="refresh_interval_minutes",
1872 )
1873 if validation_error is not None:
1874 return validation_error
1875 # ``_update_subscription_folder_sync``'s blind setattr loop writes any
1876 # matching column, so ``custom_endpoint`` reaches storage here too -- a
1877 # third write path alongside create_subscription / update_subscription.
1878 # Validate it at the boundary, before the sync helper's unredacted log
1879 # line, so this route cannot persist an SSRF endpoint its siblings
1880 # reject. Ported from #5603, which closed the same gap on the Flask
1881 # blueprint this router replaced.
1882 if "custom_endpoint" in data:
1883 bad_endpoint = await _reject_custom_endpoint_async(
1884 data.get("custom_endpoint")
1885 )
1886 if bad_endpoint is not None:
1887 return bad_endpoint
1888 return await run_db_sync(
1889 _update_subscription_folder_sync, data, subscription_id, username
1890 )
1893def _update_subscription_folder_sync(data, subscription_id, username):
1894 try:
1895 validation_error = _validate_subscription_payload(
1896 data,
1897 query_field="query_or_topic",
1898 refresh_field="refresh_interval_minutes",
1899 )
1900 if validation_error is not None:
1901 return validation_error
1903 logger.info(
1904 f"Updating subscription {subscription_id} with data: {data}"
1905 )
1907 with get_user_db_session(username) as session:
1908 # Manually handle the update to ensure next_refresh is recalculated
1909 from ...database.models.news import (
1910 NewsSubscription as BaseSubscription,
1911 )
1912 from datetime import datetime, timedelta, timezone
1914 sub = (
1915 session.query(BaseSubscription)
1916 .filter_by(id=subscription_id)
1917 .first()
1918 )
1919 if not sub:
1920 return JSONResponse(
1921 {"error": "Subscription not found"}, status_code=404
1922 )
1924 # `status` is the single source of truth for active/paused (the
1925 # scheduler keys on it, not the legacy is_active column). Translate
1926 # an is_active toggle into status and keep both out of the blind
1927 # setattr loop below -- otherwise a body like {"is_active": false}
1928 # would flip only the legacy column while status stayed "active"
1929 # and the scheduler would keep running the subscription. Mirrors
1930 # api.update_subscription's translation.
1931 if "is_active" in data:
1932 sub.status = "active" if data["is_active"] else "paused"
1933 if "status" in data:
1934 sub.status = data["status"]
1936 # Update remaining fields
1937 for key, value in data.items():
1938 if hasattr(sub, key) and key not in [
1939 "id",
1940 "user_id",
1941 "created_at",
1942 "is_active",
1943 "status",
1944 ]:
1945 setattr(sub, key, value)
1947 # Recalculate next_refresh if refresh_interval_minutes changed
1948 if "refresh_interval_minutes" in data:
1949 new_minutes = data["refresh_interval_minutes"]
1950 if sub.last_refresh:
1951 sub.next_refresh = sub.last_refresh + timedelta(
1952 minutes=new_minutes
1953 )
1954 else:
1955 sub.next_refresh = datetime.now(timezone.utc) + timedelta(
1956 minutes=new_minutes
1957 )
1958 logger.info(f"Recalculated next_refresh: {sub.next_refresh}")
1960 sub.updated_at = datetime.now(timezone.utc)
1961 session.commit()
1963 # NewsSubscription has no to_dict(); serialize the fields the UI
1964 # needs explicitly. is_active is derived from status (the source of
1965 # truth) so the response stays consistent with the toggle above.
1966 result = {
1967 "id": sub.id,
1968 "name": sub.name,
1969 "status": sub.status,
1970 "is_active": sub.status == "active",
1971 "folder_id": sub.folder_id,
1972 "refresh_interval_minutes": sub.refresh_interval_minutes,
1973 "next_refresh": sub.next_refresh.isoformat()
1974 if sub.next_refresh
1975 else None,
1976 "last_refresh": sub.last_refresh.isoformat()
1977 if sub.last_refresh
1978 else None,
1979 }
1980 logger.info(
1981 f"Updated subscription result: refresh_interval_minutes={result.get('refresh_interval_minutes')}, next_refresh={result.get('next_refresh')}"
1982 )
1983 return result
1985 except Exception as e:
1986 return JSONResponse(
1987 {"error": safe_error_message(e, "updating subscription")},
1988 status_code=500,
1989 )
1992@router.get("/subscription/stats")
1993def get_subscription_stats(
1994 request: Request, username: Annotated[str, Depends(require_auth)]
1995):
1996 """Get subscription statistics"""
1997 try:
1998 user_id = username
2000 with get_user_db_session(username) as session:
2001 manager = FolderManager(session)
2002 return manager.get_subscription_stats(user_id)
2004 except Exception as e:
2005 return JSONResponse(
2006 {"error": safe_error_message(e, "getting stats")},
2007 status_code=500,
2008 )
2011@router.get("/search-history")
2012def get_search_history(
2013 request: Request, username: Annotated[str, Depends(require_auth)]
2014):
2015 """Get search history for current user."""
2016 try:
2017 # Get search history from user's encrypted database
2018 from ...database.session_context import get_user_db_session
2019 from ...database.models import UserNewsSearchHistory
2021 password = None
2023 with get_user_db_session(username, password) as db_session:
2024 history = (
2025 db_session.query(UserNewsSearchHistory)
2026 .order_by(UserNewsSearchHistory.created_at.desc())
2027 .limit(20)
2028 .all()
2029 )
2031 return {"search_history": [item.to_dict() for item in history]}
2033 except Exception as e:
2034 return JSONResponse(
2035 {"error": safe_error_message(e, "getting search history")},
2036 status_code=500,
2037 )
2040@router.post("/search-history")
2041async def add_search_history(
2042 request: Request, username: Annotated[str, Depends(require_auth)]
2043):
2044 """Add a search to the history."""
2045 data = await request.json()
2047 def _impl():
2048 try:
2049 # `data` may be any JSON value here (a bare list, string, or
2050 # number is valid JSON but not a dict) — check the type before
2051 # calling dict-only methods on it, else a truthy non-dict body
2052 # raises AttributeError and lands in the `except Exception`
2053 # below as a 500 instead of the 400 a malformed request should
2054 # get.
2055 logger.info(
2056 "add_search_history received data keys: {}",
2057 list(data.keys()) if isinstance(data, dict) else "None",
2058 )
2059 if not isinstance(data, dict) or not data.get("query"):
2060 logger.warning("Invalid search history data: missing query")
2061 return JSONResponse(
2062 {"error": "query is required"}, status_code=400
2063 )
2065 # Add to user's encrypted database
2066 from ...database.session_context import get_user_db_session
2067 from ...database.models import UserNewsSearchHistory
2069 password = None
2071 with get_user_db_session(username, password) as db_session:
2072 search_history = UserNewsSearchHistory(
2073 query=data["query"],
2074 search_type=data.get("type", "filter"),
2075 result_count=data.get("resultCount", 0),
2076 )
2077 db_session.add(search_history)
2078 db_session.commit()
2080 return {"status": "success", "id": search_history.id}
2082 except Exception as e:
2083 logger.exception("Error adding search history")
2084 return JSONResponse(
2085 {"error": safe_error_message(e, "adding search history")},
2086 status_code=500,
2087 )
2089 return await run_db_sync(_impl)
2092@router.delete("/search-history")
2093def clear_search_history(
2094 request: Request, username: Annotated[str, Depends(require_auth)]
2095):
2096 """Clear all search history for current user."""
2097 try:
2098 # Clear from user's encrypted database
2099 from ...database.session_context import get_user_db_session
2100 from ...database.models import UserNewsSearchHistory
2102 password = None
2104 with get_user_db_session(username, password) as db_session:
2105 db_session.query(UserNewsSearchHistory).delete()
2106 db_session.commit()
2108 return {"status": "success"}
2110 except Exception as e:
2111 return JSONResponse(
2112 {"error": safe_error_message(e, "clearing search history")},
2113 status_code=500,
2114 )