# retoor <retoor@molodetz.nl>

import asyncio
import ipaddress
import logging
import os
import socket
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse
from PIL import Image
from io import BytesIO
import httpx
from devplacepy.net_guard import BlockedAddressError, guarded_async_client
from devplacepy.database import get_table, db, get_setting
from devplacepy.config import UPLOADS_DIR, ATTACHMENTS_DIR
from devplacepy.utils import generate_uid

logger = logging.getLogger(__name__)

REMOTE_FETCH_TIMEOUT = 20.0
REMOTE_FETCH_USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
)

THUMBNAIL_SIZE = (200, 200)
THUMBNAIL_QUALITY = 80
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".tiff"}
THUMBNAIL_EXTENSIONS = IMAGE_EXTENSIONS - {".gif"}
POST_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp"}

ALLOWED_UPLOAD_TYPES = {
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".png": "image/png",
    ".gif": "image/gif",
    ".webp": "image/webp",
    ".bmp": "image/bmp",
    ".tiff": "image/tiff",
    ".pdf": "application/pdf",
    ".zip": "application/zip",
    ".mp4": "video/mp4",
    ".webm": "video/webm",
    ".ogv": "video/ogg",
    ".mov": "video/quicktime",
    ".m4v": "video/x-m4v",
    ".mp3": "audio/mpeg",
    ".txt": "text/plain",
    ".py": "text/x-python",
    ".js": "text/javascript",
    ".css": "text/css",
    ".md": "text/markdown",
    ".wav": "audio/wav",
    ".flac": "audio/flac",
    ".ogg": "audio/ogg",
    ".aac": "audio/aac",
    ".wma": "audio/x-ms-wma",
    ".m4a": "audio/mp4",
    ".avi": "video/x-msvideo",
    ".mkv": "video/x-matroska",
    ".flv": "video/x-flv",
    ".wmv": "video/x-ms-wmv",
    ".3gp": "video/3gpp",
    ".csv": "text/csv",
    ".doc": "application/msword",
    ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    ".xls": "application/vnd.ms-excel",
    ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    ".ppt": "application/vnd.ms-powerpoint",
    ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
    ".odt": "application/vnd.oasis.opendocument.text",
    ".rtf": "application/rtf",
    ".json": "application/json",
    ".xml": "application/xml",
    ".yaml": "text/yaml",
    ".yml": "text/yaml",
    ".toml": "text/x-toml",
    ".sh": "text/x-sh",
    ".bat": "text/x-bat",
    ".ts": "text/typescript",
    ".java": "text/x-java",
    ".cpp": "text/x-c++",
    ".c": "text/x-c",
    ".h": "text/x-c-header",
    ".rb": "text/x-ruby",
    ".go": "text/x-go",
    ".rs": "text/x-rust",
    ".sql": "text/x-sql",
    ".php": "text/x-php",
    ".swift": "text/x-swift",
    ".kt": "text/x-kotlin",
    ".cfg": "text/x-config",
    ".ini": "text/x-config",
    ".log": "text/plain",
    ".tar": "application/x-tar",
    ".gz": "application/gzip",
    ".rar": "application/vnd.rar",
    ".7z": "application/x-7z-compressed",
}

