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

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

2 

3from pathlib import Path 

4from typing import Dict, Any, 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 delete_report( 

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

119 ) -> bool: 

120 """Delete report files.""" 

121 try: 

122 report_path = self._get_report_path(research_id) 

123 metadata_path = self._get_metadata_path(research_id) 

124 

125 deleted = False 

126 

127 if report_path.exists(): 

128 report_path.unlink() 

129 deleted = True 

130 

131 if metadata_path.exists(): 

132 metadata_path.unlink() 

133 

134 return deleted 

135 

136 except Exception: 

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

138 return False 

139 

140 def report_exists( 

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

142 ) -> bool: 

143 """Check if report file exists.""" 

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