Coverage for src/notify.py: 100%
325 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-05 00:26 +0000
1"""Send notifications when 2FA is required for iCloud authentication."""
3import datetime
4import smtplib
6import requests
8from src import config_parser, get_logger
9from src.email_message import EmailMessage as Message
11LOGGER = get_logger()
13# Throttling period for notifications (24 hours)
14THROTTLE_HOURS = 24
17def _is_throttled(last_send) -> bool:
18 """
19 Check if notification should be throttled based on last send time.
21 Args:
22 last_send: The datetime when notification was last sent, or None
24 Returns:
25 True if notification should be throttled, False otherwise
26 """
27 if last_send is None:
28 return False
29 if not isinstance(last_send, datetime.datetime):
30 return False
31 return last_send > datetime.datetime.now() - datetime.timedelta(hours=THROTTLE_HOURS)
34def _create_2fa_message(
35 username: str,
36 region: str = "global",
37 dashboard_url: str | None = None,
38) -> tuple[str, str]:
39 """
40 Create the 2FA notification message and subject.
42 Args:
43 username: The iCloud username requiring 2FA
44 region: The iCloud region (default: "global")
45 dashboard_url: Web UI URL (optional). When provided, the message
46 tells the user to tap the URL instead of running the docker
47 exec command. Set from ``app.web_ui.public_url`` (or the
48 host:port fallback) by ``send()`` when ``app.web_ui.enabled``
49 is true.
51 Returns:
52 Tuple of (message, subject)
53 """
54 if dashboard_url:
55 message = (
56 f"icloud-docker: iCloud login required. Sign in at {dashboard_url}/auth"
57 )
58 else:
59 region_opt = "" if region == "global" else f"--region={region} "
60 message = f"""Two-step authentication for iCloud Drive, Photos (Docker) is required.
61 Please login to your server and authenticate. Please run -
62 `docker exec -it icloud /bin/sh -c "su-exec abc icloud --session-directory=/config/session_data {region_opt}--username={username}"`.""" # noqa: E501
63 subject = f"icloud-docker: iCloud login required for {username}"
64 return message, subject
67def _create_trust_expiring_message(
68 username: str,
69 days_remaining: int,
70 dashboard_url: str | None = None,
71) -> tuple[str, str]:
72 """
73 Create the trust-expiring notification message and subject.
75 Fires once per cookie lifetime when ``days_remaining`` first drops
76 below ``app.trust_expiry_warn_days`` so the user can refresh trust
77 BEFORE the sync loop hits a failed-auth state.
79 Args:
80 username: The iCloud username whose trust window is expiring
81 days_remaining: Days until ``X-APPLE-WEBAUTH-HSA-TRUST`` expires
82 dashboard_url: Web UI URL (optional). When provided, the message
83 tells the user to tap the URL to refresh trust without
84 retyping their password.
86 Returns:
87 Tuple of (message, subject)
88 """
89 horizon = (
90 "today"
91 if days_remaining <= 0
92 else f"in {days_remaining} day{'s' if days_remaining != 1 else ''}"
93 )
94 if dashboard_url:
95 message = (
96 f"icloud-docker: iCloud login expires {horizon}, refresh at {dashboard_url}"
97 )
98 else:
99 message = (
100 f"icloud-docker: iCloud login expires {horizon}. "
101 f"Sign in to the container to refresh before the next sync fails."
102 )
103 subject = f"icloud-docker: iCloud login for {username} expires {horizon}"
104 return message, subject
107def _get_current_timestamp() -> datetime.datetime:
108 """
109 Get the current timestamp for notification tracking.
111 Returns:
112 Current datetime
113 """
114 return datetime.datetime.now()
117def _get_telegram_config(config) -> tuple[str | None, str | None, bool]:
118 """
119 Extract Telegram configuration from config.
121 Args:
122 config: The configuration dictionary
124 Returns:
125 Tuple of (bot_token, chat_id, is_configured)
126 """
127 bot_token = config_parser.get_telegram_bot_token(config=config)
128 chat_id = config_parser.get_telegram_chat_id(config=config)
129 is_configured = bool(bot_token and chat_id)
130 return bot_token, chat_id, is_configured
133def notify_telegram(config, message, last_send=None, dry_run=False):
134 """
135 Send Telegram notification with throttling and error handling.
137 Args:
138 config: Configuration dictionary
139 message: Message to send
140 last_send: Timestamp of last send for throttling
141 dry_run: If True, don't actually send the message
143 Returns:
144 Timestamp when message was sent, or last_send if throttled, or None if failed
145 """
146 if _is_throttled(last_send):
147 LOGGER.info("Throttling telegram to once a day")
148 return last_send
150 bot_token, chat_id, is_configured = _get_telegram_config(config)
151 if not is_configured:
152 LOGGER.warning("Not sending 2FA notification because Telegram is not configured.")
153 return None
155 sent_on = _get_current_timestamp()
156 if dry_run:
157 return sent_on
159 # bot_token and chat_id are guaranteed to be non-None due to is_configured check
160 if post_message_to_telegram(bot_token, chat_id, message): # type: ignore[arg-type]
161 return sent_on
162 return None
165def post_message_to_telegram(bot_token: str, chat_id: str, message: str) -> bool:
166 """
167 Post message to Telegram bot using API.
169 Args:
170 bot_token: Telegram bot token
171 chat_id: Telegram chat ID
172 message: Message to send
174 Returns:
175 True if message was sent successfully, False otherwise
176 """
177 url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
178 params = {"chat_id": chat_id, "text": message}
179 response = requests.post(url, params=params, timeout=10)
180 if response.status_code == 200:
181 return True
182 # Log error message
183 LOGGER.error(f"Failed to send telegram notification. Response: {response.text}")
184 return False
187def _get_discord_config(config) -> tuple[str | None, str | None, bool]:
188 """
189 Extract Discord configuration from config.
191 Args:
192 config: The configuration dictionary
194 Returns:
195 Tuple of (webhook_url, username, is_configured)
196 """
197 webhook_url = config_parser.get_discord_webhook_url(config=config)
198 username = config_parser.get_discord_username(config=config)
199 is_configured = bool(webhook_url and username)
200 return webhook_url, username, is_configured
203def post_message_to_discord(webhook_url: str, username: str, message: str) -> bool:
204 """
205 Post message to Discord webhook.
207 Args:
208 webhook_url: Discord webhook URL
209 username: Username to display in Discord
210 message: Message to send
212 Returns:
213 True if message was sent successfully, False otherwise
214 """
215 data = {"username": username, "content": message}
216 response = requests.post(webhook_url, data=data, timeout=10)
217 if response.status_code == 204:
218 return True
219 # Log error message
220 LOGGER.error(f"Failed to send Discord notification. Response: {response.text}")
221 return False
224def notify_discord(config, message, last_send=None, dry_run=False):
225 """
226 Send Discord notification with throttling and error handling.
228 Args:
229 config: Configuration dictionary
230 message: Message to send
231 last_send: Timestamp of last send for throttling
232 dry_run: If True, don't actually send the message
234 Returns:
235 Timestamp when message was sent, or last_send if throttled, or None if failed
236 """
237 if _is_throttled(last_send):
238 LOGGER.info("Throttling discord to once a day")
239 return last_send
241 webhook_url, username, is_configured = _get_discord_config(config)
242 if not is_configured:
243 LOGGER.warning("Not sending 2FA notification because Discord is not configured.")
244 return None
246 sent_on = _get_current_timestamp()
247 if dry_run or post_message_to_discord(webhook_url, username, message): # type: ignore[arg-type]
248 return sent_on
249 return None
252def _get_pushover_config(config) -> tuple[str | None, str | None, int | None, bool]:
253 """
254 Extract Pushover configuration from config.
256 Args:
257 config: The configuration dictionary
259 Returns:
260 Tuple of (user_key, api_token, priority, is_configured)
261 """
262 user_key = config_parser.get_pushover_user_key(config=config)
263 api_token = config_parser.get_pushover_api_token(config=config)
264 priority = config_parser.get_pushover_notification_priority(config=config)
265 is_configured = bool(user_key and api_token)
266 return user_key, api_token, priority, is_configured
269def post_message_to_pushover(api_token: str, user_key: str, priority: int | None, message: str) -> bool:
270 """
271 Post message to Pushover API.
273 Args:
274 api_token: Pushover API token
275 user_key: Pushover user key
276 priority: Pushover notification priority (-2 to 2, optional)
277 message: Message to send
279 Returns:
280 True if message was sent successfully, False otherwise
281 """
282 url = "https://api.pushover.net/1/messages.json"
283 data = {"token": api_token, "user": user_key, "message": message}
284 if priority is not None:
285 data["priority"] = priority
286 response = requests.post(url, data=data, timeout=10)
287 if response.status_code == 200:
288 return True
289 LOGGER.error(f"Failed to send Pushover notification. Response: {response.text}")
290 return False
293def notify_pushover(config, message, last_send=None, dry_run=False):
294 """
295 Send Pushover notification with throttling and error handling.
297 Args:
298 config: Configuration dictionary
299 message: Message to send
300 last_send: Timestamp of last send for throttling
301 dry_run: If True, don't actually send the message
303 Returns:
304 Timestamp when message was sent, or last_send if throttled, or None if failed
305 """
306 if _is_throttled(last_send):
307 LOGGER.info("Throttling Pushover to once a day")
308 return last_send
310 user_key, api_token, priority, is_configured = _get_pushover_config(config)
311 if not is_configured:
312 LOGGER.warning("Not sending 2FA notification because Pushover is not configured.")
313 return None
315 sent_on = _get_current_timestamp()
316 if dry_run:
317 return sent_on
319 # user_key and api_token are guaranteed to be non-None due to is_configured check
320 if post_message_to_pushover(api_token, user_key, priority, message): # type: ignore[arg-type]
321 return sent_on
322 return None
325def notify_email(config, message: str, subject: str, last_send=None, dry_run=False):
326 """
327 Send email notification with throttling and error handling.
329 Args:
330 config: Configuration dictionary
331 message: Message to send
332 subject: Email subject
333 last_send: Timestamp of last send for throttling
334 dry_run: If True, don't actually send the message
336 Returns:
337 Timestamp when message was sent, or last_send if throttled, or None if failed
338 """
339 if _is_throttled(last_send):
340 LOGGER.info("Throttling email to once a day")
341 return last_send
343 email, to_email, host, port, no_tls, username, password, is_configured = (
344 _get_smtp_config(config)
345 )
346 if not is_configured:
347 LOGGER.warning("Not sending 2FA notification because SMTP is not configured")
348 return None
350 sent_on = _get_current_timestamp()
351 if dry_run:
352 return sent_on
354 try:
355 # All necessary config values are guaranteed to be non-None due to is_configured check
356 smtp = _create_smtp_connection(host, port, no_tls) # type: ignore[arg-type]
358 if password:
359 _authenticate_smtp(smtp, email, username, password) # type: ignore[arg-type]
361 # to_email could be None, use email as fallback
362 recipient = to_email if to_email else email
363 msg = build_message(email, recipient, message, subject) # type: ignore[arg-type]
364 _send_email_message(smtp, email, recipient, msg) # type: ignore[arg-type]
365 smtp.quit()
366 return sent_on
367 except Exception as e:
368 LOGGER.error(f"Failed to send email: {e!s}.")
369 return None
372def send(
373 config,
374 username,
375 last_send=None,
376 dry_run=False,
377 region="global",
378 dashboard_url=None,
379):
380 """
381 Send 2FA notification to all configured notification services.
383 Args:
384 config: Configuration dictionary
385 username: iCloud username requiring 2FA
386 last_send: Timestamp of last send for throttling
387 dry_run: If True, don't actually send notifications
388 region: iCloud region (default: "global")
389 dashboard_url: When set, message body links the user to this URL
390 (web UI ``/auth``) instead of the docker-exec command. Caller
391 should resolve from ``app.web_ui.public_url`` (or the
392 host:port fallback) when ``app.web_ui.enabled`` is true.
394 Returns:
395 Timestamp when notifications were sent, or None if all failed
396 """
397 message, subject = _create_2fa_message(
398 username,
399 region,
400 dashboard_url=dashboard_url,
401 )
403 # Send to all notification services
404 telegram_sent = notify_telegram(
405 config=config,
406 message=message,
407 last_send=last_send,
408 dry_run=dry_run,
409 )
410 discord_sent = notify_discord(
411 config=config,
412 message=message,
413 last_send=last_send,
414 dry_run=dry_run,
415 )
416 pushover_sent = notify_pushover(
417 config=config,
418 message=message,
419 last_send=last_send,
420 dry_run=dry_run,
421 )
422 email_sent = notify_email(
423 config=config,
424 message=message,
425 subject=subject,
426 last_send=last_send,
427 dry_run=dry_run,
428 )
430 # Return the timestamp if any notification was sent successfully
431 sent_timestamps = [
432 t
433 for t in [telegram_sent, discord_sent, pushover_sent, email_sent]
434 if t is not None
435 ]
436 return sent_timestamps[0] if sent_timestamps else None
439def send_trust_expiring(
440 config,
441 username,
442 days_remaining,
443 last_send=None,
444 dry_run=False,
445 dashboard_url=None,
446):
447 """Send trust-expiring notification to all configured services.
449 Mirrors ``send()`` but for the pre-emptive "trust window closing"
450 event. Caller is responsible for the once-per-cookie-lifetime
451 debounce (see ``web_signals.get_trust_state`` /
452 ``record_trust_state``); the ``last_send`` arg here is the
453 daily-throttle from ``_is_throttled`` shared with the 2FA flow,
454 not the per-cookie one.
456 Args:
457 config: Configuration dictionary
458 username: iCloud username whose trust window is expiring
459 days_remaining: Days until ``X-APPLE-WEBAUTH-HSA-TRUST`` expires
460 last_send: Timestamp of last send for daily throttling
461 dry_run: If True, don't actually send notifications
462 dashboard_url: When set, message body links the user to this
463 URL (web UI dashboard) instead of the docker-exec
464 instruction. Caller resolves from ``app.web_ui.public_url``
465 (or the host:port fallback) when web UI is enabled.
467 Returns:
468 Timestamp when notifications were sent, or None if all failed.
469 """
470 message, subject = _create_trust_expiring_message(
471 username,
472 days_remaining,
473 dashboard_url=dashboard_url,
474 )
475 telegram_sent = notify_telegram(
476 config=config,
477 message=message,
478 last_send=last_send,
479 dry_run=dry_run,
480 )
481 discord_sent = notify_discord(
482 config=config,
483 message=message,
484 last_send=last_send,
485 dry_run=dry_run,
486 )
487 pushover_sent = notify_pushover(
488 config=config,
489 message=message,
490 last_send=last_send,
491 dry_run=dry_run,
492 )
493 email_sent = notify_email(
494 config=config,
495 message=message,
496 subject=subject,
497 last_send=last_send,
498 dry_run=dry_run,
499 )
500 sent_timestamps = [
501 t
502 for t in [telegram_sent, discord_sent, pushover_sent, email_sent]
503 if t is not None
504 ]
505 return sent_timestamps[0] if sent_timestamps else None
508def _get_smtp_config(
509 config,
510) -> tuple[str | None, str | None, str | None, int | None, bool, str | None, str | None, bool]:
511 """
512 Extract SMTP configuration from config.
514 Args:
515 config: The configuration dictionary
517 Returns:
518 Tuple of (email, to_email, host, port, no_tls, username, password, is_configured)
519 """
520 email = config_parser.get_smtp_email(config=config)
521 to_email = config_parser.get_smtp_to_email(config=config)
522 host = config_parser.get_smtp_host(config=config)
523 port = config_parser.get_smtp_port(config=config)
524 no_tls = config_parser.get_smtp_no_tls(config=config)
525 username = config_parser.get_smtp_username(config=config)
526 password = config_parser.get_smtp_password(config=config)
527 is_configured = bool(email and host and port)
528 return email, to_email, host, port, no_tls, username, password, is_configured
531def _create_smtp_connection(host: str, port: int, no_tls: bool) -> smtplib.SMTP:
532 """
533 Create and configure SMTP connection.
535 Args:
536 host: SMTP host
537 port: SMTP port
538 no_tls: Whether to skip TLS
540 Returns:
541 Configured SMTP connection
542 """
543 smtp = smtplib.SMTP(host, port)
544 smtp.set_debuglevel(0)
545 smtp.connect(host, port)
546 if not no_tls:
547 smtp.starttls()
548 return smtp
551def _authenticate_smtp(smtp: smtplib.SMTP, email: str, username: str | None, password: str) -> None:
552 """
553 Authenticate SMTP connection.
555 Args:
556 smtp: SMTP connection
557 email: Email address for fallback authentication
558 username: SMTP username (optional)
559 password: SMTP password
560 """
561 if username:
562 smtp.login(username, password)
563 else:
564 smtp.login(email, password)
567def _send_email_message(smtp: smtplib.SMTP, email: str, to_email: str, message_obj: Message) -> None:
568 """
569 Send email message through SMTP connection.
571 Args:
572 smtp: SMTP connection
573 email: From email address
574 to_email: To email address
575 message_obj: Email message object
576 """
577 smtp.sendmail(from_addr=email, to_addrs=to_email, msg=message_obj.as_string())
580def _contains_non_ascii(text: str | None) -> bool:
581 """Determine if the provided text contains non-ASCII characters."""
583 if text is None:
584 return False
586 try:
587 text.encode("ascii")
588 except UnicodeEncodeError:
589 return True
590 return False
593def build_message(email: str, to_email: str, message: str, subject: str) -> Message:
594 """
595 Create email message with proper headers.
597 Args:
598 email: From email address
599 to_email: To email address
600 message: Message body
601 subject: Message subject
603 Returns:
604 Configured email message object
605 """
606 requires_utf8 = _contains_non_ascii(message) or _contains_non_ascii(subject)
607 charset = "utf-8" if requires_utf8 else "us-ascii"
609 msg = Message(to=to_email, charset=charset)
610 msg.sender = "icloud-docker <" + email + ">"
611 msg.date = datetime.datetime.now().strftime("%d/%m/%Y %H:%M")
612 msg.subject = subject
613 msg.body = message
614 return msg
617# =============================================================================
618# Sync Summary Notification Functions
619# =============================================================================
622def _format_sync_summary_message(summary) -> tuple[str, str]:
623 """
624 Format sync summary as notification message.
626 Args:
627 summary: SyncSummary object containing sync statistics
629 Returns:
630 Tuple of (message, subject)
631 """
632 from src.sync_stats import format_bytes, format_duration
634 has_errors = summary.has_errors()
635 status_emoji = "⚠️" if has_errors else "✅"
636 status_text = "Completed with Errors" if has_errors else "Complete"
638 message_lines = [f"{status_emoji} iCloud Sync {status_text}", ""]
640 # Drive statistics
641 if summary.drive_stats and summary.drive_stats.has_activity():
642 drive = summary.drive_stats
643 message_lines.append("📁 Drive:")
644 if drive.files_downloaded > 0:
645 size_str = format_bytes(drive.bytes_downloaded)
646 message_lines.append(f" • Downloaded: {drive.files_downloaded} files ({size_str})")
647 if drive.files_skipped > 0:
648 message_lines.append(f" • Skipped: {drive.files_skipped} files (up-to-date)")
649 if drive.files_removed > 0:
650 message_lines.append(f" • Removed: {drive.files_removed} obsolete files")
651 if drive.duration_seconds > 0:
652 duration_str = format_duration(drive.duration_seconds)
653 message_lines.append(f" • Duration: {duration_str}")
654 if drive.has_errors():
655 message_lines.append(f" • Errors: {len(drive.errors)} failed")
656 message_lines.append("")
658 # Photos statistics
659 if summary.photo_stats and summary.photo_stats.has_activity():
660 photos = summary.photo_stats
661 message_lines.append("📷 Photos:")
662 if photos.photos_downloaded > 0:
663 size_str = format_bytes(photos.bytes_downloaded)
664 message_lines.append(f" • Downloaded: {photos.photos_downloaded} photos ({size_str})")
665 if photos.photos_hardlinked > 0:
666 message_lines.append(f" • Hard-linked: {photos.photos_hardlinked} photos")
667 if photos.bytes_saved_by_hardlinks > 0:
668 saved_str = format_bytes(photos.bytes_saved_by_hardlinks)
669 message_lines.append(f" • Storage saved: {saved_str}")
670 if photos.albums_synced:
671 albums_str = ", ".join(photos.albums_synced[:5])
672 if len(photos.albums_synced) > 5:
673 albums_str += f" (+{len(photos.albums_synced) - 5} more)"
674 message_lines.append(f" • Albums: {albums_str}")
675 if photos.duration_seconds > 0:
676 duration_str = format_duration(photos.duration_seconds)
677 message_lines.append(f" • Duration: {duration_str}")
678 if photos.has_errors():
679 message_lines.append(f" • Errors: {len(photos.errors)} failed")
680 message_lines.append("")
682 # Error details if present
683 if has_errors:
684 message_lines.append("Failed items:")
685 all_errors = []
686 if summary.drive_stats:
687 all_errors.extend(summary.drive_stats.errors[:5]) # Limit to first 5
688 if summary.photo_stats:
689 all_errors.extend(summary.photo_stats.errors[:5]) # Limit to first 5
690 message_lines.extend([f" • {error}" for error in all_errors[:10]])
691 total_errors = 0
692 if summary.drive_stats:
693 total_errors += len(summary.drive_stats.errors)
694 if summary.photo_stats:
695 total_errors += len(summary.photo_stats.errors)
696 if total_errors > 10:
697 message_lines.append(f" ... and {total_errors - 10} more errors")
698 message_lines.append("")
700 message = "\n".join(message_lines)
701 subject = f"icloud-docker: Sync {status_text}"
702 return message, subject
705def _should_send_sync_summary(config, summary) -> bool:
706 """
707 Determine if sync summary notification should be sent.
709 Args:
710 config: Configuration dictionary
711 summary: SyncSummary object
713 Returns:
714 True if notification should be sent, False otherwise
715 """
716 # Check if sync summary is enabled
717 if not config_parser.get_sync_summary_enabled(config=config):
718 return False
720 # Check if there was any activity
721 if not summary.has_activity():
722 return False
724 # Check error/success preferences
725 has_errors = summary.has_errors()
726 on_error = config_parser.get_sync_summary_on_error(config=config)
727 on_success = config_parser.get_sync_summary_on_success(config=config)
729 if has_errors and not on_error:
730 return False
731 if not has_errors and not on_success:
732 return False
734 # Check minimum downloads threshold
735 min_downloads = config_parser.get_sync_summary_min_downloads(config=config)
736 total_downloads = 0
737 if summary.drive_stats:
738 total_downloads += summary.drive_stats.files_downloaded
739 if summary.photo_stats:
740 total_downloads += summary.photo_stats.photos_downloaded
742 if total_downloads < min_downloads:
743 return False
745 return True
748def send_sync_summary(config, summary, dry_run=False):
749 """
750 Send sync summary notification to all configured services.
752 Note: Sync summaries are NOT throttled like 2FA notifications,
753 as they provide valuable operational information for each sync.
755 Args:
756 config: Configuration dictionary
757 summary: SyncSummary object containing sync statistics
758 dry_run: If True, don't actually send notifications
760 Returns:
761 True if at least one notification was sent successfully, False otherwise
762 """
763 if not _should_send_sync_summary(config, summary):
764 LOGGER.debug("Sync summary notification skipped (not enabled or no activity)")
765 return False
767 message, subject = _format_sync_summary_message(summary)
769 # Send to all notification services (no throttling for sync summaries)
770 telegram_sent = _send_telegram_no_throttle(config, message, dry_run)
771 discord_sent = _send_discord_no_throttle(config, message, dry_run)
772 pushover_sent = _send_pushover_no_throttle(config, message, dry_run)
773 email_sent = _send_email_no_throttle(config, message, subject, dry_run)
775 # Return True if any notification was sent successfully
776 any_sent = any([telegram_sent, discord_sent, pushover_sent, email_sent])
777 if any_sent:
778 LOGGER.info("Sync summary notification sent successfully")
779 return any_sent
782def _send_telegram_no_throttle(config, message: str, dry_run: bool) -> bool:
783 """Send Telegram notification without throttling.
785 Args:
786 config: Configuration dictionary
787 message: Message to send
788 dry_run: If True, don't actually send
790 Returns:
791 True if sent successfully, False otherwise
792 """
793 bot_token, chat_id, is_configured = _get_telegram_config(config)
794 if not is_configured:
795 return False
797 if dry_run:
798 return True
800 return post_message_to_telegram(bot_token, chat_id, message)
803def _send_discord_no_throttle(config, message: str, dry_run: bool) -> bool:
804 """Send Discord notification without throttling.
806 Args:
807 config: Configuration dictionary
808 message: Message to send
809 dry_run: If True, don't actually send
811 Returns:
812 True if sent successfully, False otherwise
813 """
814 webhook_url, username, is_configured = _get_discord_config(config)
815 if not is_configured:
816 return False
818 if dry_run:
819 return True
821 return post_message_to_discord(webhook_url, username, message)
824def _send_pushover_no_throttle(config, message: str, dry_run: bool) -> bool:
825 """Send Pushover notification without throttling.
827 Args:
828 config: Configuration dictionary
829 message: Message to send
830 dry_run: If True, don't actually send
832 Returns:
833 True if sent successfully, False otherwise
834 """
835 user_key, api_token, priority, is_configured = _get_pushover_config(config)
836 if not is_configured:
837 return False
839 if dry_run:
840 return True
842 return post_message_to_pushover(api_token, user_key, priority, message)
845def _send_email_no_throttle(config, message: str, subject: str, dry_run: bool) -> bool:
846 """Send email notification without throttling.
848 Args:
849 config: Configuration dictionary
850 message: Message to send
851 subject: Email subject
852 dry_run: If True, don't actually send
854 Returns:
855 True if sent successfully, False otherwise
856 """
857 email, to_email, host, port, no_tls, username, password, is_configured = (
858 _get_smtp_config(config)
859 )
860 if not is_configured:
861 return False
863 if dry_run:
864 return True
866 try:
867 smtp = _create_smtp_connection(host, port, no_tls) # type: ignore[arg-type]
869 if password:
870 _authenticate_smtp(smtp, email, username, password) # type: ignore[arg-type]
872 recipient = to_email if to_email else email
873 msg = build_message(email, recipient, message, subject) # type: ignore[arg-type]
874 _send_email_message(smtp, email, recipient, msg) # type: ignore[arg-type]
875 smtp.quit()
876 return True
877 except Exception as e:
878 LOGGER.error(f"Failed to send sync summary email: {e!s}")
879 return False