Coverage for src/local_deep_research/storage/file.py: 97%

1"""File-based report storage implementation.""" 

2 

3from pathlib import Path 

4from typing import Dict, Any, List, Optional 

5from loguru import logger 

6 

7from .base import ReportStorage 

8from ..config.paths import get_research_outputs_directory 

9 

10 

11class FileReportStorage(ReportStorage): 

12 """Store reports as files on disk.""" 

13 

14 def __init__(self, base_dir: Optional[Path] = None): 

15 """Initialize file storage. 

16 

17 Args: 

18 base_dir: Base directory for storing reports. 

19 If None, uses default research outputs directory. 

20 """ 

21 # Lazy import: mirrors the other lazy security imports in this 

22 # module (e.g. file_write_verifier) and avoids pulling security 

23 # into this module's top-level import graph. 

24 from ..security.directory_creation import create_directory 

25 

26 self.base_dir = base_dir or get_research_outputs_directory() 

27 create_directory( 

28 self.base_dir, 

29 context="file report storage base directory", 

30 ) 

31 

32 def _get_report_path(self, research_id: str) -> Path: 

33 """Get the file path for a report.""" 

34 candidate = (self.base_dir / f"{research_id}.md").resolve() 

35 if not candidate.is_relative_to(self.base_dir.resolve()): 

36 raise ValueError( 

37 f"Path traversal attempt with research_id: {research_id!r}" 

38 ) 

39 return candidate 

40 

41 def _get_metadata_path(self, research_id: str) -> Path: 

42 """Get the file path for report metadata.""" 

43 candidate = (self.base_dir / f"{research_id}_metadata.json").resolve() 

44 if not candidate.is_relative_to(self.base_dir.resolve()): 

45 raise ValueError( 

46 f"Path traversal attempt with research_id: {research_id!r}" 

47 ) 

48 return candidate 

49 

50 def save_report( 

51 self, 

52 research_id: str, 

53 content: str, 

54 metadata: Optional[Dict[str, Any]] = None, 

55 username: Optional[str] = None, 

56 ) -> bool: 

57 """Save report to file.""" 

58 try: 

59 from ..security.file_write_verifier import ( 

60 write_file_verified, 

61 write_json_verified, 

62 ) 

63 

64 report_path = self._get_report_path(research_id) 

65 

66 # Save content 

67 write_file_verified( 

68 report_path, 

69 content, 

70 "report.enable_file_backup", 

71 context="file storage backup", 

72 ) 

73 

74 # Save metadata if provided 

75 if metadata: 

76 metadata_path = self._get_metadata_path(research_id) 

77 write_json_verified( 

78 metadata_path, 

79 metadata, 

80 "report.enable_file_backup", 

81 context="file storage metadata", 

82 ) 

83 

84 logger.info( 

85 f"Saved report for research {research_id} to {report_path}" 

86 ) 

87 return True 

88 

89 except Exception: 

90 logger.exception("Error saving report to file") 

91 return False 

92 

93 def get_report( 

94 self, research_id: str, username: Optional[str] = None 

95 ) -> Optional[str]: 

96 """Get report from file.""" 

97 try: 

98 report_path = self._get_report_path(research_id) 

99 

100 if not report_path.exists(): 

101 return None 

102 

103 with open(report_path, "r", encoding="utf-8") as f: 

104 return f.read() 

105 

106 except Exception: 

107 logger.exception("Error reading report from file") 

108 return None 

109 

110 def get_report_with_metadata( 

111 self, research_id: str, username: Optional[str] = None 

112 ) -> Optional[Dict[str, Any]]: 

113 """Get report with metadata from files.""" 

114 try: 

115 content = self.get_report(research_id) 

116 if not content: 

117 return None 

118 

119 result = {"content": content, "metadata": {}} 

120 

121 # Try to load metadata 

122 metadata_path = self._get_metadata_path(research_id) 

123 if metadata_path.exists(): 

124 import json 

125 

126 with open(metadata_path, "r", encoding="utf-8") as f: 

127 result["metadata"] = json.load(f) 

128 

129 return result 

130 

131 except Exception: 

132 logger.exception("Error getting report with metadata") 

133 return None 

134 

135 def list_reports( 

136 self, username: Optional[str] = None 

137 ) -> List[Dict[str, Any]]: 

138 """List reports from file system.""" 

139 try: 

140 from datetime import datetime, timezone 

141 

142 reports = [] 

143 for report_path in self.base_dir.glob("*.md"): 

144 research_id = report_path.stem 

145 metadata = self._load_metadata(research_id) 

146 reports.append( 

147 { 

148 "id": research_id, 

149 "query": metadata.get("query") if metadata else None, 

150 "mode": metadata.get("mode", "unknown") 

151 if metadata 

152 else "unknown", 

153 "created_at": datetime.fromtimestamp( 

154 report_path.stat().st_mtime, tz=timezone.utc 

155 ).isoformat(), 

156 "completed_at": metadata.get("completed_at") 

157 if metadata 

158 else None, 

159 } 

160 ) 

161 return reports 

162 except Exception: 

163 logger.exception("Error listing reports from file system") 

164 return [] 

165 

166 def _load_metadata(self, research_id: str) -> Optional[Dict[str, Any]]: 

167 """Load metadata JSON for a report if it exists.""" 

168 try: 

169 metadata_path = self._get_metadata_path(research_id) 

170 if metadata_path.exists(): 

171 import json 

172 

173 with open(metadata_path, "r", encoding="utf-8") as f: 

174 return json.load(f) 

175 except Exception: 

176 logger.debug(f"Failed to load metadata for {research_id}") 

177 return None 

178 

179 def delete_report( 

180 self, research_id: str, username: Optional[str] = None 

181 ) -> bool: 

182 """Delete report files.""" 

183 try: 

184 report_path = self._get_report_path(research_id) 

185 metadata_path = self._get_metadata_path(research_id) 

186 

187 deleted = False 

188 

189 if report_path.exists(): 

190 report_path.unlink() 

191 deleted = True 

192 

193 if metadata_path.exists(): 

194 metadata_path.unlink() 

195 

196 return deleted 

197 

198 except Exception: 

199 logger.exception("Error deleting report files") 

200 return False 

201 

202 def report_exists( 

203 self, research_id: str, username: Optional[str] = None 

204 ) -> bool: 

205 """Check if report file exists.""" 

206 return self._get_report_path(research_id).exists()