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