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 self.base_dir = base_dir or get_research_outputs_directory() 

22 self.base_dir.mkdir(parents=True, exist_ok=True) 

23 

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

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

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

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

28 raise ValueError( 

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

30 ) 

31 return candidate 

32 

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

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

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

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

37 raise ValueError( 

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

39 ) 

40 return candidate 

41 

42 def save_report( 

43 self, 

44 research_id: str, 

45 content: str, 

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

47 username: Optional[str] = None, 

48 ) -> bool: 

49 """Save report to file.""" 

50 try: 

51 from ..security.file_write_verifier import ( 

52 write_file_verified, 

53 write_json_verified, 

54 ) 

55 

56 report_path = self._get_report_path(research_id) 

57 

58 # Save content 

59 write_file_verified( 

60 report_path, 

61 content, 

62 "report.enable_file_backup", 

63 context="file storage backup", 

64 ) 

65 

66 # Save metadata if provided 

67 if metadata: 

68 metadata_path = self._get_metadata_path(research_id) 

69 write_json_verified( 

70 metadata_path, 

71 metadata, 

72 "report.enable_file_backup", 

73 context="file storage metadata", 

74 ) 

75 

76 logger.info( 

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

78 ) 

79 return True 

80 

81 except Exception: 

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

83 return False 

84 

85 def get_report( 

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

87 ) -> Optional[str]: 

88 """Get report from file.""" 

89 try: 

90 report_path = self._get_report_path(research_id) 

91 

92 if not report_path.exists(): 

93 return None 

94 

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

96 return f.read() 

97 

98 except Exception: 

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

100 return None 

101 

102 def get_report_with_metadata( 

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

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

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

106 try: 

107 content = self.get_report(research_id) 

108 if not content: 

109 return None 

110 

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

112 

113 # Try to load metadata 

114 metadata_path = self._get_metadata_path(research_id) 

115 if metadata_path.exists(): 

116 import json 

117 

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

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

120 

121 return result 

122 

123 except Exception: 

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

125 return None 

126 

127 def list_reports( 

128 self, username: Optional[str] = None 

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

130 """List reports from file system.""" 

131 try: 

132 from datetime import datetime, timezone 

133 

134 reports = [] 

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

136 research_id = report_path.stem 

137 metadata = self._load_metadata(research_id) 

138 reports.append( 

139 { 

140 "id": research_id, 

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

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

143 if metadata 

144 else "unknown", 

145 "created_at": datetime.fromtimestamp( 

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

147 ).isoformat(), 

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

149 if metadata 

150 else None, 

151 } 

152 ) 

153 return reports 

154 except Exception: 

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

156 return [] 

157 

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

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

160 try: 

161 metadata_path = self._get_metadata_path(research_id) 

162 if metadata_path.exists(): 

163 import json 

164 

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

166 return json.load(f) 

167 except Exception: 

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

169 return None 

170 

171 def delete_report( 

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

173 ) -> bool: 

174 """Delete report files.""" 

175 try: 

176 report_path = self._get_report_path(research_id) 

177 metadata_path = self._get_metadata_path(research_id) 

178 

179 deleted = False 

180 

181 if report_path.exists(): 

182 report_path.unlink() 

183 deleted = True 

184 

185 if metadata_path.exists(): 

186 metadata_path.unlink() 

187 

188 return deleted 

189 

190 except Exception: 

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

192 return False 

193 

194 def report_exists( 

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

196 ) -> bool: 

197 """Check if report file exists.""" 

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