MIME_TO_EXT = {
    "image/jpeg": ".jpg",
    "image/png": ".png",
    "image/gif": ".gif",
    "image/webp": ".webp",
    "image/bmp": ".bmp",
    "image/tiff": ".tiff",
    "application/pdf": ".pdf",
    "application/zip": ".zip",
    "video/mp4": ".mp4",
    "video/webm": ".webm",
    "video/ogg": ".ogv",
    "video/quicktime": ".mov",
    "video/x-m4v": ".m4v",
    "audio/mpeg": ".mp3",
    "text/plain": ".txt",
    "text/markdown": ".md",
    "audio/wav": ".wav",
    "audio/flac": ".flac",
    "audio/ogg": ".ogg",
    "audio/aac": ".aac",
    "audio/x-ms-wma": ".wma",
    "audio/mp4": ".m4a",
    "video/x-msvideo": ".avi",
    "video/x-matroska": ".mkv",
    "video/x-flv": ".flv",
    "video/x-ms-wmv": ".wmv",
    "video/3gpp": ".3gp",
    "text/csv": ".csv",
    "application/msword": ".doc",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
    "application/vnd.ms-excel": ".xls",
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
    "application/vnd.ms-powerpoint": ".ppt",
    "application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx",
    "application/vnd.oasis.opendocument.text": ".odt",
    "application/rtf": ".rtf",
    "application/json": ".json",
    "application/xml": ".xml",
    "text/yaml": ".yaml",
    "text/x-toml": ".toml",
    "text/x-sh": ".sh",
    "text/x-bat": ".bat",
    "text/typescript": ".ts",
    "text/x-java": ".java",
    "text/x-c++": ".cpp",
    "text/x-c": ".c",
    "text/x-c-header": ".h",
    "text/x-ruby": ".rb",
    "text/x-go": ".go",
    "text/x-rust": ".rs",
    "text/x-sql": ".sql",
    "text/x-php": ".php",
    "text/x-swift": ".swift",
    "text/x-kotlin": ".kt",
    "text/x-config": ".cfg",
    "application/x-tar": ".tar",
    "application/gzip": ".gz",
    "application/vnd.rar": ".rar",
    "application/x-7z-compressed": ".7z",
}

FILE_ICONS = {
    ".pdf": "\U0001f4c4",
    ".zip": "\U0001f4e6",
    ".gz": "\U0001f4e6",
    ".tar": "\U0001f4e6",
    ".rar": "\U0001f4e6",
    ".7z": "\U0001f4e6",
    ".mp4": "\U0001f3ac",
    ".webm": "\U0001f3ac",
    ".ogv": "\U0001f3ac",
    ".mov": "\U0001f3ac",
    ".m4v": "\U0001f3ac",
    ".mp3": "\U0001f3b5",
    ".py": "\U0001f4bb",
    ".js": "\U0001f4bb",
    ".ts": "\U0001f4bb",
    ".html": "\U0001f4bb",
    ".css": "\U0001f4bb",
    ".json": "\U0001f4bb",
    ".md": "\U0001f4bb",
    ".csv": "\U0001f4ca",
    ".xls": "\U0001f4ca",
    ".xlsx": "\U0001f4ca",
    ".doc": "\U0001f4dd",
    ".docx": "\U0001f4dd",
    ".txt": "\U0001f4c4",
    ".exe": "\u2699",
    ".bin": "\u2699",
}
DEFAULT_FILE_ICON = "\U0001f4ce"


def _get_max_upload_bytes():
    return int(get_setting("max_upload_size_mb", "10")) * 1024 * 1024


WILDCARD_TOKENS = {"*", ".*", "*.*"}


def allowed_extensions():
    raw = get_setting("allowed_file_types", "").strip()
    if not raw:
        return set(ALLOWED_UPLOAD_TYPES)
    tokens = {part.strip().lower() for part in raw.split(",") if part.strip()}
    if tokens & WILDCARD_TOKENS:
        return set(ALLOWED_UPLOAD_TYPES)
    return {token if token.startswith(".") else f".{token}" for token in tokens}


def is_extension_allowed(ext):
    return ext in allowed_extensions()


def _directory_for(uid):
    tail = uid.replace("-", "")
    return f"{tail[-2:]}/{tail[-4:-2]}"


def _detect_mime(file_bytes, original_filename):
    ext = Path(original_filename).suffix.lower()
    return ALLOWED_UPLOAD_TYPES.get(ext, "application/octet-stream")


def _image_dimensions(file_bytes):
    try:
        img = Image.open(BytesIO(file_bytes))
        return img.width, img.height
    except Exception as e:
        logger.warning(f"Could not read image dimensions: {e}")
        return None, None


def _generate_thumbnail(file_bytes, thumb_path):
    try:
        img = Image.open(BytesIO(file_bytes))
        img.thumbnail(THUMBNAIL_SIZE, Image.LANCZOS)
        if img.mode in ("RGBA", "P"):
            img = img.convert("RGBA")
            thumb_path = thumb_path.with_suffix(".png")
            img.save(str(thumb_path), "PNG")
            return thumb_path.name
        img = img.convert("RGB")
        img.save(str(thumb_path), "JPEG", quality=THUMBNAIL_QUALITY)
        return thumb_path.name
    except Exception as e:
        logger.warning(f"Thumbnail generation failed: {e}")
        return None


