Coverage for health / cli / sync.py: 0%

1""" 

2CLI tool for Garmin health data synchronization. 

3 

4Usage: 

5 python -m health.cli.sync status # Show sync status 

6 python -m health.cli.sync sync-now # Sync today 

7 python -m health.cli.sync sync --days 7 # Sync last 7 days 

8 python -m health.cli.sync backfill --start-date 2024-01-01 # Historical backfill 

9""" 

10 

11import sys 

12from datetime import date, timedelta 

13from typing import Optional 

14 

15import click 

16 

17from health import config 

18from health.services.data_sync import HealthDataSync 

19from health.utils.date_utils import parse_date, get_yesterday 

20from health.utils.logging_config import setup_logger 

21from health.utils.exceptions import GarminAuthError, SyncError 

22 

23logger = setup_logger(__name__) 

24 

25 

26@click.group() 

27def cli() -> None: 

28 """Garmin Health Data Sync CLI.""" 

29 pass 

30 

31 

32@cli.command() 

33def status() -> None: 

34 """Show synchronization status for all data types.""" 

35 click.echo("📊 Health Data Sync Status\n") 

36 

37 sync_service = HealthDataSync() 

38 status_info = sync_service.get_sync_status() 

39 

40 # Group by status 

41 synced = [] 

42 never_synced = [] 

43 

44 for data_type, info in status_info.items(): 

45 if info["status"] == "synced": 

46 synced.append((data_type, info)) 

47 else: 

48 never_synced.append(data_type) 

49 

50 # Display synced data types 

51 if synced: 

52 click.echo("✅ Synced Data Types:") 

53 for data_type, info in synced: 

54 description = config.DATA_TYPE_CONFIG.get(data_type, {}).get( 

55 "description", "" 

56 ) 

57 click.echo( 

58 f" • {data_type:20} Last sync: {info['last_sync_date']} " 

59 f"({info['total_records']} records) - {description}" 

60 ) 

61 

62 # Display never synced 

63 if never_synced: 

64 click.echo("\n⚠️ Never Synced:") 

65 for data_type in never_synced: 

66 description = config.DATA_TYPE_CONFIG.get(data_type, {}).get( 

67 "description", "" 

68 ) 

69 click.echo(f" • {data_type:20} {description}") 

70 

71 click.echo(f"\n📁 Data directory: {config.DATA_DIR}") 

72 click.echo(f"🗄️ Database: {config.DB_PATH}") 

73 

74 

75@cli.command() 

76@click.option( 

77 "--days", 

78 "-d", 

79 type=int, 

80 default=1, 

81 help="Number of days to sync (from yesterday backwards)", 

82) 

83@click.option("--force", "-f", is_flag=True, help="Force re-sync existing data") 

84def sync(days: int, force: bool) -> None: 

85 """Sync health data for recent days.""" 

86 try: 

87 # Calculate date range 

88 end_date = get_yesterday() 

89 start_date = end_date - timedelta(days=days - 1) 

90 

91 click.echo(f"🔄 Syncing health data from {start_date} to {end_date}") 

92 if force: 

93 click.echo(" (Force mode: re-syncing existing data)") 

94 

95 # Initialize sync service 

96 sync_service = HealthDataSync() 

97 

98 # Authenticate 

99 click.echo("🔐 Authenticating with Garmin Connect China...") 

100 sync_service.authenticate() 

101 click.echo("✅ Authentication successful\n") 

102 

103 # Sync all metrics 

104 results = sync_service.sync_all_metrics( 

105 start_date, end_date, force=force 

106 ) 

107 

108 # Display results 

109 click.echo("\n📊 Sync Results:") 

110 total_synced = 0 

111 total_errors = 0 

112 

113 for metric_type, stats in results.items(): 

114 synced = stats.get("synced", 0) 

115 skipped = stats.get("skipped", 0) 

116 errors = stats.get("errors", 0) 

117 

118 total_synced += synced 

119 total_errors += errors 

120 

121 status_icon = "✓" if errors == 0 else "⚠" 

122 click.echo( 

123 f" {status_icon} {metric_type:20} " 

124 f"Synced: {synced:3}, Skipped: {skipped:3}, Errors: {errors:3}" 

125 ) 

126 

127 click.echo( 

128 f"\n{'✅' if total_errors == 0 else '⚠️ '} " 

129 f"Total: {total_synced} records synced, {total_errors} errors" 

130 ) 

131 

132 sys.exit(0 if total_errors == 0 else 1) 

133 

134 except GarminAuthError as e: 

135 click.echo(f"\n❌ Authentication failed: {e}", err=True) 

136 click.echo( 

137 " Please check your Garmin credentials in the .env file.", err=True 

138 ) 

139 sys.exit(1) 

140 except Exception as e: 

141 click.echo(f"\n❌ Sync failed: {e}", err=True) 

142 logger.exception("Sync failed") 

143 sys.exit(1) 

144 

145 

146@cli.command() 

147@click.option( 

148 "--start-date", 

149 "-s", 

150 required=True, 

151 help="Start date for backfill (YYYY-MM-DD)", 

152) 

153@click.option( 

154 "--end-date", 

155 "-e", 

156 default=None, 

157 help="End date for backfill (defaults to yesterday)", 

158) 

