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 return self.base_dir / f"{research_id}.md" 

27 

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

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

30 return self.base_dir / f"{research_id}_metadata.json" 

31 

32 def save_report( 

33 self, 

34 research_id: str, 

35 content: str, 

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

37 username: Optional[str] = None, 

38 ) -> bool: 

39 """Save report to file.""" 

40 try: 

41 from ..security.file_write_verifier import ( 

42 write_file_verified, 

43 write_json_verified, 

44 ) 

45 

46 report_path = self._get_report_path(research_id) 

47 

48 # Save content 

49 write_file_verified( 

50 report_path, 

51 content, 

52 "storage.allow_file_backup", 

53 context="file storage backup", 

54 ) 

55 

56 # Save metadata if provided 

57 if metadata: 

58 metadata_path = self._get_metadata_path(research_id) 

59 write_json_verified( 

60 metadata_path, 

61 metadata, 

62 "storage.allow_file_backup", 

63 context="file storage metadata", 

64 ) 

65 

66 logger.info( 

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

68 ) 

69 return True 

70 

71 except Exception: 

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

73 return False 

74 

75 def get_report( 

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

77 ) -> Optional[str]: 

78 """Get report from file.""" 

79 try: 

80 report_path = self._get_report_path(research_id) 

81 

82 if not report_path.exists(): 

83 return None 

84 

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

86 return f.read() 

87 

88 except Exception: 

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

90 return None 

91 

92 def get_report_with_metadata( 

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

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

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

96 try: 

97 content = self.get_report(research_id) 

98 if not content: 

99 return None 

100 

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

102 

103 # Try to load metadata 

104 metadata_path = self._get_metadata_path(research_id) 

105 if metadata_path.exists(): 

106 import json 

107 

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

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

110 

111 return result 

112 

113 except Exception: 

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

115 return None 

116 

117 def list_reports( 

118 self, username: Optional[str] = None 

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

120 """List reports from file system.""" 

121 try: 

122 from datetime import datetime, timezone 

123 

124 reports = [] 

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

126 research_id = report_path.stem 

127 metadata = self._load_metadata(research_id) 

128 reports.append( 

129 { 

130 "id": research_id, 

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

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

133 if metadata 

134 else "unknown", 

135 "created_at": datetime.fromtimestamp( 

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

137 ).isoformat(), 

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

139 if metadata 

140 else None, 

141 } 

142 ) 

143 return reports 

144 except Exception: 

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

146 return [] 

147 

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

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

150 try: 

151 metadata_path = self._get_metadata_path(research_id) 

152 if metadata_path.exists(): 

153 import json 

154 

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

156 return json.load(f) 

157 except Exception: 

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

159 return None 

160 

161 def delete_report( 

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

163 ) -> bool: 

164 """Delete report files.""" 

165 try: 

166 report_path = self._get_report_path(research_id) 

167 metadata_path = self._get_metadata_path(research_id) 

168 

169 deleted = False 

170 

171 if report_path.exists(): 

172 report_path.unlink() 

173 deleted = True 

174 

175 if metadata_path.exists(): 

176 metadata_path.unlink() 

177 

178 return deleted 

179 

180 except Exception: 

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

182 return False 

183 

184 def report_exists( 

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

186 ) -> bool: 

187 """Check if report file exists.""" 

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