def save_inline_image(file_bytes, original_filename):
    if len(file_bytes) > _get_max_upload_bytes():
        logger.warning(f"Inline image too large: {original_filename}")
        return None
    ext = Path(original_filename).suffix.lower()
    if ext not in POST_IMAGE_EXTENSIONS:
        logger.warning(f"Unsupported inline image type: {ext}")
        return None
    UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
    filename = f"{generate_uid()}{ext}"
    (UPLOADS_DIR / filename).write_bytes(file_bytes)
    logger.info(f"Inline image saved: {filename}")
    return filename


def delete_inline_image(filename):
    if not filename:
        return
    try:
        (UPLOADS_DIR / filename).unlink(missing_ok=True)
    except Exception as e:
        logger.warning(f"Failed to delete inline image {filename}: {e}")


def store_attachment(file_bytes, original_filename, user_uid):
    if len(file_bytes) > _get_max_upload_bytes():
        return None
    ext = Path(original_filename).suffix.lower()
    if not is_extension_allowed(ext):
        return None

    uid = generate_uid()
    stored_name = f"{uid}{ext}"
    directory = _directory_for(uid)
    file_dir = ATTACHMENTS_DIR / directory
    file_dir.mkdir(parents=True, exist_ok=True)
    (file_dir / stored_name).write_bytes(file_bytes)

    mime = _detect_mime(file_bytes, original_filename)
    is_image = mime.startswith("image/")
    image_width, image_height = None, None
    thumbnail = None
    if is_image:
        image_width, image_height = _image_dimensions(file_bytes)
        if ext not in (".gif",):
            thumbnail = _generate_thumbnail(file_bytes, file_dir / f"{uid}_thumb.jpg")

    is_audio = mime.startswith("audio/")

    get_table("attachments").insert(
        {
            "uid": uid,
            "target_type": "",
            "target_uid": "",
            "user_uid": user_uid,
            "original_filename": original_filename,
            "stored_name": stored_name,
            "directory": directory,
            "file_size": len(file_bytes),
            "mime_type": mime,
            "image_width": image_width,
            "image_height": image_height,
            "has_thumbnail": 1 if thumbnail else 0,
            "thumbnail_name": thumbnail,
            "gitea_asset_id": None,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "deleted_at": None,
            "deleted_by": None,
        }
    )
    return {
        "uid": uid,
        "original_filename": original_filename,
        "file_size": len(file_bytes),
        "mime_type": mime,
        "url": f"/static/uploads/attachments/{directory}/{stored_name}",
        "thumbnail_url": f"/static/uploads/attachments/{directory}/{thumbnail}"
        if thumbnail
        else None,
        "has_thumbnail": thumbnail is not None,
        "is_image": is_image,
        "is_video": mime.startswith("video/"),
        "is_audio": is_audio,
    }


class RemoteFetchError(Exception):
    def __init__(self, message, status=400):
        super().__init__(message)
        self.message = message
        self.status = status


async def _guard_public_url(url):
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        raise RemoteFetchError("Only http and https URLs can be attached.", 400)
    host = parsed.hostname
    if not host:
        raise RemoteFetchError("The URL has no host.", 400)
    try:
        infos = await asyncio.to_thread(socket.getaddrinfo, host, None)
    except socket.gaierror as exc:
        raise RemoteFetchError(f"Could not resolve host: {host}", 400) from exc
    for info in infos:
        address = ipaddress.ip_address(info[4][0])
        if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
            address = address.ipv4_mapped
        if (
            address.is_private
            or address.is_loopback
            or address.is_link_local
            or address.is_reserved
            or address.is_multicast
            or address.is_unspecified
        ):
            raise RemoteFetchError(
                f"Refusing to attach a private or local address ({address}).", 400
            )


def _resolve_remote_filename(final_url, content_type, override):
    name = (override or "").strip() or Path(urlparse(final_url).path).name
    ext = Path(name).suffix.lower()
    if name and ext and is_extension_allowed(ext):
        return name
    base_mime = (content_type or "").split(";")[0].strip().lower()
    mapped = MIME_TO_EXT.get(base_mime)
    if mapped is None or not is_extension_allowed(mapped):
        return None
    stem = Path(name).stem or "download"
    return f"{stem}{mapped}"


