Coverage for src / local_deep_research / web / models / database.py: 48%

1import os 

2from datetime import datetime, UTC 

3 

4from loguru import logger 

5 

6from ...config.paths import get_data_directory 

7from ...database.models import ResearchLog 

8from ...database.session_context import get_user_db_session 

9 

10# Database paths using new centralized configuration 

11DATA_DIR = get_data_directory() 

12if DATA_DIR: 12 ↛ 19line 12 didn't jump to line 19 because the condition on line 12 was always true

13 DATA_DIR = str(DATA_DIR) 

14 os.makedirs(DATA_DIR, exist_ok=True) 

15 

16# DB_PATH removed - use per-user encrypted databases instead 

17 

18 

19def get_db_connection(): 

20 """ 

21 Get a connection to the SQLite database. 

22 DEPRECATED: This uses the shared database which should not be used. 

23 Use get_db_session() instead for per-user databases. 

24 """ 

25 raise RuntimeError( 

26 "Shared database access is deprecated. Use get_db_session() for per-user databases." 

27 ) 

28 

29 

30def calculate_duration(created_at_str, completed_at_str=None): 

31 """ 

32 Calculate duration in seconds between created_at timestamp and completed_at or now. 

33 Handles various timestamp formats and returns None if calculation fails. 

34 

35 Args: 

36 created_at_str: The start timestamp 

37 completed_at_str: Optional end timestamp, defaults to current time if None 

38 

39 Returns: 

40 Duration in seconds or None if calculation fails 

41 """ 

42 if not created_at_str: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 return None 

44 

45 end_time = None 

46 if completed_at_str: 46 ↛ 81line 46 didn't jump to line 81 because the condition on line 46 was always true

47 # Use completed_at time if provided 

48 try: 

49 if "T" in completed_at_str: # ISO format with T separator 49 ↛ 53line 49 didn't jump to line 53 because the condition on line 49 was always true

50 end_time = datetime.fromisoformat(completed_at_str) 

51 else: # Older format without T 

52 # Try different formats 

53 try: 

54 end_time = datetime.strptime( 

55 completed_at_str, "%Y-%m-%d %H:%M:%S.%f" 

56 ) 

57 except ValueError: 

58 try: 

59 end_time = datetime.strptime( 

60 completed_at_str, "%Y-%m-%d %H:%M:%S" 

61 ) 

62 except ValueError: 

63 # Last resort fallback 

64 end_time = datetime.fromisoformat( 

65 completed_at_str.replace(" ", "T") 

66 ) 

67 except Exception: 

68 logger.exception("Error parsing completed_at timestamp") 

69 try: 

70 from dateutil import parser 

71 

72 end_time = parser.parse(completed_at_str) 

73 except Exception: 

74 logger.exception( 

75 f"Fallback parsing also failed for completed_at: {completed_at_str}" 

76 ) 

77 # Fall back to current time 

78 end_time = datetime.now(UTC) 

79 else: 

80 # Use current time if no completed_at provided 

81 end_time = datetime.now(UTC) 

82 # Ensure end_time is UTC. 

83 end_time = end_time.astimezone(UTC) 

84 

85 start_time = None 

86 try: 

87 # Proper parsing of ISO format 

88 if "T" in created_at_str: # ISO format with T separator 88 ↛ 92line 88 didn't jump to line 92 because the condition on line 88 was always true

89 start_time = datetime.fromisoformat(created_at_str) 

90 else: # Older format without T 

91 # Try different formats 

92 try: 

93 start_time = datetime.strptime( 

94 created_at_str, "%Y-%m-%d %H:%M:%S.%f" 

95 ) 

96 except ValueError: 

97 try: 

98 start_time = datetime.strptime( 

99 created_at_str, "%Y-%m-%d %H:%M:%S" 

100 ) 

101 except ValueError: 

102 # Last resort fallback 

103 start_time = datetime.fromisoformat( 

104 created_at_str.replace(" ", "T") 

105 ) 

106 except Exception: 

107 logger.exception("Error parsing created_at timestamp") 

108 # Fallback method if parsing fails 

109 try: 

110 from dateutil import parser 

111 

112 start_time = parser.parse(created_at_str) 

113 except Exception: 

114 logger.exception( 

115 f"Fallback parsing also failed for created_at: {created_at_str}" 

116 ) 

117 return None 

118 

119 # Calculate duration if both timestamps are valid 

120 if start_time and end_time: 120 ↛ 126line 120 didn't jump to line 126 because the condition on line 120 was always true

121 try: 

122 return int((end_time - start_time).total_seconds()) 

123 except Exception: 

124 logger.exception("Error calculating duration") 

125 

126 return None 

127 

128 

129def get_logs_for_research(research_id): 

130 """ 

131 Retrieve all logs for a specific research ID 

132 

133 Args: 

134 research_id: ID of the research 

135 

136 Returns: 

137 List of log entries as dictionaries 

138 """ 

139 try: 

140 with get_user_db_session() as session: 

141 log_results = ( 

142 session.query(ResearchLog) 

143 .filter(ResearchLog.research_id == research_id) 

144 .order_by(ResearchLog.timestamp.asc()) 

145 .all() 

146 ) 

147 

148 logs = [] 

149 for result in log_results: 149 ↛ 151line 149 didn't jump to line 151 because the loop on line 149 never started

150 # Convert entry for frontend consumption 

151 formatted_entry = { 

152 "time": result.timestamp, 

153 "message": result.message, 

154 "type": result.level, 

155 "module": result.module, 

156 "line_no": result.line_no, 

157 } 

158 logs.append(formatted_entry) 

159 

160 return logs 

161 except Exception: 

162 logger.exception("Error retrieving logs from database") 

163 return [] 

164 

165 

166@logger.catch 

167def get_total_logs_for_research(research_id): 

168 """ 

169 Returns the total number of logs for a given `research_id`. 

170 

171 Args: 

172 research_id (int): The ID of the research. 

173 

174 Returns: 

175 int: Total number of logs for the specified research ID. 

176 """ 

177 with get_user_db_session() as session: 

178 total_logs = ( 

179 session.query(ResearchLog) 

180 .filter(ResearchLog.research_id == research_id) 

181 .count() 

182 ) 

183 return total_logs