159@click.option( 

160 "--batch-size", 

161 "-b", 

162 type=int, 

163 default=config.DEFAULT_BATCH_SIZE_DAYS, 

164 help=f"Days per batch (default: {config.DEFAULT_BATCH_SIZE_DAYS})", 

165) 

166@click.option( 

167 "--metrics", 

168 "-m", 

169 multiple=True, 

170 help="Specific metrics to sync (can be specified multiple times)", 

171) 

172def backfill( 

173 start_date: str, 

174 end_date: Optional[str], 

175 batch_size: int, 

176 metrics: tuple, 

177) -> None: 

178 """Backfill historical health data.""" 

179 try: 

180 # Parse dates 

181 start = parse_date(start_date) 

182 end = parse_date(end_date) if end_date else get_yesterday() 

183 

184 # Validate date range 

185 if start > end: 

186 click.echo("❌ Start date must be before end date", err=True) 

187 sys.exit(1) 

188 

189 total_days = (end - start).days + 1 

190 

191 click.echo(f"📚 Historical Backfill") 

192 click.echo(f" Start date: {start}") 

193 click.echo(f" End date: {end}") 

194 click.echo(f" Total days: {total_days}") 

195 click.echo(f" Batch size: {batch_size} days") 

196 

197 if metrics: 

198 click.echo(f" Metrics: {', '.join(metrics)}") 

199 

200 # Confirm large backfills 

201 if total_days > 90: 

202 click.confirm( 

203 f"\n⚠️ This will backfill {total_days} days of data. Continue?", 

204 abort=True, 

205 ) 

206 

207 # Initialize sync service 

208 sync_service = HealthDataSync() 

209 

210 # Authenticate 

211 click.echo("\n🔐 Authenticating with Garmin Connect China...") 

212 sync_service.authenticate() 

213 click.echo("✅ Authentication successful\n") 

214 

215 # Perform backfill 

216 metric_list = list(metrics) if metrics else None 

217 results = sync_service.backfill_historical( 

218 start, end, metric_types=metric_list, batch_size=batch_size 

219 ) 

220 

221 # Display final results 

222 click.echo("\n📊 Backfill Complete!") 

223 total_synced = sum(s.get("synced", 0) for s in results.values()) 

224 total_errors = sum(s.get("errors", 0) for s in results.values()) 

225 

226 for metric_type, stats in results.items(): 

227 synced = stats.get("synced", 0) 

228 errors = stats.get("errors", 0) 

229 status_icon = "✓" if errors == 0 else "⚠" 

230 

231 click.echo( 

232 f" {status_icon} {metric_type:20} {synced:4} records " 

233 f"({errors} errors)" if errors else f" {status_icon} {metric_type:20} {synced:4} records" 

234 ) 

235 

236 click.echo( 

237 f"\n{'✅' if total_errors == 0 else '⚠️ '} " 

238 f"Total: {total_synced} records synced" 

239 ) 

240 

241 sys.exit(0 if total_errors == 0 else 1) 

242 

243 except GarminAuthError as e: 

244 click.echo(f"\n❌ Authentication failed: {e}", err=True) 

245 sys.exit(1) 

246 except Exception as e: 

247 click.echo(f"\n❌ Backfill failed: {e}", err=True) 

248 logger.exception("Backfill failed") 

249 sys.exit(1) 

250 

251 

252@cli.command() 

253def sync_incremental() -> None: 

254 """Perform incremental sync (sync new data since last sync).""" 

255 try: 

256 click.echo("🔄 Incremental Sync (syncing new data only)\n") 

257 

258 # Initialize sync service 

259 sync_service = HealthDataSync() 

260 

261 # Authenticate 

262 click.echo("🔐 Authenticating...") 

263 sync_service.authenticate() 

264 click.echo("✅ Authenticated\n") 

265 

266 # Perform incremental sync 

267 results = sync_service.sync_incremental() 

268 

269 # Display results 

270 click.echo("📊 Sync Results:") 

271 total_synced = 0 

272 total_errors = 0 

273 

274 for metric_type, stats in results.items(): 

275 synced = stats.get("synced", 0) 

276 errors = stats.get("errors", 0) 

277 

278 if synced > 0 or errors > 0: 

279 total_synced += synced 

280 total_errors += errors 

281 

282 status_icon = "✓" if errors == 0 else "⚠" 

283 click.echo( 

284 f" {status_icon} {metric_type:20} {synced:3} new records " 

285 f"({errors} errors)" if errors else f" {status_icon} {metric_type:20} {synced:3} new records" 

286 ) 

287 

288 if total_synced == 0: 

289 click.echo(" ℹ️ All data is up to date!") 

290 

291 click.echo( 

292 f"\n{'✅' if total_errors == 0 else '⚠️ '} " 

293 f"Synced {total_synced} new records" 

294 ) 

295 

296 sys.exit(0 if total_errors == 0 else 1) 

297 

298 except GarminAuthError as e: 

299 click.echo(f"\n❌ Authentication failed: {e}", err=True) 

300 sys.exit(1) 

301 except Exception as e: 

302 click.echo(f"\n❌ Incremental sync failed: {e}", err=True) 

303 logger.exception("Incremental sync failed") 

304 sys.exit(1) 

305 

306 

307if __name__ == "__main__": 

308 cli()