add routes and upload functions

2025-10-24 08:25:57 +02:00
parent 175f382ec0
commit 8c2259d1de
7 changed files with 583 additions and 0 deletions
@@ -0,0 +1,79 @@
+from my_modules.file_helper_functions import verify_signed_url
+from my_modules.decoratory.header import login_required
+from my_modules.app.setup import LIMITER
+from my_modules.app.logger import logger
+
+from quart import Blueprint, request, session, Response, send_from_directory, render_template, abort, current_app
+from datetime import datetime, timezone
+
+side_main_bp = Blueprint('side_main', __name__)
+
+@side_main_bp.route('/')
+@LIMITER.limit("10 per minute")
+async def index():
+  if session.get("user") is not None:
+    return await render_template("views/webpage/upload.htm")
+  return await render_template("views/webpage/index.htm")
+
+@side_main_bp.route('/files')
+@LIMITER.limit("10 per minute")
+@login_required
+async def files(user):
+  files_data = await current_app.edgedb.get_files(current_datetime=datetime.now(timezone.utc), user_id=user['sub'])
+  return await render_template("views/webpage/files_list.htm", files=files_data)
+
+@side_main_bp.route('/files/<file_id>/info')
+@LIMITER.limit("10 per minute")
+@login_required
+async def file_info(file_id, user):
+  files_data = await current_app.edgedb.get_files(user_id=user['sub'])
+  return await render_template("views/webpage/.htm", files=files_data)
+
+@side_main_bp.route('/files/<file_id>/edit')
+@LIMITER.limit("10 per minute")
+@login_required
+async def file_edit(file_id, user):
+  files_data = await current_app.edgedb.get_files(user_id=user['sub'])
+  return await render_template("views/webpage/.htm", files=files_data)
+
+def is_expired(expires_at):
+  if not expires_at:
+    return False
+  if expires_at.tzinfo is None:
+    expires_at = expires_at.replace(tzinfo=timezone.utc)
+  else:
+    expires_at = expires_at.astimezone(timezone.utc)
+  return expires_at <= datetime.now(timezone.utc)
+
+@side_main_bp.route("/-<file_id>")
+@LIMITER.limit("10 per minute")
+async def serve_file(file_id: str):
+  file_data = await current_app.edgedb.get_file(file_id=file_id)
+  if not file_data:
+    abort(404)
+
+  if is_expired(file_data.get("expires_at")):
+    return Response("This file has expired.", status=410, headers={
+      "Cache-Control": "no-store",
+      "X-Content-Type-Options": "nosniff",
+    })
+
+  file_name = file_data.get("file_name")
+  content_type = file_data.get("content_type") or "application/octet-stream"
+
+  force_download = request.args.get("download") in {"1", "true", "yes"}
+
+  path = current_app.upload_folder / file_name
+  if not path.exists() or not path.is_file():
+    abort(404)
+
+  return await send_from_directory(
+    directory=current_app.upload_folder,
+    file_name=file_name,
+    mimetype=content_type,
+    as_attachment=force_download,
+    attachment_filename=file_name,
+    conditional=True,
+    cache_timeout=60,
+    last_modified=path.stat().st_mtime
+  )
@@ -0,0 +1,216 @@
+from my_modules.file_helper_functions import generate_short_id
+from my_modules.decoratory.header import login_required
+
+from quart import Blueprint, request, jsonify, current_app
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+import aiofiles, asyncio, re
+
+upload_bp = Blueprint("upload_bp", __name__)
+
+# --- Helpers -----------------------------------------------------
+
+PRESET_H = re.compile(r"^(\d+)h$")
+PRESET_D = re.compile(r"^(\d+)d$")
+
+def iso_stamp_filename(prefix: str, ext: str) -> str:
+  """Generate timestamped filename, e.g. pasted-2025-10-23T121212Z.png"""
+  ts = datetime.now(timezone.utc).isoformat()
+  ts = ts.replace(":", "").split(".")[0]
+  if ts.endswith("+00:00"):
+    ts = ts.replace("+00:00", "Z")
+  return f"{prefix}-{ts}.{ext}"
+
+def safe_name(name: str) -> str:
+  """Restrict filename to safe ASCII subset."""
+  return re.sub(r"[^A-Za-z0-9._-]", "_", name)
+
+def parse_expires(value: str | None) -> datetime | None:
+  """Parse expiration presets or ISO datetime."""
+  if not value:
+    return None
+  value = value.strip()
+  if m := PRESET_H.match(value):
+    return datetime.now(timezone.utc) + timedelta(hours=int(m.group(1)))
+  if m := PRESET_D.match(value):
+    return datetime.now(timezone.utc) + timedelta(days=int(m.group(1)))
+  try:
+    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
+  except Exception:
+    return None
+
+def format_size(num_bytes: int) -> str:
+  """Return a human-readable file size (e.g., '2.3 MB', '10 Bytes')."""
+  if num_bytes < 1024:
+    return f"{num_bytes} Byte{'s' if num_bytes != 1 else ''}"
+
+  units = ["KB", "MB", "GB", "TB", "PB", "EB"]
+  size = float(num_bytes)
+  for unit in units:
+    size /= 1024.0
+    if size < 1024.0 or unit == units[-1]:
+      # 1 decimal place; drop trailing .0 (optional)
+      val = f"{size:.1f}"
+      if val.endswith(".0"):
+        val = val[:-2]
+      return f"{val} {unit}"
+  return f"{num_bytes} Bytes"  # fallback
+
+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
+
+def ensure_utc(dt):
+    """Ensure a timezone-aware UTC datetime or None."""
+    if dt is None:
+        return None
+    if dt.tzinfo is None:
+        return dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+# --- 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)
+
+    fname = safe_name(fname)
+    path = current_app.upload_folder / fname
+
+    data = await read_all(uploaded)
+
+    # write to disk
+    async with aiofiles.open(path, "wb") as f:
+      await f.write(data)
+
+    size_bytes = len(data)
+    file_size_pretty = format_size(size_bytes)
+
+    await current_app.edgedb.add_file(
+      file_id=generate_short_id(),
+      file_name=fname,
+      file_size=file_size_pretty,
+      note=note,
+      content_type=content_type,
+      uploaded_at=datetime.now(timezone.utc),
+      expires_at=expires_at_dt,
+      user_id=user['sub'],
+    )
+
+  # --- text upload path ---
+  elif text.strip():
+    data = text.encode("utf-8")
+    fname = iso_stamp_filename("pasted", "txt")
+    path = current_app.upload_folder / fname
+
+    async with aiofiles.open(path, "wb") as f:
+      await f.write(data)
+
+    size_bytes = len(data)
+    file_size_pretty = format_size(size_bytes)
+
+    await current_app.edgedb.add_file(
+      file_id=generate_short_id(),
+      file_name=fname,
+      file_size=file_size_pretty,
+      note=note,
+      content_type="text/plain",
+      uploaded_at=datetime.now(timezone.utc),
+      expires_at=expires_at_dt,
+      user_id=user['sub'],
+    )
+
+  return jsonify({"ok": True})
+
+# --- Background cleanup ------------------------------------------------------
+
+async def cleanup_task():
+  """Hourly cleanup of expired files based on EdgeDB."""
+  await asyncio.sleep(3)  # allow app startup
+  while True:
+    try:
+      now = datetime.now(timezone.utc)
+
+      expired = await current_app.edgedb.get_expired_files(now)
+      if not expired:
+        await asyncio.sleep(3600)
+        continue
+
+      upload_dir: Path = current_app.upload_folder  # ensure Path
+      removed_ids: list[str] = []
+
+      for rec in expired:
+        try:
+          # Defensive: only touch files under your upload dir
+          fpath = (upload_dir / rec['file_name']).resolve()
+          if upload_dir.resolve() in fpath.parents or fpath == upload_dir.resolve():
+            fpath.unlink(missing_ok=True)
+            removed_ids.append(rec['file_id'])
+          else:
+            current_app.logger.warning("Refusing to delete outside upload dir: %s", fpath)
+        except Exception as e:
+          current_app.logger.exception("Failed to delete file %s (%s)", rec['file_name'], rec['file_id'])
+
+      # Remove DB rows for files we actually deleted from disk
+      if removed_ids:
+        try:
+          await current_app.edgedb.delete_files_by_ids(removed_ids)
+        except Exception:
+          current_app.logger.exception("Failed to delete DB rows for expired files")
+
+    except Exception:
+      current_app.logger.exception("Cleanup task iteration failed")
+
+    await asyncio.sleep(3600)  # every hour
+
+
+@upload_bp.before_app_serving
+async def start_cleanup():
+  asyncio.create_task(cleanup_task())