107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
from my_modules.decoratory.header import login_required
|
|
from my_modules.expiry import parse_expires, ensure_utc
|
|
from my_modules.file_meta import iso_stamp_filename, format_size
|
|
|
|
from quart import Blueprint, request, jsonify, current_app
|
|
import asyncio
|
|
|
|
upload_bp = Blueprint("upload_bp", __name__)
|
|
|
|
# --- Helpers -----------------------------------------------------
|
|
|
|
async def read_all(uploaded) -> bytes:
|
|
"""Read all bytes from an uploaded file, handling sync or async .read()."""
|
|
reader = getattr(uploaded, "read", None)
|
|
if reader is None:
|
|
return b""
|
|
if asyncio.iscoroutinefunction(reader):
|
|
return await reader()
|
|
|
|
data = reader()
|
|
if asyncio.iscoroutine(data):
|
|
return await data
|
|
return data
|
|
|
|
# --- Routes ------------------------------------------------------
|
|
|
|
@upload_bp.post("/api/upload")
|
|
@login_required
|
|
async def api_upload(user):
|
|
"""
|
|
POST /upload/api/upload
|
|
Accepts:
|
|
- multipart form with 'file' or 'text'
|
|
- 'expires' can be '1h', '7d', or ISO timestamp
|
|
- 'note' optional
|
|
"""
|
|
form = await request.form
|
|
files = await request.files
|
|
note = form.get("note", "")
|
|
expires_raw = form.get("expires", "")
|
|
text = form.get("text", "")
|
|
|
|
uploaded = files.get("file")
|
|
expires_at_dt = ensure_utc(parse_expires(expires_raw))
|
|
|
|
if not uploaded and not text.strip():
|
|
return jsonify({"ok": False, "error": "No content provided"}), 400
|
|
|
|
content_type = None
|
|
|
|
# --- binary upload path ---
|
|
if uploaded:
|
|
fname = uploaded.filename or ""
|
|
ctype = uploaded.mimetype or "application/octet-stream"
|
|
content_type = ctype
|
|
|
|
# generate filename if missing/placeholder
|
|
if not fname or fname.lower() in {"blob", "file"}:
|
|
ext = {
|
|
"image/png": "png",
|
|
"image/jpeg": "jpg",
|
|
"image/gif": "gif",
|
|
"image/webp": "webp",
|
|
"application/pdf": "pdf",
|
|
"text/plain": "txt",
|
|
}.get(ctype, "bin")
|
|
fname = iso_stamp_filename("pasted", ext)
|
|
|
|
data = await read_all(uploaded)
|
|
|
|
storage_id = await current_app.convex.send_to_storage(data=data, content_type=content_type)
|
|
|
|
size_bytes = len(data)
|
|
file_size_pretty = format_size(size_bytes)
|
|
|
|
await current_app.convex.add_file(
|
|
file_name=fname,
|
|
file_size=file_size_pretty,
|
|
note=note,
|
|
content_type=content_type,
|
|
expires_at=expires_at_dt,
|
|
storage_id=storage_id,
|
|
user_id=user['sub'],
|
|
)
|
|
|
|
# --- text upload path ---
|
|
elif text.strip():
|
|
data = text.encode("utf-8")
|
|
fname = iso_stamp_filename("pasted", "txt")
|
|
|
|
storage_id = await current_app.convex.send_to_storage(data=data, content_type="text/plain")
|
|
|
|
size_bytes = len(data)
|
|
file_size_pretty = format_size(size_bytes)
|
|
|
|
await current_app.convex.add_file(
|
|
file_name=fname,
|
|
file_size=file_size_pretty,
|
|
note=note,
|
|
content_type="text/plain",
|
|
expires_at=expires_at_dt,
|
|
storage_id=storage_id,
|
|
user_id=user['sub'],
|
|
)
|
|
|
|
return jsonify({"ok": True})
|