Coverage for src/local_deep_research/config/paths.py: 100%

54 statements  

« prev     ^ index     » next       coverage.py v7.15.1, created at 2026-07-19 23:35 +0000

1""" 

2Centralized path configuration for Local Deep Research. 

3Handles database location using platformdirs for proper user data storage. 

4""" 

5 

6import hashlib 

7import os 

8from pathlib import Path 

9 

10import platformdirs 

11from loguru import logger 

12 

13 

14def get_data_directory() -> Path: 

15 """ 

16 Get the appropriate data directory for storing application data. 

17 Uses platformdirs to get platform-specific user data directory. 

18 

19 Environment variable: 

20 LDR_DATA_DIR: Override the default data directory location. 

21 All subdirectories (research_outputs, cache, logs, database) 

22 will be created under this directory. 

23 

24 Returns: 

25 Path to data directory 

26 """ 

27 # Check for explicit override via environment variable 

28 custom_path = os.getenv("LDR_DATA_DIR") 

29 if custom_path: 

30 if not Path(custom_path).is_absolute(): 

31 raise ValueError("LDR_DATA_DIR must be an absolute path") 

32 data_dir = Path(custom_path).resolve() 

33 # Reject only control characters that are never valid in a path and 

34 # could break a SQL/ATTACH statement (null byte, newline, carriage 

35 # return). Quotes are intentionally NOT rejected here: legitimate POSIX 

36 # home directories contain apostrophes (e.g. /home/O'Brien/ldr), and the 

37 # actual ATTACH-DATABASE sink (backup_service) already validates the 

38 # full interpolated path against a stricter denylist before use. 

39 # The offending path is not echoed, mirroring the redaction below. 

40 path_str = str(data_dir) 

41 if any(c in path_str for c in ("\0", "\n", "\r")): 

42 raise ValueError( 

43 "LDR_DATA_DIR contains unsafe control characters " 

44 "(null byte, newline, or carriage return)" 

45 ) 

46 logger.debug( 

47 f"Using custom data directory from LDR_DATA_DIR: {data_dir}" 

48 ) 

49 return data_dir 

50 

51 # Use platformdirs for platform-specific user data directory 

52 # Windows: C:\Users\Username\AppData\Local\local-deep-research 

53 # macOS: ~/Library/Application Support/local-deep-research 

54 # Linux: ~/.local/share/local-deep-research 

55 data_dir = Path(platformdirs.user_data_dir("local-deep-research")) 

56 # Log only the directory pattern, not the full path which may contain username 

57 logger.debug( 

58 f"Using platformdirs data directory pattern: .../{data_dir.name}" 

59 ) 

60 

61 return data_dir 

62 

63 

64def _ensure_dir(subdir: str, label: str | None = None) -> Path: 

65 """Create (if needed) and return ``<data_dir>/<subdir>``. 

66 

67 Args: 

68 subdir: Path segment below the data directory. May be multi-segment 

69 (e.g. ``"encrypted_databases/backups"``). 

70 label: If provided, emit a debug log ``"Using {label} directory: 

71 {path}"``. ``None`` suppresses logging. 

72 

73 Returns: 

74 The ensured directory path. 

75 """ 

76 path = get_data_directory() / subdir 

77 path.mkdir(parents=True, exist_ok=True) 

78 if label is not None: 

79 logger.debug(f"Using {label} directory: {path}") 

80 return path 

81 

82 

83def get_research_outputs_directory() -> Path: 

84 """ 

85 Get the directory for storing research outputs (reports, etc.). 

86 

87 Returns: 

88 Path to research outputs directory 

89 """ 

90 return _ensure_dir("research_outputs", label="research outputs") 

91 

92 

93def get_journal_data_directory() -> Path: 

94 """Get the directory for downloaded journal quality data files. 

95 

96 Contains openalex_sources.json.gz, doaj_journals.json, and the 

97 compiled journal_reference.db. Fetched on first use from 

98 OpenAlex and DOAJ APIs. 

99 

100 Returns: 

101 Path to journal data directory 

102 """ 

103 return _ensure_dir("journal_data") 

104 

105 

106def get_cache_directory() -> Path: 

107 """ 

108 Get the directory for storing cache files (search cache, etc.). 

109 

110 Returns: 

111 Path to cache directory 

112 """ 

113 return _ensure_dir("cache", label="cache") 

114 

115 

116def get_logs_directory() -> Path: 

117 """ 

118 Get the directory for storing log files. 

119 

120 Returns: 

121 Path to logs directory 

122 """ 

123 return _ensure_dir("logs", label="logs") 

124 

125 

126def get_encrypted_database_path() -> Path: 

127 """Get the path to the encrypted databases directory. 

128 

129 Returns: 

130 Path to the encrypted databases directory 

131 """ 

132 return _ensure_dir("encrypted_databases") 

133 

134 

135def get_user_database_filename(username: str) -> str: 

136 """Get the database filename for a specific user. 

137 

138 Args: 

139 username: The username to generate a filename for 

140 

141 Returns: 

142 The database filename (not full path) for the user 

143 """ 

144 # Use username hash to avoid filesystem issues with special characters 

145 username_hash = hashlib.sha256(username.encode()).hexdigest()[:16] 

146 return f"ldr_user_{username_hash}.db" 

147 

148 

149def get_library_directory() -> Path: 

150 """ 

151 Get the directory for storing library files (documents, PDFs, etc.). 

152 

153 Returns: 

154 Path to library directory 

155 """ 

156 return _ensure_dir("library", label="library") 

157 

158 

159def get_config_directory() -> Path: 

160 """ 

161 Get the directory for storing configuration files. 

162 

163 Returns: 

164 Path to config directory 

165 """ 

166 return _ensure_dir("config", label="config") 

167 

168 

169def get_models_directory() -> Path: 

170 """ 

171 Get the directory for storing downloaded models. 

172 

173 Returns: 

174 Path to models directory 

175 """ 

176 return _ensure_dir("models", label="models") 

177 

178 

179def get_backup_directory() -> Path: 

180 """Get the base backup directory for all users.""" 

181 return _ensure_dir("encrypted_databases/backups") 

182 

183 

184def get_user_backup_directory(username: str) -> Path: 

185 """Get backup directory for a specific user.""" 

186 username_hash = hashlib.sha256(username.encode()).hexdigest()[:16] 

187 user_backup_dir = get_backup_directory() / username_hash 

188 user_backup_dir.mkdir(parents=True, exist_ok=True, mode=0o700) 

189 # Enforce 0o700 regardless of umask (mkdir mode is umask-masked) 

190 user_backup_dir.chmod(0o700) 

191 return user_backup_dir 

192 

193 

194# Convenience functions for backward compatibility 

195def get_data_dir() -> str: 

196 """Get data directory as string for backward compatibility.""" 

197 return str(get_data_directory())