async def fetch_remote_file(url, filename=None):
    if "://" not in url:
        url = "https://" + url
    await _guard_public_url(url)
    max_bytes = _get_max_upload_bytes()
    try:
        async with guarded_async_client(
            follow_redirects=True,
            timeout=REMOTE_FETCH_TIMEOUT,
            headers={"User-Agent": REMOTE_FETCH_USER_AGENT},
        ) as client:
            async with client.stream("GET", url) as response:
                if response.status_code >= 400:
                    raise RemoteFetchError(
                        f"The remote server returned {response.status_code}.", 400
                    )
                final_url = str(response.url)
                content_type = response.headers.get("content-type", "")
                chunks = []
                total = 0
                async for chunk in response.aiter_bytes():
                    chunks.append(chunk)
                    total += len(chunk)
                    if total > max_bytes:
                        raise RemoteFetchError(
                            f"The file exceeds the {max_bytes // (1024 * 1024)}MB limit.",
                            413,
                        )
                data = b"".join(chunks)
    except BlockedAddressError as exc:
        raise RemoteFetchError(str(exc), 400) from exc
    except httpx.HTTPError as exc:
        raise RemoteFetchError(f"Could not fetch {url}: {exc}", 400) from exc

    name = _resolve_remote_filename(final_url, content_type, filename)
    if name is None:
        raise RemoteFetchError(
            "Could not determine an allowed file type for the URL. Pass a filename "
            "with an allowed extension.",
            415,
        )
    return name, data


async def store_attachment_from_url(url, user_uid, filename=None):
    name, data = await fetch_remote_file(url, filename)
    result = store_attachment(data, name, user_uid)
    if result is None:
        raise RemoteFetchError(
            "The downloaded file is not an allowed type or exceeds the size limit.",
            413,
        )
    return result


def link_attachments(uids, target_type, target_uid):
    flat = [
        uid.strip() for raw in uids or [] for uid in str(raw).split(",") if uid.strip()
    ]
    if not flat:
        return
    placeholders = ",".join(f":p{i}" for i in range(len(flat)))
    params = {f"p{i}": uid for i, uid in enumerate(flat)}
    with db:
        db.query(
            f"UPDATE attachments SET target_type=:tt, target_uid=:tu WHERE uid IN ({placeholders})",
            tt=target_type,
            tu=target_uid,
            **params,
        )


def split_attachment_uids(raw):
    return [
        uid.strip() for item in raw or [] for uid in str(item).split(",") if uid.strip()
    ]


def get_orphan_attachments_batch(uids, user, admin=False):
    if not uids:
        return []
    placeholders = ",".join(f":p{i}" for i in range(len(uids)))
    params = {f"p{i}": uid for i, uid in enumerate(uids)}
    rows = db.query(
        f"SELECT * FROM attachments WHERE uid IN ({placeholders}) AND deleted_at IS NULL",
        **params,
    )
    by_uid = {row["uid"]: row for row in rows}
    owned = []
    for uid in uids:
        row = by_uid.get(uid)
        if not row or row.get("target_uid"):
            continue
        if row.get("user_uid") and row["user_uid"] != user["uid"] and not admin:
            continue
        owned.append(uid)
    return owned


def set_gitea_asset_id(uid, asset_id):
    get_table("attachments").update(
        {"uid": uid, "gitea_asset_id": int(asset_id)}, ["uid"]
    )


