feat: file upload, allowlist, and download/preview

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
肉末
2026-07-14 12:47:58 +08:00
parent 0e59936e34
commit eda7010cae
4 changed files with 219 additions and 7 deletions
+54 -5
View File
@@ -7,7 +7,7 @@ from pathlib import Path
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.config import settings
from app import config
from app.models import Item
from app.schemas import ItemCreate, TTLOption
@@ -83,6 +83,57 @@ def create_text_item(db: Session, payload: ItemCreate, created_ip: str) -> Item:
return item
def create_file_item(
db: Session,
*,
file_name: str,
file_path: str,
mime: str,
size_bytes: int,
is_public: bool,
burn_after_read: bool,
ttl: TTLOption,
created_ip: str,
title: str | None = None,
) -> Item:
now = datetime.utcnow()
item = Item(
slug=_slug(),
kind="file",
title=(title or file_name)[:255],
body=None,
file_name=file_name,
file_path=file_path,
mime=mime,
size_bytes=size_bytes,
is_public=is_public,
burn_after_read=burn_after_read,
expires_at=_expires_at(ttl, now),
created_ip=created_ip,
created_at=now,
)
db.add(item)
db.commit()
db.refresh(item)
return item
def resolve_file_path(item: Item) -> Path | None:
if not item.file_path:
return None
path = Path(item.file_path)
if not path.is_absolute():
path = config.settings.data_dir / path
return path
def get_file_item(db: Session, slug: str) -> Item | None:
item = db.scalar(select(Item).where(Item.slug == slug))
if item is None or item.kind != "file" or _is_unavailable(item):
return None
return item
def get_item(db: Session, slug: str) -> Item | None:
item = db.scalar(select(Item).where(Item.slug == slug))
if item is None or _is_unavailable(item):
@@ -91,10 +142,8 @@ def get_item(db: Session, slug: str) -> Item | None:
item.view_count += 1
if item.burn_after_read:
item.burned = True
if item.file_path:
path = Path(item.file_path)
if not path.is_absolute():
path = settings.data_dir / path
path = resolve_file_path(item)
if path is not None:
try:
path.unlink(missing_ok=True)
except OSError: