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

1""" 

2Flask blueprint for news system web routes. 

3""" 

4 

5from flask import Blueprint, jsonify, render_template 

6from loguru import logger 

7from . import api 

8 

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

10 

11 

12def create_news_blueprint(): 

13 """ 

14 Create Flask blueprint for news routes. 

15 

16 Returns: 

17 Flask Blueprint instance with both page routes and API routes 

18 """ 

19 bp = Blueprint("news", __name__) 

20 

21 # Import the Flask API blueprint 

22 from .flask_api import news_api_bp 

23 

24 # Register the API blueprint as sub-blueprint 

25 bp.register_blueprint(news_api_bp) 

26 

27 # Page routes 

28 @bp.route("/") 

29 def news_page(): 

30 """Render the main news page.""" 

31 # Get available strategies for dropdown 

32 strategies = [ 

33 "topic_based", 

34 "news_aggregation", 

35 "source_based", 

36 "focused_iteration", 

37 ] 

38 

39 default_strategy = "topic_based" # Default strategy 

40 

41 return render_template( 

42 "pages/news.html", 

43 strategies=strategies, 

44 default_strategy=default_strategy, 

45 ) 

46 

47 @bp.route("/insights") 

48 def insights_page(): 

49 """Render the news insights/transparency page.""" 

50 return render_template("pages/news_insights.html") 

51 

52 @bp.route("/preferences") 

53 def preferences_page(): 

54 """Render the user preferences page.""" 

55 return render_template("pages/news_preferences.html") 

56 

57 @bp.route("/subscriptions") 

58 def subscriptions_page(): 

59 """Render the subscriptions management page.""" 

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

61 

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

63 def new_subscription_page(): 

64 """Render the create subscription page.""" 

65 from flask import session 

66 

67 # Get username from session 

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

69 

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

71 default_settings = { 

72 "iterations": 3, 

73 "questions_per_iteration": 5, 

74 "search_engine": "auto", 

75 "model_provider": "OLLAMA", 

76 "model": "", 

77 "search_strategy": "source-based", 

78 } 

79 

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

81 if username != "anonymous": 

82 # Load user settings using the extracted function 

83 from local_deep_research.database.session_context import ( 

84 get_user_db_session, 

85 ) 

86 

87 with get_user_db_session(username) as db_session: 

88 load_user_settings(default_settings, db_session, username) 

89 

90 return render_template( 

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

92 subscription=None, 

93 default_settings=default_settings, 

94 ) 

95 

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

97 def edit_subscription_page(subscription_id): 

98 """Render the edit subscription page.""" 

99 from flask import session 

100 

101 # Get username from session 

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

103 

104 # Load subscription data 

105 subscription = None 

106 default_settings = { 

107 "iterations": 3, 

108 "questions_per_iteration": 5, 

109 "search_engine": "auto", 

110 "model_provider": "OLLAMA", 

111 "model": "", 

112 "search_strategy": "source-based", 

113 } 

114 

115 try: 

116 # Load the subscription using the API 

117 subscription = api.get_subscription(subscription_id) 

118 logger.info( 

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

120 ) 

121 

122 if not subscription: 

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

124 # Could redirect to 404 or subscriptions page 

125 return render_template( 

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

127 subscription=None, 

128 error="Subscription not found", 

129 default_settings=default_settings, 

130 ) 

131 

132 # Load user's default settings if logged in 

133 if username != "anonymous": 

134 # Load user settings using the extracted function 

135 from local_deep_research.database.session_context import ( 

136 get_user_db_session, 

137 ) 

138 

139 with get_user_db_session(username) as db_session: 

140 load_user_settings(default_settings, db_session, username) 

141 

142 except Exception as e: 

143 logger.exception( 

144 f"Error loading subscription {subscription_id}: {e}" 

145 ) 

146 return render_template( 

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

148 subscription=None, 

149 error="Error loading subscription", 

150 default_settings=default_settings, 

151 ) 

152 

153 return render_template( 

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

155 subscription=subscription, 

156 default_settings=default_settings, 

157 ) 

158 

159 # Health check 

160 @bp.route("/health") 

161 def health_check(): 

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

163 try: 

164 # Check if database is accessible 

165 from .core.storage_manager import StorageManager 

166 

167 storage = StorageManager() 

168 

169 # Try a simple query 

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

171 

172 return jsonify( 

173 { 

174 "status": "healthy", 

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

176 "database": "connected", 

177 } 

178 ) 

179 except Exception: 

180 logger.exception("Health check failed") 

181 return jsonify( 

182 { 

183 "status": "unhealthy", 

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

185 } 

186 ), 500 

187 

188 return bp 

189 

190 

191def load_user_settings(default_settings, db_session=None, username=None): 

192 """ 

193 Load user settings and update default_settings dictionary. 

194 Extracted to avoid code duplication as suggested by djpetti. 

195 

196 Args: 

197 default_settings: Dictionary to update with user settings 

198 db_session: Database session for accessing settings 

199 username: Username for settings context 

200 """ 

201 if not db_session: 

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

203 return 

204 

205 try: 

206 from ..utilities.db_utils import get_settings_manager 

207 

208 settings_manager = get_settings_manager(db_session, username) 

209 

210 default_settings.update( 

211 { 

212 "iterations": settings_manager.get_setting( 

213 "search.iterations", 3 

214 ), 

215 "questions_per_iteration": settings_manager.get_setting( 

216 "search.questions_per_iteration", 5 

217 ), 

218 "search_engine": settings_manager.get_setting( 

219 "search.tool", "auto" 

220 ), 

221 "model_provider": settings_manager.get_setting( 

222 "llm.provider", "OLLAMA" 

223 ), 

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

225 "search_strategy": settings_manager.get_setting( 

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

227 ), 

228 "custom_endpoint": settings_manager.get_setting( 

229 "llm.openai_endpoint.url", "" 

230 ), 

231 } 

232 ) 

233 except Exception as e: 

234 logger.warning(f"Could not load user settings: {e}") 

235 # Use defaults