async def mirror_attachment_to_gitea(uid):
    from devplacepy.services.gitea import runtime
    from devplacepy.services.gitea.client import GiteaError

    row = get_table("attachments").find_one(uid=uid, deleted_at=None)
    if not row:
        return None
    target_type = row.get("target_type", "")
    target_uid = row.get("target_uid", "")
    if target_type not in ("issue", "issue_comment") or not target_uid:
        return None
    path = ATTACHMENTS_DIR / row.get("directory", "") / row.get("stored_name", "")
    try:
        data = await asyncio.to_thread(path.read_bytes)
    except OSError as exc:
        logger.warning("Cannot read attachment %s for Gitea mirror: %s", uid, exc)
        return None
    filename = row.get("original_filename") or row.get("stored_name") or "file"
    mime = row.get("mime_type") or "application/octet-stream"
    client = runtime.get_client()
    try:
        if target_type == "issue":
            asset = await client.create_issue_asset(
                int(target_uid), filename, data, mime
            )
        else:
            asset = await client.create_comment_asset(
                int(target_uid), filename, data, mime
            )
    except (GiteaError, ValueError) as exc:
        logger.warning("Gitea asset mirror failed for %s: %s", uid, exc)
        return None
    asset_id = int(asset.get("id", 0) or 0)
    if asset_id:
        set_gitea_asset_id(uid, asset_id)
    return asset_id


async def remove_gitea_asset(row):
    from devplacepy.services.gitea import runtime
    from devplacepy.services.gitea.client import GiteaError

    asset_id = int(row.get("gitea_asset_id") or 0)
    target_type = row.get("target_type", "")
    target_uid = row.get("target_uid", "")
    if not asset_id or not target_uid:
        return
    client = runtime.get_client()
    try:
        if target_type == "issue":
            await client.delete_issue_asset(int(target_uid), asset_id)
        elif target_type == "issue_comment":
            await client.delete_comment_asset(int(target_uid), asset_id)
    except (GiteaError, ValueError) as exc:
        logger.warning("Gitea asset delete failed for %s: %s", row.get("uid"), exc)


_pending_gitea_tasks: set[asyncio.Task] = set()


def _fire_and_forget(coro) -> None:
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        coro.close()
        return
    task = loop.create_task(coro)
    _pending_gitea_tasks.add(task)
    task.add_done_callback(_pending_gitea_tasks.discard)


def schedule_gitea_mirror(uid: str) -> None:
    _fire_and_forget(mirror_attachment_to_gitea(uid))


def schedule_gitea_removal(row: dict) -> None:
    _fire_and_forget(remove_gitea_asset(row))


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):
    row = get_table("attachments").find_one(uid=uid)
    if row:
        _delete_attachment_row(row)


def rename_attachment(uid, filename):
    row = get_table("attachments").find_one(uid=uid, deleted_at=None)
    if not row:
        return None
    ext = Path(row.get("stored_name", "")).suffix.lower()
    stem = Path(str(filename)).name.strip()
    if ext:
        stem = Path(stem).stem
    if not stem:
        return None
    clean = f"{stem}{ext}"
    get_table("attachments").update({"uid": uid, "original_filename": clean}, ["uid"])
    return clean


def soft_delete_attachment(uid, deleted_by="system"):
    row = get_table("attachments").find_one(uid=uid)
    if not row or row.get("deleted_at"):
        return None
    get_table("attachments").update(
        {
            "uid": uid,
            "deleted_at": datetime.now(timezone.utc).isoformat(),
            "deleted_by": deleted_by,
        },
        ["uid"],
    )
    return row


def restore_attachment(uid):
    row = get_table("attachments").find_one(uid=uid)
    if not row or not row.get("deleted_at"):
        return False
    get_table("attachments").update(
        {"uid": uid, "deleted_at": None, "deleted_by": None}, ["uid"]
    )
    return True


def purge_soft_deleted_attachments(*, dry_run: bool = False) -> tuple[int, int]:
    if "attachments" not in db.tables:
        return 0, 0
    table = get_table("attachments")
    rows = list(table.find(table.table.columns.deleted_at.isnot(None)))
    removed = 0
    freed = 0
    for row in rows:
        directory = row.get("directory")
        stored_name = row.get("stored_name")
        if directory and stored_name:
            path = ATTACHMENTS_DIR / directory / stored_name
            try:
                freed += path.stat().st_size
            except OSError:
                pass
            for thumb in (ATTACHMENTS_DIR / directory).glob(
                f"{Path(stored_name).stem}_thumb.*"
            ):
                try:
                    freed += thumb.stat().st_size
                except OSError:
                    pass
            if not dry_run:
                _unlink_attachment_files(row)
        if not dry_run:
            table.delete(id=row["id"])
        removed += 1
    return removed, freed


