Coverage for src/lexigram/admin/media/library.py: 90%
81 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""MediaLibrary — metadata management for the admin media library."""
3from __future__ import annotations
5from datetime import UTC, datetime
6from typing import Any
7import uuid
9from lexigram.admin.media.models import MediaItem, MediaPage
12class MediaLibrary:
13 """Manages the metadata layer for the admin media library.
15 Args:
16 storage: Optional storage service instance (used for
17 :meth:`delete_storage`).
18 default_folder: Default folder prefix for new uploads.
19 """
21 def __init__(
22 self,
23 storage: Any = None,
24 default_folder: str = "uploads",
25 ) -> None:
26 self._storage = storage
27 self._default_folder = default_folder
28 self._items: dict[str, MediaItem] = {}
30 # ------------------------------------------------------------------
31 # Register / update
32 # ------------------------------------------------------------------
34 async def register(
35 self,
36 path: str,
37 *,
38 filename: str = "",
39 mime_type: str = "",
40 size_bytes: int = 0,
41 uploaded_by: str = "system",
42 alt_text: str = "",
43 caption: str = "",
44 tags: list[str] | None = None,
45 width: int = 0,
46 height: int = 0,
47 item_id: str | None = None,
48 metadata: dict[str, Any] | None = None,
49 ) -> MediaItem:
50 """Register a newly uploaded file in the library.
52 Args:
53 path: Storage path.
54 filename: Human-readable filename.
55 mime_type: MIME type.
56 size_bytes: File size in bytes.
57 uploaded_by: Uploader user ID.
58 alt_text: Accessible alt text.
59 caption: Optional caption.
60 tags: Categorisation tags.
61 width: Image width (pixels).
62 height: Image height (pixels).
63 item_id: Explicit ID (auto-generated if omitted).
64 metadata: Extra metadata.
66 Returns:
67 The registered :class:`MediaItem`.
68 """
69 folder = "/".join(path.split("/")[:-1]) if "/" in path else self._default_folder
70 if not filename:
71 filename = path.rsplit("/", maxsplit=1)[-1]
73 item = MediaItem(
74 item_id=item_id or str(uuid.uuid4())[:12],
75 path=path,
76 filename=filename,
77 mime_type=mime_type,
78 size_bytes=size_bytes,
79 folder=folder,
80 uploaded_by=uploaded_by,
81 alt_text=alt_text,
82 caption=caption,
83 tags=tags or [],
84 width=width,
85 height=height,
86 metadata=metadata or {},
87 )
88 self._items[item.item_id] = item
89 return item
91 def update(
92 self,
93 item_id: str,
94 *,
95 alt_text: str | None = None,
96 caption: str | None = None,
97 tags: list[str] | None = None,
98 ) -> MediaItem | None:
99 """Update the metadata of a registered item.
101 Args:
102 item_id: Library item ID.
103 alt_text: New alt text (``None`` = no change).
104 caption: New caption.
105 tags: New tags list.
107 Returns:
108 Updated :class:`MediaItem` or ``None`` if not found.
109 """
110 item = self._items.get(item_id)
111 if item is None:
112 return None
113 if alt_text is not None:
114 item.alt_text = alt_text
115 if caption is not None:
116 item.caption = caption
117 if tags is not None:
118 item.tags = list(tags)
119 return item
121 # ------------------------------------------------------------------
122 # Browse
123 # ------------------------------------------------------------------
125 def get(self, item_id: str) -> MediaItem | None:
126 """Return a single item by ID.
128 Args:
129 item_id: Library item ID.
130 """
131 return self._items.get(item_id)
133 def list_folder(
134 self,
135 folder: str = "",
136 *,
137 page: int = 1,
138 page_size: int = 24,
139 include_deleted: bool = False,
140 ) -> MediaPage:
141 """Return a paginated list of items in *folder*.
143 Args:
144 folder: Folder path prefix (empty = all items).
145 page: 1-based page number.
146 page_size: Items per page.
147 include_deleted: When ``True``, soft-deleted items are included.
149 Returns:
150 :class:`MediaPage` with items for the requested page.
151 """
152 all_items = [
153 item
154 for item in self._items.values()
155 if (not folder or item.folder.startswith(folder))
156 and (include_deleted or not item.is_deleted)
157 ]
158 all_items.sort(key=lambda i: i.uploaded_at, reverse=True)
159 total = len(all_items)
160 start = (page - 1) * page_size
161 end = start + page_size
162 return MediaPage(
163 items=all_items[start:end],
164 page=page,
165 page_size=page_size,
166 total=total,
167 )
169 def search(
170 self,
171 query: str,
172 *,
173 mime_prefix: str = "",
174 tags: list[str] | None = None,
175 include_deleted: bool = False,
176 ) -> list[MediaItem]:
177 """Search for items by filename, alt text, caption, or tags.
179 Args:
180 query: Search string (case-insensitive).
181 mime_prefix: Filter by MIME type prefix (e.g. ``"image/"``).
182 tags: Filter by tags (item must have all listed tags).
183 include_deleted: Include soft-deleted items.
185 Returns:
186 Matching items sorted by relevance (exact filename match first).
187 """
188 q = query.lower()
189 results = []
190 for item in self._items.values():
191 if not include_deleted and item.is_deleted:
192 continue
193 if mime_prefix and not item.mime_type.startswith(mime_prefix):
194 continue
195 if tags:
196 if not all(t in item.tags for t in tags):
197 continue
198 if (
199 q in item.filename.lower()
200 or q in item.alt_text.lower()
201 or q in item.caption.lower()
202 or any(q in tag.lower() for tag in item.tags)
203 ):
204 results.append(item)
206 results.sort(
207 key=lambda i: (
208 0 if i.filename.lower() == q else 1,
209 -i.uploaded_at.timestamp(),
210 )
211 )
212 return results
214 def list_folders(self) -> list[str]:
215 """Return all unique folder paths (sorted, excluding deleted items)."""
216 return sorted(
217 {item.folder for item in self._items.values() if not item.is_deleted}
218 )
220 # ------------------------------------------------------------------
221 # Soft delete / restore
222 # ------------------------------------------------------------------
224 def soft_delete(self, item_id: str) -> MediaItem | None:
225 """Soft-delete an item (moves to trash without destroying storage).
227 Args:
228 item_id: Library item ID.
230 Returns:
231 Updated :class:`MediaItem` or ``None`` if not found.
232 """
233 item = self._items.get(item_id)
234 if item is None:
235 return None
236 item.is_deleted = True
237 item.deleted_at = datetime.now(UTC)
238 return item
240 def restore(self, item_id: str) -> MediaItem | None:
241 """Restore a soft-deleted item.
243 Args:
244 item_id: Library item ID.
246 Returns:
247 Updated :class:`MediaItem` or ``None`` if not found.
248 """
249 item = self._items.get(item_id)
250 if item is None:
251 return None
252 item.is_deleted = False
253 item.deleted_at = None
254 return item
256 async def purge(self, item_id: str) -> bool:
257 """Permanently remove an item from the library (and optionally storage).
259 Args:
260 item_id: Library item ID.
262 Returns:
263 ``True`` if removed, ``False`` if not found.
264 """
265 item = self._items.pop(item_id, None)
266 if item is None:
267 return False
268 if self._storage:
269 try:
270 await self._storage.delete(item.path)
271 except Exception: # noqa: BLE001, S110
272 pass
273 return True
275 def trash(self) -> list[MediaItem]:
276 """Return all soft-deleted items."""
277 return [i for i in self._items.values() if i.is_deleted]
280__all__ = [
281 "MediaLibrary",
282]