feat: add thumbnail_name field to attachment storage and refactor link/delete to batch SQL queries

devplacepy/attachments.py

@@ -168,6 +168,7 @@ def store_attachment(file_bytes, original_filename, user_uid):
         "image_width": image_width,
         "image_height": image_height,
         "has_thumbnail": 1 if thumbnail else 0,
+        "thumbnail_name": thumbnail,
         "created_at": datetime.now(timezone.utc).isoformat(),
     })
     return {
@@ -183,43 +184,66 @@ def store_attachment(file_bytes, original_filename, user_uid):
 
 
 def link_attachments(uids, target_type, target_uid):
-    if not uids:
+    flat = [uid.strip() for raw in uids or [] for uid in str(raw).split(",") if uid.strip()]
+    if not flat:
         return
-    attachments = get_table("attachments")
-    for raw in uids:
-        for uid in str(raw).split(","):
-            uid = uid.strip()
-            if not uid:
-                continue
-            existing = attachments.find_one(uid=uid)
-            if existing:
-                attachments.update({"id": existing["id"], "target_type": target_type, "target_uid": target_uid}, ["id"])
+    placeholders = ",".join(f":p{i}" for i in range(len(flat)))
+    params = {f"p{i}": uid for i, uid in enumerate(flat)}
+    db.query(
+        f"UPDATE attachments SET target_type=:tt, target_uid=:tu WHERE uid IN ({placeholders})",
+        tt=target_type, tu=target_uid, **params,
+    )
+
+
+def _unlink_attachment_files(row):
+    stored_name = row.get("stored_name", "")
+    directory = row.get("directory", "")
+    if not (stored_name and directory):
+        return
+    file_path = ATTACHMENTS_DIR / directory / stored_name
+    try:
+        file_path.unlink(missing_ok=True)
+    except Exception as e:
+        logger.warning(f"Failed to delete attachment file {file_path}: {e}")
+    for thumb_path in (ATTACHMENTS_DIR / directory).glob(f"{Path(stored_name).stem}_thumb.*"):
+        try:
+            thumb_path.unlink(missing_ok=True)
+        except Exception as e:
+            logger.warning(f"Failed to delete thumbnail {thumb_path}: {e}")
+
+
+def _delete_attachment_row(row):
+    _unlink_attachment_files(row)
+    get_table("attachments").delete(id=row["id"])
 
 
 def delete_attachment(uid):
-    attachments = get_table("attachments")
-    attachment = attachments.find_one(uid=uid)
-    if not attachment:
-        return
-    stored_name = attachment.get("stored_name", "")
-    directory = attachment.get("directory", "")
-    if stored_name and directory:
-        file_path = ATTACHMENTS_DIR / directory / stored_name
-        try:
-            file_path.unlink(missing_ok=True)
-        except Exception as e:
-            logger.warning(f"Failed to delete attachment file {file_path}: {e}")
-        for thumb_path in (ATTACHMENTS_DIR / directory).glob(f"{Path(stored_name).stem}_thumb.*"):
-            try:
-                thumb_path.unlink(missing_ok=True)
-            except Exception as e:
-                logger.warning(f"Failed to delete thumbnail {thumb_path}: {e}")
-    attachments.delete(id=attachment["id"])
+    row = get_table("attachments").find_one(uid=uid)
+    if row:
+        _delete_attachment_row(row)
 
 
 def delete_target_attachments(target_type, target_uid):
-    for attachment in get_table("attachments").find(target_type=target_type, target_uid=target_uid):
-        delete_attachment(attachment["uid"])
+    for row in get_table("attachments").find(target_type=target_type, target_uid=target_uid):
+        _delete_attachment_row(row)
+
+
+def delete_attachments_for(target_type, target_uids):
+    uids = [uid for uid in target_uids if uid]
+    if not uids or "attachments" not in db.tables:
+        return
+    placeholders = ",".join(f":p{i}" for i in range(len(uids)))
+    params = {f"p{i}": uid for i, uid in enumerate(uids)}
+    rows = list(db.query(
+        f"SELECT * FROM attachments WHERE target_type=:tt AND target_uid IN ({placeholders})",
+        tt=target_type, **params,
+    ))
+    if not rows:
+        return
+    for row in rows:
+        _unlink_attachment_files(row)
+    ids = ",".join(str(row["id"]) for row in rows)
+    db.query(f"DELETE FROM attachments WHERE id IN ({ids})")
 
 
 def get_attachments(target_type, target_uid):
@@ -252,9 +276,11 @@ def _row_to_attachment(row):
     directory = row.get("directory", "")
     thumb_name = None
     if row.get("has_thumbnail"):
-        stem = Path(stored_name).stem
-        matches = sorted((ATTACHMENTS_DIR / directory).glob(f"{stem}_thumb.*"))
-        thumb_name = matches[0].name if matches else f"{stem}_thumb.jpg"
+        thumb_name = row.get("thumbnail_name")
+        if not thumb_name:
+            stem = Path(stored_name).stem
+            png = f"{stem}_thumb.png"
+            thumb_name = png if (ATTACHMENTS_DIR / directory / png).exists() else f"{stem}_thumb.jpg"
     return {
         "uid": row["uid"],
         "original_filename": row.get("original_filename", ""),