def sweep_orphan_attachment_blobs(*, dry_run: bool = False) -> tuple[int, int]:
    if not ATTACHMENTS_DIR.exists():
        return 0, 0
    referenced_stems = set()
    if "attachments" in db.tables:
        for row in get_table("attachments").find():
            directory = row.get("directory")
            stored_name = row.get("stored_name")
            if directory and stored_name:
                referenced_stems.add((directory, Path(stored_name).stem))
    removed = 0
    freed = 0
    base = str(ATTACHMENTS_DIR)
    for root, _dirs, files in os.walk(base):
        directory = os.path.relpath(root, base)
        for name in files:
            stem = Path(name).stem
            if stem.endswith("_thumb"):
                stem = stem[: -len("_thumb")]
            if (directory, stem) in referenced_stems:
                continue
            file_path = os.path.join(root, name)
            try:
                size = os.path.getsize(file_path)
            except OSError:
                continue
            if not dry_run:
                try:
                    os.unlink(file_path)
                except OSError:
                    continue
            removed += 1
            freed += size
    return removed, freed


def soft_delete_target_attachments(target_type, target_uid, deleted_by):
    stamp = datetime.now(timezone.utc).isoformat()
    for row in get_table("attachments").find(
        target_type=target_type, target_uid=target_uid, deleted_at=None
    ):
        get_table("attachments").update(
            {"uid": row["uid"], "deleted_at": stamp, "deleted_by": deleted_by}, ["uid"]
        )


def soft_delete_attachments_for(target_type, target_uids, deleted_by):
    from devplacepy.database import soft_delete_in

    soft_delete_in(
        "attachments",
        "target_uid",
        target_uids,
        deleted_by,
        target_type=target_type,
    )


def delete_target_attachments(target_type, target_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)
    with db:
        db.query(f"DELETE FROM attachments WHERE id IN ({ids})")


def get_attachments(target_type, target_uid):
    if "attachments" not in db.tables:
        return []
    rows = list(
        get_table("attachments").find(
            target_type=target_type,
            target_uid=target_uid,
            deleted_at=None,
            order_by=["created_at"],
        )
    )
    return [_row_to_attachment(r) for r in rows]


def get_attachments_batch(target_type, uids):
    if not uids:
        return {}
    if "attachments" not in db.tables:
        return {uid: [] for uid in uids}
    placeholders = ",".join(f":p{i}" for i in range(len(uids)))
    params = {f"p{i}": uid for i, uid in enumerate(uids)}
    rows = db.query(
        f"SELECT * FROM attachments WHERE target_type=:tt AND target_uid IN ({placeholders}) AND deleted_at IS NULL ORDER BY created_at",
        tt=target_type,
        **params,
    )
    result = {uid: [] for uid in uids}
    for row in rows:
        if row["target_uid"] in result:
            result[row["target_uid"]].append(_row_to_attachment(row))
    return result


def _row_to_attachment(row):
    stored_name = row.get("stored_name", "")
    directory = row.get("directory", "")
    thumb_name = None
    if row.get("has_thumbnail"):
        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", ""),
        "file_size": row.get("file_size", 0),
        "mime_type": row.get("mime_type", ""),
        "url": f"/static/uploads/attachments/{directory}/{stored_name}",
        "thumbnail_url": f"/static/uploads/attachments/{directory}/{thumb_name}"
        if thumb_name
        else None,
        "has_thumbnail": bool(row.get("has_thumbnail")),
        "is_image": row.get("mime_type", "").startswith("image/"),
        "is_video": row.get("mime_type", "").startswith("video/"),
        "is_audio": row.get("mime_type", "").startswith("audio/"),
        "target_type": row.get("target_type", ""),
        "target_uid": row.get("target_uid", ""),
        "user_uid": row.get("user_uid", ""),
        "gitea_asset_id": row.get("gitea_asset_id") or None,
        "created_at": row.get("created_at", ""),
    }


def format_file_size(size):
    if size < 1024:
        return f"{size} B"
    if size < 1048576:
        return f"{size / 1024:.1f} KB"
    return f"{size / 1048576:.1f} MB"


def file_icon_emoji(filename):
    ext = Path(filename).suffix.lower()
    return FILE_ICONS.get(ext, DEFAULT_FILE_ICON)
