209 lines
6.8 KiB
Python
209 lines
6.8 KiB
Python
"""Local image and video attachments for journal entries.
|
|
|
|
Files never leave the trusted zone. Images have EXIF/text chunks stripped.
|
|
Video containers are stored as-is; GPS in video metadata is not stripped yet.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from db import DATA_DIR, get_db, row_to_dict
|
|
from dialogue_store import StoreError
|
|
from journal_store import get_entry
|
|
|
|
IMAGE_TYPES = {
|
|
"image/jpeg": ".jpg",
|
|
"image/jpg": ".jpg",
|
|
"image/png": ".png",
|
|
"image/gif": ".gif",
|
|
"image/webp": ".webp",
|
|
}
|
|
VIDEO_TYPES = {
|
|
"video/mp4": ".mp4",
|
|
"video/webm": ".webm",
|
|
"video/quicktime": ".mov",
|
|
}
|
|
ALLOWED_TYPES = {**IMAGE_TYPES, **VIDEO_TYPES}
|
|
IMAGE_MAX_BYTES = 8 * 1024 * 1024
|
|
VIDEO_MAX_BYTES = 64 * 1024 * 1024
|
|
MEDIA_ROOT = DATA_DIR / "media"
|
|
MAX_BYTES = IMAGE_MAX_BYTES # tests and older imports
|
|
|
|
|
|
def _owned(conn, table: str, record_id: str, profile_id: str) -> dict | None:
|
|
return row_to_dict(
|
|
conn.execute(f"SELECT * FROM {table} WHERE id = ? AND profile_id = ?", (record_id, profile_id)).fetchone()
|
|
)
|
|
|
|
|
|
def media_kind(content_type: str) -> str:
|
|
kind = (content_type or "").lower()
|
|
if kind in VIDEO_TYPES or kind.startswith("video/"):
|
|
return "video"
|
|
return "image"
|
|
|
|
|
|
def public_media(row: dict | None) -> dict | None:
|
|
if not row:
|
|
return None
|
|
out = dict(row)
|
|
out["kind"] = media_kind(row.get("content_type") or "")
|
|
return out
|
|
|
|
|
|
def sniff_content_type(data: bytes, declared: str) -> str:
|
|
declared = (declared or "").lower().split(";")[0].strip()
|
|
if data.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
return "image/png"
|
|
if data[:3] == b"\xff\xd8\xff":
|
|
return "image/jpeg"
|
|
if data[:6] in {b"GIF87a", b"GIF89a"}:
|
|
return "image/gif"
|
|
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
|
return "image/webp"
|
|
if len(data) >= 12 and data[4:8] == b"ftyp":
|
|
brand = data[8:12]
|
|
if brand in {b"qt ", b"moov"}:
|
|
return "video/quicktime"
|
|
return "video/mp4"
|
|
if data[:4] == b"\x1a\x45\xdf\xa3":
|
|
return "video/webm"
|
|
if declared in ALLOWED_TYPES:
|
|
return declared
|
|
return declared
|
|
|
|
|
|
def strip_jpeg_exif(data: bytes) -> bytes:
|
|
if len(data) < 4 or data[0:2] != b"\xff\xd8":
|
|
return data
|
|
out = bytearray(b"\xff\xd8")
|
|
i = 2
|
|
while i + 1 < len(data):
|
|
if data[i] != 0xFF:
|
|
out.extend(data[i:])
|
|
break
|
|
marker = data[i + 1]
|
|
if marker == 0xDA:
|
|
out.extend(data[i:])
|
|
break
|
|
if marker == 0xD9:
|
|
out.extend(data[i : i + 2])
|
|
break
|
|
if marker in {0xD0, 0xD1, 0xD2, 0xD3, 0xD4, 0xD5, 0xD6, 0xD7, 0x01}:
|
|
out.extend(data[i : i + 2])
|
|
i += 2
|
|
continue
|
|
if i + 4 > len(data):
|
|
break
|
|
length = int.from_bytes(data[i + 2 : i + 4], "big")
|
|
block_end = i + 2 + length
|
|
if marker == 0xE1 and data[i + 4 : i + 10] == b"Exif\x00\x00":
|
|
i = block_end
|
|
continue
|
|
out.extend(data[i:block_end])
|
|
i = block_end
|
|
return bytes(out)
|
|
|
|
|
|
def strip_png_metadata(data: bytes) -> bytes:
|
|
signature = b"\x89PNG\r\n\x1a\n"
|
|
if not data.startswith(signature):
|
|
return data
|
|
out = bytearray(signature)
|
|
drop = {b"eXIf", b"eXif", b"tEXt", b"zTXt", b"iTXt", b"tIME"}
|
|
i = 8
|
|
while i + 12 <= len(data):
|
|
length = int.from_bytes(data[i : i + 4], "big")
|
|
chunk_type = data[i + 4 : i + 8]
|
|
end = i + 12 + length
|
|
if chunk_type not in drop:
|
|
out.extend(data[i:end])
|
|
if chunk_type == b"IEND":
|
|
break
|
|
i = end
|
|
return bytes(out)
|
|
|
|
|
|
def prepare_image(data: bytes, content_type: str) -> bytes:
|
|
if content_type in {"image/jpeg", "image/jpg"}:
|
|
return strip_jpeg_exif(data)
|
|
if content_type == "image/png":
|
|
return strip_png_metadata(data)
|
|
return data
|
|
|
|
|
|
def save_media(profile_id: str, entry_id: str, filename: str, content_type: str, data: bytes) -> dict:
|
|
entry = get_entry(profile_id, entry_id)
|
|
if not data:
|
|
raise StoreError("empty_media", "Datei ist leer")
|
|
kind = sniff_content_type(data, content_type)
|
|
if kind not in ALLOWED_TYPES:
|
|
raise StoreError("unsupported_media", "Nur Bilder (JPEG, PNG, GIF, WebP) und Videos (MP4, WebM, MOV) sind erlaubt")
|
|
is_video = kind in VIDEO_TYPES
|
|
limit = VIDEO_MAX_BYTES if is_video else IMAGE_MAX_BYTES
|
|
if len(data) > limit:
|
|
label = "Video" if is_video else "Bild"
|
|
mb = limit // (1024 * 1024)
|
|
raise StoreError("media_too_large", f"{label} ist größer als {mb} MB")
|
|
payload = data if is_video else prepare_image(data, kind)
|
|
asset_id = str(uuid.uuid4())
|
|
ext = ALLOWED_TYPES[kind]
|
|
rel_path = f"{profile_id}/{asset_id}{ext}"
|
|
dest = MEDIA_ROOT / rel_path
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_bytes(payload)
|
|
prefix = "video" if is_video else "image"
|
|
safe_name = Path(filename or f"{prefix}{ext}").name
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO media_assets
|
|
(id, profile_id, entry_id, entry_version_id, filename, content_type, rel_path)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(asset_id, profile_id, entry_id, entry.get("current_version_id"), safe_name, kind, rel_path),
|
|
)
|
|
row = row_to_dict(conn.execute("SELECT * FROM media_assets WHERE id = ?", (asset_id,)).fetchone())
|
|
return public_media(row)
|
|
|
|
|
|
def save_image(profile_id: str, entry_id: str, filename: str, content_type: str, data: bytes) -> dict:
|
|
return save_media(profile_id, entry_id, filename, content_type, data)
|
|
|
|
|
|
def list_media(profile_id: str, entry_id: str) -> list[dict]:
|
|
get_entry(profile_id, entry_id)
|
|
with get_db() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT * FROM media_assets
|
|
WHERE profile_id = ? AND entry_id = ?
|
|
ORDER BY created
|
|
""",
|
|
(profile_id, entry_id),
|
|
).fetchall()
|
|
return [public_media(row_to_dict(row)) for row in rows]
|
|
|
|
|
|
def get_media(profile_id: str, media_id: str) -> tuple[dict, Path]:
|
|
with get_db() as conn:
|
|
row = _owned(conn, "media_assets", media_id, profile_id)
|
|
if not row:
|
|
raise StoreError("not_found", "Medium nicht gefunden", 404)
|
|
path = MEDIA_ROOT / row["rel_path"]
|
|
if not path.is_file():
|
|
raise StoreError("not_found", "Mediendatei fehlt", 404)
|
|
return public_media(row), path
|
|
|
|
|
|
def delete_media(profile_id: str, media_id: str) -> dict:
|
|
row, path = get_media(profile_id, media_id)
|
|
with get_db() as conn:
|
|
conn.execute("DELETE FROM media_assets WHERE id = ? AND profile_id = ?", (media_id, profile_id))
|
|
try:
|
|
path.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
return {"id": media_id, "deleted": True}
|