"""Вложения проектных карточек. По решению: файлы хранятся в MinIO (креды в env), в БД — метаданные и ключ объекта. Тип определяется автоматически по MIME и расширению. """ from __future__ import annotations import json import time from ..db import store KIND_BY_EXT = { "image": {"png", "jpg", "jpeg", "gif", "webp", "svg", "bmp", "avif", "heic"}, "video": {"mp4", "mov", "avi", "mkv", "webm", "m4v"}, "audio": {"mp3", "wav", "ogg", "m4a", "flac", "aac"}, "archive": {"zip", "rar", "7z", "tar", "gz", "bz2"}, "document": {"pdf", "doc", "docx", "xls", "xlsx", "csv", "txt", "md", "rtf", "ppt", "pptx", "odt", "ods"}, } KIND_LABELS = { "image": "Изображение", "video": "Видео", "audio": "Аудио", "archive": "Архив", "document": "Документ", "other": "Файл", } def detect(name: str, mime: str = "") -> dict: ext = (name.split(".")[-1] if "." in name else "").lower() kind = "other" if mime.startswith("image/"): kind = "image" elif mime.startswith("video/"): kind = "video" elif mime.startswith("audio/"): kind = "audio" else: for k, exts in KIND_BY_EXT.items(): if ext in exts: kind = k break return {"kind": kind, "label": KIND_LABELS[kind]} def _files_of(card_id: str) -> list[dict]: from .projects import patch_card row = store.query_one("SELECT files FROM projects WHERE id = ?", [card_id]) if not row: raise KeyError(card_id) return json.loads(row["files"] or "[]") def add_file(card_id: str, name: str, data: bytes, mime: str = "") -> dict: """Сохраняет тело в MinIO и метаданные в карточку.""" from . import object_store from .projects import patch_card files = _files_of(card_id) info = detect(name, mime) object_key = object_store.put(card_id, name, data, mime) entry = { "id": store.uid("pf_"), "name": name, "size": len(data), "kind": info["kind"], "label": info["label"], "objectKey": object_key, } files.append(entry) patch_card(card_id, {"files": files}) return entry def get_file_entry(card_id: str, file_id: str) -> tuple[dict, list[dict]]: files = _files_of(card_id) entry = next((f for f in files if f["id"] == file_id), None) if not entry: raise KeyError(file_id) return entry, files def remove_file(card_id: str, file_id: str) -> None: from . import object_store from .projects import patch_card files = _files_of(card_id) entry = next((f for f in files if f["id"] == file_id), None) if entry and entry.get("objectKey"): object_store.remove(entry["objectKey"]) patch_card(card_id, {"files": [f for f in files if f["id"] != file_id]}) def object_storage_available() -> bool: from . import object_store return object_store.configured()