Coverage for custom_components/supernotify/media_grab.py: 94%
346 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1from __future__ import annotations
3import asyncio
4import datetime as dt
5import io
6import logging
7import time
8from enum import StrEnum, auto
9from http import HTTPStatus
10from io import BytesIO
11from typing import TYPE_CHECKING, Any, cast
13import aiofiles
14import aiofiles.os
15import homeassistant.util.dt as dt_util
16from aiohttp import ClientResponse, ClientSession, ClientTimeout
17from anyio import Path
18from homeassistant.const import STATE_HOME, STATE_UNAVAILABLE
19from PIL import Image
21from custom_components.supernotify.const import (
22 ATTR_JPEG_OPTS,
23 ATTR_MEDIA_CAMERA_DELAY,
24 ATTR_MEDIA_CAMERA_ENTITY_ID,
25 ATTR_MEDIA_CAMERA_PTZ_PRESET,
26 ATTR_MEDIA_SNAPSHOT_PATH,
27 ATTR_MEDIA_SNAPSHOT_URL,
28 ATTR_PNG_OPTS,
29 CONF_ALT_CAMERA,
30 CONF_CAMERA,
31 CONF_DEVICE_TRACKER,
32 CONF_OPTIONS,
33 CONF_PTZ_CAMERA,
34 CONF_PTZ_DELAY,
35 CONF_PTZ_METHOD,
36 CONF_PTZ_PRESET_DEFAULT,
37 MEDIA_OPTION_REPROCESS,
38 OPTION_JPEG,
39 OPTION_PNG,
40 PLATFORM_FRIGATE,
41 PTZ_DELAY_DEFAULT,
42 PTZ_METHOD_FRIGATE,
43 PTZ_METHOD_ONVIF,
44)
46if TYPE_CHECKING:
47 from homeassistant.components.image import ImageEntity
48 from homeassistant.core import State
50 from .context import Context
51 from .hass_api import HomeAssistantAPI
53_LOGGER = logging.getLogger(__name__)
56class ReprocessOption(StrEnum):
57 ALWAYS = auto()
58 NEVER = auto()
59 PRESERVE = auto()
62async def snapshot_from_url(
63 hass_api: HomeAssistantAPI,
64 snapshot_url: str,
65 notification_id: str,
66 media_path: Path,
67 hass_base_url: str | None,
68 remote_timeout: int = 15,
69) -> Path | None:
70 """Download a snapshot URL and save raw bytes. No reprocessing."""
71 hass_base_url = hass_base_url or ""
72 try:
73 raw_dir: Path = Path(media_path) / "raw"
74 await raw_dir.mkdir(parents=True, exist_ok=True)
76 image_url = snapshot_url if snapshot_url.startswith("http") else f"{hass_base_url}{snapshot_url}"
77 websession: ClientSession = hass_api.http_session()
78 r: ClientResponse = await websession.get(image_url, timeout=ClientTimeout(total=remote_timeout))
79 if r.status != HTTPStatus.OK:
80 _LOGGER.warning("SUPERNOTIFY Unable to retrieve %s: %s", image_url, r.status)
81 else:
82 bitmap: bytes | None = await r.content.read()
83 if bitmap:
84 ext = await _detect_image_ext(hass_api, bitmap)
85 raw_path: Path = raw_dir / f"{notification_id}.{ext}"
86 async with aiofiles.open(raw_path, "wb") as f:
87 await f.write(bitmap)
88 _LOGGER.debug("SUPERNOTIFY Fetched raw image from %s to %s", image_url, raw_path)
89 return raw_path
91 _LOGGER.warning("SUPERNOTIFY Failed to snap image from %s", snapshot_url)
92 except Exception:
93 _LOGGER.exception("SUPERNOTIFY Image snap fail")
95 return None
98def infer_ptz_method(hass_api: HomeAssistantAPI, camera_entity_id: str) -> str:
99 """Guess a PTZ method from the integration that owns the camera entity.
101 Used only as a fallback for cameras with no entry in the cameras: config, where
102 there's no explicit ptz_method to consult.
103 """
104 ent_reg = hass_api.entity_registry()
105 reg_entry = ent_reg.async_get(camera_entity_id) if ent_reg else None
106 if reg_entry and reg_entry.platform == PLATFORM_FRIGATE:
107 return PTZ_METHOD_FRIGATE
108 return PTZ_METHOD_ONVIF
111async def move_camera_to_ptz_preset(
112 hass_api: HomeAssistantAPI, camera_entity_id: str, preset: str | int, method: str = PTZ_METHOD_ONVIF
113) -> None:
114 try:
115 _LOGGER.info("SUPERNOTIFY Executing PTZ by %s to %s for %s", method, preset, camera_entity_id)
116 if method == PTZ_METHOD_FRIGATE:
117 await hass_api.call_service(
118 "frigate",
119 "ptz",
120 service_data={"action": "preset", "argument": preset},
121 target={"entity_id": camera_entity_id},
122 return_response=False,
123 blocking=True,
124 )
126 elif method == PTZ_METHOD_ONVIF:
127 await hass_api.call_service(
128 "onvif",
129 "ptz",
130 service_data={"move_mode": "GotoPreset", "preset": preset},
131 target={"entity_id": camera_entity_id},
132 return_response=False,
133 blocking=True,
134 )
135 else:
136 _LOGGER.warning("SUPERNOTIFY Unknown PTZ method %s", method)
137 except Exception as e:
138 _LOGGER.warning("SUPERNOTIFY Unable to move %s to ptz preset %s: %s", camera_entity_id, preset, e)
141async def snap_image_entity(
142 hass_api: HomeAssistantAPI,
143 entity_id: str,
144 media_path: Path,
145 notification_id: str,
146) -> Path | None:
147 """Read an image entity and save raw bytes. No reprocessing."""
148 raw_path: Path | None = None
149 try:
150 image_entity: ImageEntity | None = cast("ImageEntity|None", hass_api.domain_entity("image", entity_id))
151 if image_entity:
152 bitmap: bytes | None = await image_entity.async_image()
153 if bitmap:
154 raw_dir: Path = Path(media_path) / "raw"
155 await raw_dir.mkdir(parents=True, exist_ok=True)
156 ext = await _detect_image_ext(hass_api, bitmap)
157 raw_path = raw_dir / f"{notification_id}.{ext}"
158 async with aiofiles.open(raw_path, "wb") as f:
159 await f.write(bitmap)
160 except Exception as e:
161 _LOGGER.warning("SUPERNOTIFY Unable to snap image %s: %s", entity_id, e)
162 if raw_path is None:
163 _LOGGER.warning("SUPERNOTIFY Unable to save from image entity %s", entity_id)
164 return raw_path
167async def snap_camera(
168 hass_api: HomeAssistantAPI,
169 camera_entity_id: str,
170 notification_id: str,
171 media_path: Path,
172 max_camera_wait: int = 20,
173) -> Path | None:
174 """Snap a camera and save the raw image. No reprocessing."""
175 if not camera_entity_id:
176 _LOGGER.warning("SUPERNOTIFY Empty camera entity id for snap")
177 return None
179 raw_path: Path | None = None
180 try:
181 raw_dir: Path = Path(media_path) / "raw"
182 await raw_dir.mkdir(parents=True, exist_ok=True)
183 raw_path = raw_dir / f"{notification_id}.jpg"
184 share_root = Path(media_path)
185 share_path = share_root / raw_path.relative_to(Path(media_path))
186 await hass_api.call_service(
187 "camera",
188 "snapshot",
189 service_data={"entity_id": camera_entity_id, "filename": str(share_path)},
190 return_response=False,
191 blocking=True,
192 )
194 cutoff_time = time.time() + max_camera_wait
195 while time.time() < cutoff_time and not await raw_path.exists():
196 _LOGGER.info("Image file not available yet at %s, pausing", raw_path)
197 await asyncio.sleep(1)
199 except Exception as e:
200 _LOGGER.warning("Failed to snap avail camera %s to %s: %s", camera_entity_id, raw_path, e)
201 raw_path = None
203 return raw_path
206def camera_available(hass_api: HomeAssistantAPI, camera_config: dict[str, Any], non_entity: bool = False) -> bool:
207 state: State | None = None
208 tracker_entity_id: str
209 camera_entity_id: str = camera_config[CONF_CAMERA]
210 try:
211 if camera_config.get(CONF_DEVICE_TRACKER):
212 tracker_entity_id = camera_config[CONF_DEVICE_TRACKER]
213 state = hass_api.get_state(camera_config[CONF_DEVICE_TRACKER])
214 if state and state.state == STATE_HOME:
215 return True
216 _LOGGER.debug("SUPERNOTIFY Skipping camera %s tracker %s state %s", camera_entity_id, tracker_entity_id, state)
217 else:
218 tracker_entity_id = camera_entity_id
219 state = hass_api.get_state(camera_entity_id)
220 if state and state.state != STATE_UNAVAILABLE:
221 return True
222 if state is None and non_entity:
223 return True
224 _LOGGER.debug("SUPERNOTIFY Skipping camera %s with state %s", camera_entity_id, state)
225 if state is None:
226 if tracker_entity_id == camera_entity_id:
227 _LOGGER.warning(
228 "SUPERNOTIFY Camera %s tracker %s has no entity state",
229 camera_entity_id,
230 tracker_entity_id,
231 )
232 else:
233 _LOGGER.warning(
234 "SUPERNOTIFY Camera %s device_tracker %s seems missing",
235 camera_entity_id,
236 camera_config[CONF_DEVICE_TRACKER],
237 )
238 return False
240 except Exception:
241 _LOGGER.exception("SUPERNOTIFY Unable to determine camera state: %s", camera_config)
242 return False
245def select_avail_camera(hass_api: HomeAssistantAPI, cameras: dict[str, Any], camera_entity_id: str) -> str | None:
246 avail_camera_entity_id: str | None = None
248 preferred_cam = cameras.get(camera_entity_id)
249 # test support FIXME
250 if preferred_cam and CONF_CAMERA not in preferred_cam:
251 preferred_cam[CONF_CAMERA] = camera_entity_id
252 if preferred_cam is None:
253 # assume unconfigured camera available
254 return camera_entity_id
255 if camera_available(hass_api, preferred_cam):
256 return camera_entity_id
258 alt_cams: list[dict[str, Any]] = [cameras[c] for c in preferred_cam.get(CONF_ALT_CAMERA, []) if c in cameras]
259 alt_cams.extend(
260 {CONF_CAMERA: entity_id} for entity_id in preferred_cam.get(CONF_ALT_CAMERA, []) if entity_id not in cameras
261 )
262 for alt_cam in alt_cams:
263 if camera_available(hass_api, alt_cam):
264 _LOGGER.info("SUPERNOTIFY Selecting available camera %s rather than %s", alt_cam[CONF_CAMERA], camera_entity_id)
265 return alt_cam[CONF_CAMERA]
267 if avail_camera_entity_id is None:
268 _LOGGER.warning("%s not available, finding best alternative available", camera_entity_id)
269 if camera_available(hass_api, preferred_cam, non_entity=True):
270 _LOGGER.info("SUPERNOTIFY Selecting camera %s with no known entity", camera_entity_id)
271 return camera_entity_id
272 for alt_cam in alt_cams:
273 if camera_available(hass_api, alt_cam, non_entity=True):
274 _LOGGER.info(
275 "SUPERNOTIFY Selecting alt camera %s with no known entity for %s", alt_cam[CONF_CAMERA], camera_entity_id
276 )
277 return alt_cam[CONF_CAMERA]
279 return None
282async def _detect_image_ext(hass_api: HomeAssistantAPI, bitmap: bytes) -> str:
283 """Detect image format from raw bytes, returning a file extension."""
284 try:
285 img = await hass_api.create_job(Image.open, io.BytesIO(bitmap))
286 fmt = (img.format or "").lower()
287 return "jpg" if fmt in ("jpg", "jpeg") else fmt or "img"
288 except Exception as e:
289 _LOGGER.warning("SUPERNOTIFY unable to detect image, defaulting to 'img': %s", e)
290 return "img"
293async def snap_notification_image(notification: Notification, context: Context) -> Path | None: # type: ignore # noqa: F821
294 """Delivery-neutral image acquisition: PTZ movement, camera snap, URL fetch, or image entity.
296 Caches the raw image path on notification._raw_image_path. Safe to call multiple times;
297 subsequent calls return the cached path immediately.
298 """
299 if getattr(notification, "_raw_image_path", None) is not None:
300 return notification._raw_image_path # type: ignore[attr-defined]
302 if notification.media.get(ATTR_MEDIA_SNAPSHOT_PATH) is not None:
303 return Path(notification.media[ATTR_MEDIA_SNAPSHOT_PATH])
305 snapshot_url = notification.media.get(ATTR_MEDIA_SNAPSHOT_URL)
306 camera_entity_id = notification.media.get(ATTR_MEDIA_CAMERA_ENTITY_ID)
307 media_path: Path | None = context.media_storage.media_path
309 if not media_path or (not snapshot_url and not camera_entity_id):
310 return None
311 if not context.hass_api:
312 return None
314 raw_path: Path | None = None
315 if snapshot_url:
316 raw_path = await snapshot_from_url(
317 context.hass_api, snapshot_url, notification.id, media_path, context.hass_api.internal_url
318 )
319 elif camera_entity_id.startswith("image."):
320 raw_path = await snap_image_entity(context.hass_api, camera_entity_id, media_path, notification.id)
321 else:
322 active_camera_entity_id = select_avail_camera(context.hass_api, context.cameras, camera_entity_id)
323 if active_camera_entity_id:
324 camera_config = context.cameras.get(active_camera_entity_id, {})
325 camera_ptz_entity_id: str = camera_config.get(CONF_PTZ_CAMERA, active_camera_entity_id)
326 camera_delay = notification.media.get(ATTR_MEDIA_CAMERA_DELAY, camera_config.get(CONF_PTZ_DELAY))
327 camera_delay = PTZ_DELAY_DEFAULT if camera_delay is None else camera_delay
328 camera_ptz_preset_default = camera_config.get(CONF_PTZ_PRESET_DEFAULT)
329 camera_ptz_preset = notification.media.get(ATTR_MEDIA_CAMERA_PTZ_PRESET)
330 camera_ptz_method = camera_config.get(CONF_PTZ_METHOD)
331 if camera_ptz_method is None:
332 camera_ptz_method = infer_ptz_method(context.hass_api, camera_ptz_entity_id)
334 _LOGGER.debug(
335 "SUPERNOTIFY Snapping camera %s, ptz %s->%s (%s), delay %s secs",
336 active_camera_entity_id,
337 camera_ptz_preset,
338 camera_ptz_preset_default,
339 camera_ptz_entity_id,
340 camera_delay,
341 )
342 if camera_ptz_preset:
343 await move_camera_to_ptz_preset(
344 context.hass_api, camera_ptz_entity_id, camera_ptz_preset, method=camera_ptz_method
345 )
346 if camera_delay:
347 _LOGGER.debug("SUPERNOTIFY Waiting %s secs before snapping", camera_delay)
348 await asyncio.sleep(camera_delay)
349 raw_path = await snap_camera(
350 context.hass_api,
351 active_camera_entity_id,
352 notification.id,
353 media_path=media_path,
354 max_camera_wait=15,
355 )
356 if camera_ptz_preset and camera_ptz_preset_default:
357 await move_camera_to_ptz_preset(
358 context.hass_api, camera_ptz_entity_id, camera_ptz_preset_default, method=camera_ptz_method
359 )
361 if raw_path is None:
362 _LOGGER.warning("SUPERNOTIFY No media available to attach (%s,%s)", snapshot_url, camera_entity_id)
363 notification._raw_image_path = raw_path # type: ignore[attr-defined]
364 return raw_path
367async def grab_image(notification: Notification, delivery: Delivery, context: Context) -> Path | None: # type: ignore # noqa: F821
368 """Get a delivery-ready image, reprocessing the raw snap with delivery-specific settings.
370 The raw snap is cached on the notification; reprocessed variants are cached by filename
371 so multiple deliveries with the same settings share the processed file.
373 Filename convention:
374 raw/{nid}.{ext} — delivery-neutral camera output
375 image/{nid}.jpg — default reprocessing (ALWAYS, no extra opts)
376 image/{nid}_{hashed_opts}}.jpg — delivery-specific opts or non-ALWAYS reprocess mode
377 """
378 if notification.media.get(ATTR_MEDIA_SNAPSHOT_PATH) is not None:
379 return Path(notification.media[ATTR_MEDIA_SNAPSHOT_PATH])
381 raw_path = await snap_notification_image(notification, context)
382 if raw_path is None:
383 return None
385 media_path: Path | None = context.media_storage.media_path
386 if not media_path or not context.hass_api:
387 return None
389 delivery_config = notification.delivery_data(delivery)
390 jpeg_opts = notification.media.get(ATTR_JPEG_OPTS, delivery_config.get(CONF_OPTIONS, {}).get(OPTION_JPEG))
391 png_opts = notification.media.get(ATTR_PNG_OPTS, delivery_config.get(CONF_OPTIONS, {}).get(OPTION_PNG))
392 reprocess_option = (
393 notification.media.get(MEDIA_OPTION_REPROCESS, delivery_config.get(CONF_OPTIONS, {}).get(MEDIA_OPTION_REPROCESS))
394 or "always"
395 )
396 reprocess: ReprocessOption = ReprocessOption.ALWAYS
397 try:
398 reprocess = ReprocessOption(reprocess_option)
399 except Exception:
400 _LOGGER.warning("SUPERNOTIFY Invalid reprocess option: %s", reprocess_option)
402 if reprocess == ReprocessOption.NEVER:
403 return raw_path
405 raw_ext = raw_path.suffix.lstrip(".").lower()
406 relevant_opts: dict[str, Any] = jpeg_opts if raw_ext in ("jpg", "jpeg") else png_opts if raw_ext == "png" else {}
407 is_default = not relevant_opts and reprocess == ReprocessOption.ALWAYS
408 if is_default:
409 processed_name = f"{notification.id}.jpg"
410 else:
411 key = hex(hash((reprocess_option, *tuple(relevant_opts.values()))))[-12:]
412 processed_name = f"{notification.id}_{key}.jpg"
413 processed_path = Path(media_path) / "image" / processed_name
415 if await processed_path.exists():
416 return await processed_path.resolve()
418 async with await raw_path.open("rb") as f:
419 bitmap: bytes = await f.read()
420 return await write_image_from_bitmap(
421 context.hass_api, bitmap, processed_path, reprocess=reprocess, jpeg_opts=jpeg_opts, png_opts=png_opts
422 )
425async def write_image_from_bitmap(
426 hass_api: HomeAssistantAPI,
427 bitmap: bytes | None,
428 output_path: Path,
429 reprocess: ReprocessOption = ReprocessOption.ALWAYS,
430 output_format: str | None = None,
431 jpeg_opts: dict[str, Any] | None = None,
432 png_opts: dict[str, Any] | None = None,
433) -> Path | None:
434 """Reprocess a raw image bitmap and write to an explicit output path."""
435 if bitmap is None:
436 _LOGGER.debug("SUPERNOTIFY Empty bitmap for image")
437 return None
438 input_format: str = "img"
439 try:
440 await output_path.parent.mkdir(parents=True, exist_ok=True)
442 image = await hass_api.create_job(Image.open, io.BytesIO(bitmap))
444 input_format = image.format.lower() if image.format else "img"
445 if reprocess == ReprocessOption.ALWAYS:
446 # rewrite to remove metadata, incl custom CCTV comments that confuse python MIMEImage
447 clean_image: Image.Image = await hass_api.create_job(Image.new, image.mode, image.size)
448 # Pillow API changed in 12.1.0 and the original call will be removed in 2027
449 # https://pillow.readthedocs.io/en/stable/releasenotes/12.1.0.html#image-getdata
450 if hasattr(image, "get_flattened_data"):
451 clean_image.putdata(image.get_flattened_data()) # added in jan 2026
452 else:
453 clean_image.putdata(image.getdata()) # being removed in 2027
455 image = clean_image
457 buffer = BytesIO()
458 img_args: dict[str, Any] = {}
459 if reprocess in (ReprocessOption.ALWAYS, ReprocessOption.PRESERVE):
460 if input_format in ("jpg", "jpeg") and jpeg_opts:
461 img_args.update(jpeg_opts)
462 elif input_format == "png" and png_opts:
463 img_args.update(png_opts)
465 image.save(buffer, output_format or input_format, **img_args)
467 output_path = await output_path.resolve()
468 async with aiofiles.open(output_path, "wb") as file:
469 await file.write(buffer.getbuffer())
470 return output_path
471 except TypeError:
472 # probably a jpeg or png option
473 _LOGGER.exception("SUPERNOTIFY Image snap fail")
474 except Exception:
475 _LOGGER.exception("SUPERNOTIFY Failure saving %s bitmap", input_format)
476 return None
479class MediaStorage:
480 def __init__(
481 self,
482 media_path: str | None,
483 media_url_prefix: str | None = None,
484 days: int = 7,
485 ) -> None:
486 self.media_path: Path | None = Path(media_path) if media_path else None
487 self.last_purge: dt.datetime | None = None
488 self.media_url_prefix = media_url_prefix
489 self.purge_minute_interval = 60 * 6
490 self.days = days
492 async def initialize(self, hass_api: HomeAssistantAPI) -> None:
493 self.hass_api = hass_api # TODO: should not be set on initialize
494 if self.media_path is not None and not self.media_path.is_absolute():
495 self.media_path = await self.media_path.absolute()
496 _LOGGER.debug("SUPERNOTIFY Media path updated to %s", self.media_path)
497 if self.media_path and not await self.media_path.exists():
498 _LOGGER.info("SUPERNOTIFY Media path not found at %s, attempting to create.", self.media_path)
499 try:
500 await self.media_path.mkdir(parents=True, exist_ok=True)
501 except Exception as e:
502 _LOGGER.warning("SUPERNOTIFY Media path %s cannot be created: %s", self.media_path, e)
503 hass_api.raise_issue(
504 "media_path",
505 "media_path",
506 {"path": str(self.media_path), "error": str(e)},
507 learn_more_url="https://supernotify.rhizomatics.org.uk/#getting-started",
508 )
509 self.media_path = None
510 if self.media_path is not None:
511 _LOGGER.debug("SUPERNOTIFY Abs media path: %s", await self.media_path.absolute())
513 if self.media_url_prefix is not None and self.media_path is not None:
514 if await hass_api.register_web_path(self.media_path, self.media_url_prefix):
515 _LOGGER.info("SUPERNOTIFY Media at %s available with prefixed URL %s", self.media_path, self.media_url_prefix)
516 else:
517 self.media_url_prefix = None
519 async def object_url(self, relative_path: Path) -> str | None:
520 """Convert a local image path to an externally accessible URL via the registered static path."""
521 if self.media_url_prefix is None or self.media_path is None:
522 _LOGGER.debug("SUPERNOTIFY Unable to generate object_url for %s", relative_path)
523 return None
524 try:
525 relative = relative_path.relative_to(await self.media_path.absolute())
526 return self.hass_api.abs_url(f"{self.media_url_prefix}/{relative}")
527 except ValueError as e:
528 _LOGGER.warning("SUPERNOTIFY Invalid media path for URL %s: %s", relative_path, e)
529 return None
531 async def share_path(self, artefact_path: Path) -> str | None:
532 """Return the HA media-share path for a local file (e.g. '/media/images/raw/foo.jpg').
534 Used by mobile push: the companion app resolves share-rooted paths against the HA base URL.
535 Returns None when media_path is unconfigured or path is outside the media tree.
536 """
537 if self.media_path is None or self.media_url_prefix is None:
538 return None
539 try:
540 if artefact_path.is_absolute():
541 relative = artefact_path.relative_to(await self.media_path.absolute())
542 else:
543 relative = artefact_path
544 return str(Path(self.media_url_prefix) / relative)
545 except ValueError as e:
546 _LOGGER.debug("SUPERNOTIFY Failure creating shared media path for %s:%s", artefact_path, e)
547 return None
549 async def size(self) -> int:
550 path: Path | None = self.media_path
551 if path and await path.exists():
552 return sum(1 for p in await aiofiles.os.listdir(path))
553 return 0
555 async def cleanup(self, days: int | None = None, force: bool = False) -> int:
556 if (
557 not force
558 and self.last_purge is not None
559 and self.last_purge > dt.datetime.now(dt.UTC) - dt.timedelta(minutes=self.purge_minute_interval)
560 ):
561 _LOGGER.debug(
562 "SUPERNOTIFY Media storage cleanup skipped, force: %s, last purge: %s, purge interval: %s",
563 force,
564 self.last_purge,
565 self.purge_minute_interval,
566 )
567 return 0
568 days = days or self.days
569 if days == 0 or self.media_path is None:
570 _LOGGER.debug("SUPERNOTIFY Media storage cleanup skipped, %s days, %s", days, self.media_path)
571 return 0
573 cutoff = dt.datetime.now(dt.UTC) - dt.timedelta(days=days)
574 cutoff = cutoff.astimezone(dt.UTC)
575 purged: int = 0
576 skipped: int = 0
577 if self.media_path and await self.media_path.exists():
578 try:
579 queue: list[Path] = [self.media_path]
580 while queue:
581 current = queue.pop()
582 for entry in await aiofiles.os.scandir(current):
583 if entry.is_dir():
584 queue.append(Path(entry.path))
585 elif entry.is_file() and dt_util.utc_from_timestamp(entry.stat().st_mtime) <= cutoff:
586 _LOGGER.debug("SUPERNOTIFY Purging %s", entry.path)
587 await aiofiles.os.unlink(Path(entry.path))
588 purged += 1
589 else:
590 skipped += 1
591 except Exception as e:
592 _LOGGER.warning("SUPERNOTIFY Unable to clean up media storage at %s: %s", self.media_path, e, exc_info=True)
593 _LOGGER.info("SUPERNOTIFY Purged %s media storage for cutoff %s, skipped %s", purged, cutoff, skipped)
594 self.last_purge = dt.datetime.now(dt.UTC)
595 else:
596 _LOGGER.warning("SUPERNOTIFY Skipping media storage cleanup for unknown path %s", self.media_path)
597 return purged