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 . 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("/subscriptions") 

48 def subscriptions_page(): 

49 """Render the subscriptions management page.""" 

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

51 

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

53 def new_subscription_page(): 

54 """Render the create subscription page.""" 

55 from flask import session 

56 

57 # Get username from session 

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

59 

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

61 default_settings = { 

62 "iterations": 3, 

63 "questions_per_iteration": 5, 

64 "search_engine": "auto", 

65 "model_provider": "OLLAMA", 

66 "model": "", 

67 "search_strategy": "source-based", 

68 } 

69 

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

71 if username != "anonymous": 

72 # Load user settings using the extracted function 

73 from local_deep_research.database.session_context import ( 

74 get_user_db_session, 

75 ) 

76 

77 with get_user_db_session(username) as db_session: 

78 load_user_settings(default_settings, db_session, username) 

79 

80 return render_template( 

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

82 subscription=None, 

83 default_settings=default_settings, 

84 ) 

85 

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

87 def edit_subscription_page(subscription_id): 

88 """Render the edit subscription page.""" 

89 from flask import session 

90 

91 # Get username from session 

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

93 

94 # Load subscription data 

95 subscription = None 

96 default_settings = { 

97 "iterations": 3, 

98 "questions_per_iteration": 5, 

99 "search_engine": "auto", 

100 "model_provider": "OLLAMA", 

101 "model": "", 

102 "search_strategy": "source-based", 

103 } 

104 

105 try: 

106 # Load the subscription using the API 

107 subscription = api.get_subscription(subscription_id) 

108 logger.info( 

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

110 ) 

111 

112 if not subscription: 

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

114 # Could redirect to 404 or subscriptions page 

115 return render_template( 

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

117 subscription=None, 

118 error="Subscription not found", 

119 default_settings=default_settings, 

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 ) 

140 

141 return render_template( 

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

143 subscription=subscription, 

144 default_settings=default_settings, 

145 ) 

146 

147 # Health check 

148 @bp.route("/health") 

149 def health_check(): 

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

151 try: 

152 # Check if database is accessible 

153 from .core.storage_manager import StorageManager 

154 

155 storage = StorageManager() 

156 

157 # Try a simple query 

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

159 

160 return jsonify( 

161 { 

162 "status": "healthy", 

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

164 "database": "connected", 

165 } 

166 ) 

167 except Exception: 

168 logger.exception("Health check failed") 

169 return jsonify( 

170 { 

171 "status": "unhealthy", 

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

173 } 

174 ), 500 

175 

176 return bp 

177 

178 

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

180 """ 

181 Load user settings and update default_settings dictionary. 

182 Extracted to avoid code duplication as suggested by djpetti. 

183 

184 Args: 

185 default_settings: Dictionary to update with user settings 

186 db_session: Database session for accessing settings 

187 username: Username for settings context 

188 """ 

189 if not db_session: 

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

191 return 

192 

193 try: 

194 from ..utilities.db_utils import get_settings_manager 

195 

196 settings_manager = get_settings_manager(db_session, username) 

197 

198 default_settings.update( 

199 { 

200 "iterations": settings_manager.get_setting( 

201 "search.iterations", 3 

202 ), 

203 "questions_per_iteration": settings_manager.get_setting( 

204 "search.questions_per_iteration", 5 

205 ), 

206 "search_engine": settings_manager.get_setting( 

207 "search.tool", "auto" 

208 ), 

209 "model_provider": settings_manager.get_setting( 

210 "llm.provider", "OLLAMA" 

211 ), 

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

213 "search_strategy": settings_manager.get_setting( 

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

215 ), 

216 "custom_endpoint": settings_manager.get_setting( 

217 "llm.openai_endpoint.url", "" 

218 ), 

219 } 

220 ) 

221 except Exception as e: 

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

223 # Use defaults