Coverage for health / services / obsidian.py: 0%

1import re 

2import os 

3from datetime import date 

4from pathlib import Path 

5from typing import Dict, List, Optional, Any 

6from dataclasses import dataclass 

7 

8from health.services.manual_log_storage import ManualLogStorage 

9from health.models.manual_log import SupplementEntry, AlcoholEntry 

10from health.utils.logging_config import setup_logger 

11 

12logger = setup_logger(__name__) 

13 

14# Default Obsidian Daily Notes Path (can be overridden) 

15DEFAULT_OBSIDIAN_PATH = Path.home() / "Library/Mobile Documents/iCloud~md~obsidian/Documents/obsidian/daily" 

16 

17# Mapping from Obsidian YAML keys to Supplement names 

18SUPPLEMENT_MAPPING = { 

19 "MED_001": "Medication 001", 

20 "nac": "NAC", 

21 "magnesium": "Magnesium", 

22 "fish_oil": "Fish Oil", 

23 "coq10": "CoQ10", 

24 "berberine": "Berberine", 

25 "b_complex": "Vitamin B Complex", 

26 "v_d3_k2": "Vitamin D3+K2" 

27} 

28 

29# Mapping for Fasting Modes 

30FASTING_MAPPING = { 

31 "PSMF": "PSMF", 

32 "OMAD": "OMAD", 

33 "WaterFast": "Water Fast" 

34} 

35 

36class ObsidianSyncService: 

37 """Service to sync data from Obsidian daily notes.""" 

38 

39 def __init__(self, obsidian_dir: Optional[Path] = None): 

40 self.obsidian_dir = obsidian_dir or DEFAULT_OBSIDIAN_PATH 

41 self.storage = ManualLogStorage() 

42 

43 def _get_note_path(self, target_date: date) -> Path: 

44 """Get the path to the daily note for a given date.""" 

45 filename = f"{target_date.isoformat()}.md" 

46 return self.obsidian_dir / filename 

47 

48 def _parse_frontmatter(self, content: str) -> Dict[str, Any]: 

49 """Parse YAML frontmatter from markdown content manually.""" 

50 frontmatter = {} 

51 

52 # Regex to find content between --- 

53 match = re.search(r"^---\n(.*?)\n---", content, re.DOTALL) 

54 if not match: 

55 return frontmatter 

56 

57 yaml_content = match.group(1) 

58 

59 # Parse line by line 

60 for line in yaml_content.split('\n'): 

61 line = line.strip() 

62 if not line or line.startswith('#'): 

63 continue 

64 

65 if ':' in line: 

66 key, value = line.split(':', 1) 

67 key = key.strip() 

68 value = value.strip() 

69 

70 # Handle booleans 

71 if value.lower() == 'true': 

72 value = True 

73 elif value.lower() == 'false': 

74 value = False 

75 # Handle numbers 

76 elif value.replace('.', '', 1).isdigit(): 

77 try: 

78 value = float(value) if '.' in value else int(value) 

79 except ValueError: 

80 pass 

81 # Handle nulls 

82 elif value.lower() in ('null', 'nil', ''): 

83 value = None 

84 

85 frontmatter[key] = value 

86 

87 return frontmatter 

88 

89 def sync_daily_note(self, target_date: date) -> bool: 

90 """Sync data from Obsidian for a specific date. 

91  

92 Returns: 

93 True if data was found and synced, False otherwise. 

94 """ 

95 note_path = self._get_note_path(target_date) 

96 if not note_path.exists(): 

97 logger.warning(f"No Obsidian note found for {target_date} at {note_path}") 

98 return False 

99 

100 try: 

101 with open(note_path, 'r', encoding='utf-8') as f: 

102 content = f.read() 

103 

104 data = self._parse_frontmatter(content) 

105 if not data: 

106 logger.info(f"No frontmatter found in {note_path}") 

107 return False 

108 

109 log = self.storage.load_log(target_date) 

110 changes_made = False 

111 

112 # 1. Sync Fasting Mode 

113 for key, mode in FASTING_MAPPING.items(): 

114 if data.get(key) is True: 

115 if log.fasting_mode != mode: 

116 log.fasting_mode = mode 

117 changes_made = True 

118 logger.info(f"Synced Fasting Mode: {mode}") 

119 break # Only one fasting mode per day 

120 

121 if data.get("fasting") is False and log.fasting_mode: 

122 # If explicitly false, maybe clear it?  

123 # Decided to safer: only set if True.  

124 pass 

125 

126 # 2. Sync Supplements 

127 for key, name in SUPPLEMENT_MAPPING.items(): 

128 if data.get(key) is True: 

129 # Check if already exists 

130 exists = any( 

131 s.supplement_name == name and s.time == "00:00" 

132 for s in log.supplement_entries 

133 ) 

134 

135 if not exists: 

136 entry = SupplementEntry( 

137 time="00:00", # Default time 

138 supplement_name=name, 

139 dosage=None, # Dosage not tracked in Obsidian YAML boolean 

140 timing="obsidian_sync" 

141 ) 

142 log.supplement_entries.append(entry) 

143 changes_made = True 

144 logger.info(f"Synced Supplement: {name}") 

145 

146 # 3. Sync Alcohol 

147 if data.get("alcohol") is True: 

148 # Check for existing alcohol entry from Obsidian (generic one) 

149 alcohol_ml = data.get("alcohol_ml", 0) 

150 amount_str = f"{alcohol_ml}ml pure alcohol" if alcohol_ml else "See Obsidian" 

151 

152 # Check if we already have a 'wine' entry at 00:00 (our default) 

153 exists = any( 

154 (a.time == "00:00" and a.type == "wine") 

155 for a in log.alcohol_entries 

156 ) 

157 

158 if not exists: 

159 entry = AlcoholEntry( 

160 time="00:00", 

161 type="wine", # Default type per user request 

162 amount=amount_str, 

163 notes=f"Synced from Obsidian (alcohol_ml: {alcohol_ml})" 

164 ) 

165 log.alcohol_entries.append(entry) 

166 changes_made = True 

167 logger.info(f"Synced Alcohol: {amount_str}") 

168 

169 if changes_made: 

170 self.storage.save_log(log) 

171 logger.info(f"Successfully synced Obsidian data for {target_date}") 

172 return True 

173 else: 

174 logger.info(f"No new data to sync for {target_date}") 

175 return True 

176 

177 except Exception as e: 

178 logger.error(f"Failed to sync Obsidian note for {target_date}: {e}") 

179 return False