肉末
133 lines
3.4 KiB
Python
133 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.models import Item
|
|
from app.schemas import ItemCreate, TTLOption
|
|
|
|
TTL_DELTAS: dict[TTLOption, timedelta | None] = {
|
|
"1h": timedelta(hours=1),
|
|
"24h": timedelta(days=1),
|
|
"7d": timedelta(days=7),
|
|
"never": None,
|
|
}
|
|
|
|
|
|
def _slug() -> str:
|
|
return secrets.token_urlsafe(8)
|
|
|
|
|
|
def _expires_at(ttl: TTLOption, now: datetime | None = None) -> datetime | None:
|
|
delta = TTL_DELTAS[ttl]
|
|
if delta is None:
|
|
return None
|
|
return (now or datetime.utcnow()) + delta
|
|
|
|
|
|
def _title_from_body(body: str) -> str:
|
|
first = body.split("\n", 1)[0].strip()
|
|
return first[:255]
|
|
|
|
|
|
def _is_unavailable(item: Item, now: datetime | None = None) -> bool:
|
|
if item.burned:
|
|
return True
|
|
now = now or datetime.utcnow()
|
|
if item.expires_at is not None and item.expires_at <= now:
|
|
return True
|
|
return False
|
|
|
|
|
|
def item_to_dict(item: Item, *, include_body: bool = True) -> dict:
|
|
data = {
|
|
"slug": item.slug,
|
|
"kind": item.kind,
|
|
"title": item.title,
|
|
"file_name": item.file_name,
|
|
"mime": item.mime,
|
|
"size_bytes": item.size_bytes,
|
|
"is_public": item.is_public,
|
|
"burn_after_read": item.burn_after_read,
|
|
"expires_at": item.expires_at,
|
|
"created_at": item.created_at,
|
|
"view_count": item.view_count,
|
|
}
|
|
if include_body:
|
|
data["body"] = item.body
|
|
return data
|
|
|
|
|
|
def create_text_item(db: Session, payload: ItemCreate, created_ip: str) -> Item:
|
|
now = datetime.utcnow()
|
|
title = payload.title if payload.title is not None else _title_from_body(payload.body)
|
|
item = Item(
|
|
slug=_slug(),
|
|
kind="text",
|
|
title=title[:255],
|
|
body=payload.body,
|
|
is_public=payload.is_public,
|
|
burn_after_read=payload.burn_after_read,
|
|
expires_at=_expires_at(payload.ttl, now),
|
|
created_ip=created_ip,
|
|
created_at=now,
|
|
)
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
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):
|
|
return 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
|
|
try:
|
|
path.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
def list_wall(
|
|
db: Session,
|
|
*,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
) -> tuple[list[Item], int]:
|
|
page = max(page, 1)
|
|
page_size = min(max(page_size, 1), 100)
|
|
now = datetime.utcnow()
|
|
|
|
filters = [
|
|
Item.is_public.is_(True),
|
|
Item.burned.is_(False),
|
|
(Item.expires_at.is_(None) | (Item.expires_at > now)),
|
|
]
|
|
total = db.scalar(select(func.count()).select_from(Item).where(*filters)) or 0
|
|
items = list(
|
|
db.scalars(
|
|
select(Item)
|
|
.where(*filters)
|
|
.order_by(Item.created_at.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
).all()
|
|
)
|
|
return items, total
|