devplacepy/devplacepy/attachments.py

import logging
from datetime import datetime, timezone
from pathlib import Path
from PIL import Image
from io import BytesIO
from devplacepy.database import get_table, db, get_setting
from devplacepy.config import STATIC_DIR
from devplacepy.utils import generate_uid

logger = logging.getLogger(__name__)

UPLOADS_DIR = STATIC_DIR / "uploads"
ATTACHMENTS_DIR = UPLOADS_DIR / "attachments"
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",
}

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


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


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


def _directory_for(uid):
    return f"{uid[:2]}/{uid[2:4]}"


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")

    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,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    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/"),
    }


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


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)
    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, 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}) 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/"),
    }


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)