Coverage for src/local_deep_research/news/web.py: 100%

1""" 

2Flask blueprint for news system web routes. 

3""" 

4 

5from flask import Blueprint, jsonify, render_template 

6from loguru import logger 

7from sqlalchemy.orm import Session 

8from typing import Optional 

9 

10from ..constants import get_available_strategies 

11from ..web.auth.decorators import login_required 

12from . import api 

13 

14# get_db_setting not available in merged codebase - will use defaults 

15 

16 

17def create_news_blueprint(): 

18 """ 

19 Create Flask blueprint for news routes. 

20 

21 Returns: 

22 Flask Blueprint instance with both page routes and API routes 

23 """ 

24 bp = Blueprint("news", __name__) 

25 

26 # Import the Flask API blueprint 

27 from .flask_api import news_api_bp 

28 

29 # Register the API blueprint as sub-blueprint 

30 bp.register_blueprint(news_api_bp) # type: ignore[attr-defined,unused-ignore] 

31 

32 # Page routes 

33 @bp.route("/") 

34 @login_required 

35 def news_page(): 

36 """Render the main news page.""" 

37 return render_template( 

38 "pages/news.html", 

39 strategies=get_available_strategies(), 

40 ) 

41 

42 @bp.route("/subscriptions") 

43 @login_required 

44 def subscriptions_page(): 

45 """Render the subscriptions management page.""" 

46 return render_template("pages/subscriptions.html") 

47 

48 @bp.route("/subscriptions/new") 

49 @login_required 

50 def new_subscription_page(): 

51 """Render the create subscription page.""" 

52 from flask import session 

53 

54 # Get username from session 

55 username = session.get("username", "anonymous") 

56 

57 # Try to get settings from database, fall back to defaults 

58 default_settings = { 

59 "iterations": 3, 

60 "questions_per_iteration": 5, 

61 "search_engine": "searxng", 

62 "model_provider": "ollama", 

63 "model": "", 

64 "search_strategy": "source-based", 

65 } 

66 

67 # Only try to get settings if user is logged in 

68 if username != "anonymous": 

69 # Load user settings using the extracted function 

70 from local_deep_research.database.session_context import ( 

71 get_user_db_session, 

72 ) 

73 

74 with get_user_db_session(username) as db_session: 

75 load_user_settings(default_settings, db_session, username) 

76 

77 return render_template( 

78 "pages/news-subscription-form.html", 

79 subscription=None, 

80 default_settings=default_settings, 

81 strategies=get_available_strategies(), 

82 ) 

83 

84 @bp.route("/subscriptions/<subscription_id>/edit") 

85 @login_required 

86 def edit_subscription_page(subscription_id): 

87 """Render the edit subscription page.""" 

88 from flask import session 

89 

90 # Get username from session 

91 username = session.get("username", "anonymous") 

92 

93 # Load subscription data 

94 subscription = None 

95 default_settings = { 

96 "iterations": 3, 

97 "questions_per_iteration": 5, 

98 "search_engine": "searxng", 

99 "model_provider": "ollama", 

100 "model": "", 

101 "search_strategy": "source-based", 

102 } 

103 

104 try: 

105 # Load the subscription using the API 

106 subscription = api.get_subscription(subscription_id) 

107 logger.info( 

108 f"Loaded subscription {subscription_id}: {subscription}" 

109 ) 

110 

111 if not subscription: 

112 logger.warning(f"Subscription {subscription_id} not found") 

113 # Could redirect to 404 or subscriptions page 

114 return render_template( 

115 "pages/news-subscription-form.html", 

116 subscription=None, 

117 error="Subscription not found", 

118 default_settings=default_settings, 

119 strategies=get_available_strategies(), 

120 ) 

121 

122 # Load user's default settings if logged in 

123 if username != "anonymous": 

124 # Load user settings using the extracted function 

125 from local_deep_research.database.session_context import ( 

126 get_user_db_session, 

127 ) 

128 

129 with get_user_db_session(username) as db_session: 

130 load_user_settings(default_settings, db_session, username) 

131 

132 except Exception: 

133 logger.exception(f"Error loading subscription {subscription_id}") 

134 return render_template( 

135 "pages/news-subscription-form.html", 

136 subscription=None, 

137 error="Error loading subscription", 

138 default_settings=default_settings, 

139 strategies=get_available_strategies(), 

140 ) 

141 

142 return render_template( 

143 "pages/news-subscription-form.html", 

144 subscription=subscription, 

145 default_settings=default_settings, 

146 strategies=get_available_strategies(), 

147 ) 

148 

149 # Health check 

150 @bp.route("/health") 

151 def health_check(): 

152 """Check if news system is healthy.""" 

153 try: 

154 # Check if database is accessible 

155 from .core.storage_manager import StorageManager 

156 

157 storage = StorageManager() 

158 

159 # Try a simple query 

160 storage.get_user_feed("health_check", limit=1) 

161 

162 return jsonify( 

163 { 

164 "status": "healthy", 

165 "enabled": True, # Default: get_db_setting("news.enabled", True) 

166 "database": "connected", 

167 } 

168 ) 

169 except Exception: 

170 logger.exception("Health check failed") 

171 return jsonify( 

172 { 

173 "status": "unhealthy", 

174 "error": "An internal error has occurred.", 

175 } 

176 ), 500 

177 

178 return bp 

179 

180 

181def load_user_settings( 

182 default_settings, db_session: Optional[Session] = None, username=None 

183): 

184 """ 

185 Load user settings and update default_settings dictionary. 

186 Extracted to avoid code duplication as suggested by djpetti. 

187 

188 Args: 

189 default_settings: Dictionary to update with user settings 

190 db_session: Database session for accessing settings 

191 username: Username for settings context 

192 """ 

193 if not db_session: 

194 logger.warning("No database session provided, using defaults") 

195 return 

196 

197 try: 

198 from ..utilities.db_utils import get_settings_manager 

199 

200 settings_manager = get_settings_manager(db_session, username) 

201 

202 default_settings.update( 

203 { 

204 "iterations": settings_manager.get_setting( 

205 "search.iterations", 3 

206 ), 

207 "questions_per_iteration": settings_manager.get_setting( 

208 "search.questions_per_iteration", 5 

209 ), 

210 "search_engine": settings_manager.get_setting( 

211 "search.tool", "searxng" 

212 ), 

213 "model_provider": settings_manager.get_setting( 

214 "llm.provider", "ollama" 

215 ), 

216 "model": settings_manager.get_setting("llm.model", ""), 

217 "search_strategy": settings_manager.get_setting( 

218 "search.search_strategy", "source-based" 

219 ), 

220 "custom_endpoint": settings_manager.get_setting( 

221 "llm.openai_endpoint.url", "" 

222 ), 

223 } 

224 ) 

225 except Exception: 

226 logger.warning("Could not load user settings") 

227 # Use defaults