peter
8206384c91
The owner-filter check missed an aliased() model, an import alias, .in_() and filter_by; four live inline filters on aliased VideoState go through owned_by now. Cross-unit writes are seen under any import name, through sa.update and via __table__. Core modules may commit only where CORE_COMMITS names the reason; trim_read and audit.gc flush now and their callers commit. The pin target, the debug-session header and a link's target fetch through owned_row. The docs claim what the fence holds, and say what is beyond it.
405 lines
15 KiB
Python
405 lines
15 KiB
Python
"""Capture's reads: what the library, the download tray, the share dialogs and the public watch pages
|
|
show. Each returns rows or plain values and writes nothing; whose a row is goes through
|
|
`userscope.owned_by` / `owned_row`, so a row of someone else's reads as a missing one.
|
|
|
|
The HTTP layer (`routes.downloads`, `routes.files`, `routes.public`) asks here and serializes; the
|
|
writes are the service's (`service`, `folders`, `links`, `quota`)."""
|
|
|
|
from collections.abc import Sequence
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.apitokens import FilesCaller, Right
|
|
from app.collation import collated
|
|
from app.downloads import folders, links, quota
|
|
from app.downloads.models import (
|
|
DownloadJob,
|
|
DownloadLink,
|
|
DownloadProfile,
|
|
DownloadQuota,
|
|
DownloadShare,
|
|
FileFolder,
|
|
LibraryState,
|
|
MediaAsset,
|
|
)
|
|
from app.downloads.status import FINISHED, JobStatus
|
|
from app.models import User
|
|
from app.userscope import owned_by, owned_row
|
|
from app.utils import like_escape
|
|
|
|
# Cap a listing so a heavy library can't return an unbounded result set (the old Download Center's
|
|
# own cap; far beyond any real library).
|
|
MAX_LIST = 500
|
|
|
|
|
|
# --- jobs and their files ------------------------------------------------------------------
|
|
|
|
|
|
def own_job(db: Session, user: User, job_id: int) -> DownloadJob | None:
|
|
return owned_row(db, DownloadJob, job_id, user)
|
|
|
|
|
|
def job(db: Session, job_id: int) -> DownloadJob | None:
|
|
"""A job by id, whoever's — for a public link, which already names its job."""
|
|
return db.get(DownloadJob, job_id)
|
|
|
|
|
|
def accessible_job(db: Session, user: User, job_id: int) -> DownloadJob | None:
|
|
"""A file the caller may read: their own, or one shared with them."""
|
|
row = db.get(DownloadJob, job_id)
|
|
if row is None:
|
|
return None
|
|
if row.user_id == user.id or shared_with(db, job_id, user) is not None:
|
|
return row
|
|
return None
|
|
|
|
|
|
def readable_job(db: Session, caller: FilesCaller, job_id: int) -> DownloadJob | None:
|
|
"""A file `caller` may read: as `accessible_job` for the user; for a token, one of the user's
|
|
own in a folder it may read."""
|
|
if caller.grants is None:
|
|
return accessible_job(db, caller.user, job_id)
|
|
row = own_job(db, caller.user, job_id)
|
|
if row is None or not caller.may(row.folder_id, Right.READ):
|
|
return None
|
|
return row
|
|
|
|
|
|
def asset_of(db: Session, job: DownloadJob | None) -> MediaAsset | None:
|
|
"""The file under a job, if it has one yet."""
|
|
return db.get(MediaAsset, job.asset_id) if job is not None and job.asset_id else None
|
|
|
|
|
|
def assets_for(db: Session, jobs: Sequence[DownloadJob]) -> dict[int, MediaAsset]:
|
|
ids = {j.asset_id for j in jobs if j.asset_id}
|
|
if not ids:
|
|
return {}
|
|
rows = db.execute(select(MediaAsset).where(MediaAsset.id.in_(ids))).scalars()
|
|
return {a.id: a for a in rows}
|
|
|
|
|
|
def positions_for(db: Session, user: User, jobs: Sequence[DownloadJob]) -> dict[int, int]:
|
|
"""`user`'s resume point in each of `jobs` that has one (the library player's `library_states`)."""
|
|
ids = [j.id for j in jobs]
|
|
if not ids:
|
|
return {}
|
|
rows = db.execute(
|
|
select(LibraryState.job_id, LibraryState.position_seconds).where(
|
|
owned_by(LibraryState.user_id, user), LibraryState.job_id.in_(ids)
|
|
)
|
|
).all()
|
|
return {job_id: pos for job_id, pos in rows}
|
|
|
|
|
|
def source_keys(db: Session, jobs: Sequence[DownloadJob]) -> dict[int, str]:
|
|
"""Where each file came from (its source folder's key), for the list's Source column, the
|
|
"back to its source folder" move and the download tray's site: a file in a source folder has its
|
|
folder's key; one moved into an own folder (or someone else's, shared) is traced back by
|
|
`folders.source_key_for`."""
|
|
ids = {j.folder_id for j in jobs if j.folder_id is not None}
|
|
by_id = (
|
|
{f.id: f for f in db.execute(select(FileFolder).where(FileFolder.id.in_(ids))).scalars()}
|
|
if ids
|
|
else {}
|
|
)
|
|
out: dict[int, str] = {}
|
|
for j in jobs:
|
|
folder = by_id.get(j.folder_id) if j.folder_id is not None else None
|
|
if folder is not None and folder.kind == folders.FolderKind.SOURCE and folder.source_key:
|
|
out[j.id] = folder.source_key
|
|
else:
|
|
out[j.id] = folders.source_key_for(db, j)
|
|
return out
|
|
|
|
|
|
def queue(db: Session, user: User, recent: datetime) -> list[DownloadJob]:
|
|
"""The download tray: what is waiting, running, paused or failed, and what finished (done or
|
|
canceled) since `recent`. A done job is timed from when it became done, never from its last
|
|
write: a rename, a move or the GC's expiry warning touches an old file's row, and the tray would
|
|
announce it as just finished."""
|
|
return list(
|
|
db.execute(
|
|
select(DownloadJob)
|
|
.where(
|
|
owned_by(DownloadJob.user_id, user),
|
|
DownloadJob.status.not_in(FINISHED)
|
|
| ((DownloadJob.status == JobStatus.DONE) & (DownloadJob.ready_at >= recent))
|
|
| ((DownloadJob.status == JobStatus.CANCELED) & (DownloadJob.updated_at >= recent)),
|
|
)
|
|
.order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc())
|
|
.limit(MAX_LIST)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def statuses_by_source(db: Session, user: User, statuses: Sequence[str]) -> list[tuple[str, str]]:
|
|
"""(source_ref, status) of each of `user`'s jobs in one of `statuses`."""
|
|
return list(
|
|
db.execute(
|
|
select(DownloadJob.source_ref, DownloadJob.status).where(
|
|
owned_by(DownloadJob.user_id, user), DownloadJob.status.in_(tuple(statuses))
|
|
)
|
|
).tuples()
|
|
)
|
|
|
|
|
|
def library(
|
|
db: Session,
|
|
user: User,
|
|
*,
|
|
folder_id: int | None,
|
|
listable: set[int] | None,
|
|
shared: bool,
|
|
term: str,
|
|
sort: str,
|
|
order: str,
|
|
lang: str,
|
|
) -> list[DownloadJob]:
|
|
"""The caller's done files (optionally one folder's, or only in the folders a token may list),
|
|
or the ones shared with them; searched by name and channel, sorted by when they were added, by
|
|
name (in the reader's alphabet) or by size. A file is added to its owner's list when it is done,
|
|
not when it was asked for (a download that failed for days and then went through is new); one
|
|
shared with the caller, when it was shared."""
|
|
stmt = (
|
|
select(DownloadJob)
|
|
.join(MediaAsset, MediaAsset.id == DownloadJob.asset_id)
|
|
.where(DownloadJob.status == JobStatus.DONE)
|
|
)
|
|
if shared:
|
|
stmt = stmt.join(DownloadShare, DownloadShare.job_id == DownloadJob.id).where(
|
|
owned_by(DownloadShare.to_user_id, user)
|
|
)
|
|
else:
|
|
stmt = stmt.where(owned_by(DownloadJob.user_id, user))
|
|
if folder_id is not None:
|
|
stmt = stmt.where(DownloadJob.folder_id == folder_id)
|
|
elif listable is not None:
|
|
stmt = stmt.where(DownloadJob.folder_id.in_(listable))
|
|
name = func.coalesce(DownloadJob.display_name, MediaAsset.title, "")
|
|
if term:
|
|
like = f"%{like_escape(term)}%"
|
|
stmt = stmt.where(
|
|
or_(
|
|
name.ilike(like),
|
|
func.coalesce(DownloadJob.display_uploader, MediaAsset.uploader, "").ilike(like),
|
|
)
|
|
)
|
|
key = {
|
|
"added": DownloadShare.created_at if shared else DownloadJob.ready_at,
|
|
"name": collated(name, lang),
|
|
"size": MediaAsset.size_bytes,
|
|
}[sort]
|
|
first = key.asc() if order == "asc" else key.desc()
|
|
tie = DownloadJob.id.asc() if order == "asc" else DownloadJob.id.desc()
|
|
return list(db.execute(stmt.order_by(first.nulls_last(), tie).limit(MAX_LIST)).scalars())
|
|
|
|
|
|
def recent_jobs(db: Session) -> list[DownloadJob]:
|
|
"""Everyone's newest jobs, for the admin's list."""
|
|
return list(
|
|
db.execute(
|
|
select(DownloadJob)
|
|
.order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc())
|
|
.limit(MAX_LIST)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
# --- shares --------------------------------------------------------------------------------
|
|
|
|
|
|
def shared_with(db: Session, job_id: int, user: User | int) -> DownloadShare | None:
|
|
"""The grant that hands this job to `user`, if there is one."""
|
|
return db.execute(
|
|
select(DownloadShare).where(
|
|
DownloadShare.job_id == job_id, owned_by(DownloadShare.to_user_id, user)
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def shared_at(db: Session, user: User, jobs: Sequence[DownloadJob]) -> dict[int, datetime]:
|
|
rows = db.execute(
|
|
select(DownloadShare.job_id, DownloadShare.created_at).where(
|
|
owned_by(DownloadShare.to_user_id, user),
|
|
DownloadShare.job_id.in_([j.id for j in jobs]),
|
|
)
|
|
).all()
|
|
return {job_id: at for job_id, at in rows}
|
|
|
|
|
|
# --- folders -------------------------------------------------------------------------------
|
|
|
|
|
|
def own_folder(db: Session, user: User, folder_id: int) -> FileFolder | None:
|
|
return owned_row(db, FileFolder, folder_id, user)
|
|
|
|
|
|
def folders_by_id(db: Session, ids: set[int]) -> dict[int, FileFolder]:
|
|
"""These folders, by id — for a reader that already knows which ones (a token's grants)."""
|
|
if not ids:
|
|
return {}
|
|
return {f.id: f for f in db.execute(select(FileFolder).where(FileFolder.id.in_(ids))).scalars()}
|
|
|
|
|
|
def folders_with_totals(
|
|
db: Session, user: User, listable: set[int] | None
|
|
) -> list[tuple[FileFolder, int, int]]:
|
|
"""The user's folders (only `listable` ones, when given) with how many done files each holds and
|
|
their size. A source folder its files have all left stays (count 0)."""
|
|
done = (DownloadJob.folder_id == FileFolder.id) & (DownloadJob.status == JobStatus.DONE)
|
|
stmt = select(
|
|
FileFolder,
|
|
func.count(DownloadJob.id),
|
|
func.coalesce(func.sum(MediaAsset.size_bytes), 0),
|
|
).where(owned_by(FileFolder.user_id, user))
|
|
if listable is not None:
|
|
stmt = stmt.where(FileFolder.id.in_(listable))
|
|
rows = db.execute(
|
|
stmt.outerjoin(DownloadJob, done)
|
|
.outerjoin(MediaAsset, MediaAsset.id == DownloadJob.asset_id)
|
|
.group_by(FileFolder.id)
|
|
.order_by(FileFolder.kind.desc(), FileFolder.source_key, FileFolder.name, FileFolder.id)
|
|
).tuples()
|
|
return [(f, count, int(size or 0)) for f, count, size in rows]
|
|
|
|
|
|
def folder_totals(db: Session, folder_id: int) -> tuple[int, int]:
|
|
count, size = db.execute(
|
|
select(func.count(DownloadJob.id), func.coalesce(func.sum(MediaAsset.size_bytes), 0))
|
|
.select_from(DownloadJob)
|
|
.outerjoin(MediaAsset, MediaAsset.id == DownloadJob.asset_id)
|
|
.where(DownloadJob.folder_id == folder_id, DownloadJob.status == JobStatus.DONE)
|
|
).one()
|
|
return count, int(size)
|
|
|
|
|
|
def linked_folders(db: Session, folder_ids: list[int]) -> set[int]:
|
|
"""Which of these folders a live public link opens (one not past its expiry)."""
|
|
now = datetime.now(timezone.utc)
|
|
rows = db.execute(
|
|
select(DownloadLink.folder_id).where(
|
|
DownloadLink.folder_id.in_(folder_ids),
|
|
or_(DownloadLink.expires_at.is_(None), DownloadLink.expires_at > now),
|
|
)
|
|
).scalars()
|
|
return {folder_id for folder_id in rows if folder_id is not None}
|
|
|
|
|
|
# --- links ---------------------------------------------------------------------------------
|
|
|
|
|
|
def own_link(db: Session, user: User, link_id: int) -> DownloadLink | None:
|
|
"""A link to one of the user's files or folders."""
|
|
link = db.get(DownloadLink, link_id)
|
|
if link is None:
|
|
return None
|
|
target = (
|
|
owned_row(db, DownloadJob, link.job_id, user)
|
|
if link.job_id is not None
|
|
else owned_row(db, FileFolder, link.folder_id, user)
|
|
)
|
|
return link if target is not None else None
|
|
|
|
|
|
def links_to(
|
|
db: Session, *, job_id: int | None = None, folder_id: int | None = None
|
|
) -> list[DownloadLink]:
|
|
"""The links to one file or one folder, newest first."""
|
|
target = (
|
|
DownloadLink.job_id == job_id if job_id is not None else DownloadLink.folder_id == folder_id
|
|
)
|
|
return list(
|
|
db.execute(select(DownloadLink).where(target).order_by(DownloadLink.id.desc())).scalars()
|
|
)
|
|
|
|
|
|
def link_folder(db: Session, link: DownloadLink) -> FileFolder | None:
|
|
return db.get(FileFolder, link.folder_id) if link.folder_id is not None else None
|
|
|
|
|
|
def folder_job(db: Session, folder: FileFolder, job_id: int) -> DownloadJob | None:
|
|
"""One of the files a link to `folder` reaches (`links.in_folder`), while it is in it."""
|
|
return db.execute(
|
|
select(DownloadJob).where(DownloadJob.id == job_id, *links.in_folder(folder))
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def owner_suspended(db: Session, job: DownloadJob) -> bool:
|
|
"""The job's owner is suspended (or gone): the worker leaves their queued jobs where they are."""
|
|
owner = db.get(User, job.user_id)
|
|
return owner is None or owner.is_suspended
|
|
|
|
|
|
# --- profiles ------------------------------------------------------------------------------
|
|
|
|
|
|
def profiles_for(db: Session, user: User) -> list[DownloadProfile]:
|
|
"""The built-in format profiles, then the user's own."""
|
|
return list(
|
|
db.execute(
|
|
select(DownloadProfile)
|
|
.where(DownloadProfile.is_builtin.is_(True) | owned_by(DownloadProfile.user_id, user))
|
|
.order_by(
|
|
DownloadProfile.is_builtin.desc(), DownloadProfile.sort_order, DownloadProfile.id
|
|
)
|
|
).scalars()
|
|
)
|
|
|
|
|
|
def own_profile(db: Session, user: User, profile_id: int) -> DownloadProfile | None:
|
|
"""One of the user's own profiles (a built-in is no one's to change)."""
|
|
p = owned_row(db, DownloadProfile, profile_id, user)
|
|
return None if p is None or p.is_builtin else p
|
|
|
|
|
|
def usable_profile(db: Session, user: User, profile_id: int) -> DownloadProfile | None:
|
|
"""A profile the user may enqueue with: a built-in, or their own."""
|
|
p = db.get(DownloadProfile, profile_id)
|
|
if p is None or (not p.is_builtin and p.user_id != user.id):
|
|
return None
|
|
return p
|
|
|
|
|
|
def default_profile(db: Session) -> DownloadProfile | None:
|
|
"""The first built-in, for an enqueue that named no format."""
|
|
return db.execute(
|
|
select(DownloadProfile)
|
|
.where(DownloadProfile.is_builtin.is_(True))
|
|
.order_by(DownloadProfile.sort_order)
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
# --- storage and quota (admin) -------------------------------------------------------------
|
|
|
|
|
|
def storage_totals(db: Session) -> tuple[int, int, int]:
|
|
"""(files on disk, their bytes, every asset row). On disk is `quota.ON_DISK`: ready files, and
|
|
uploads waiting for or failed in their conversion — the same disk the footprint and the GC cap
|
|
count."""
|
|
on_disk = db.execute(
|
|
select(func.count(), func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(quota.ON_DISK)
|
|
).one()
|
|
total_assets = db.execute(select(func.count()).select_from(MediaAsset)).scalar_one()
|
|
return on_disk[0], int(on_disk[1]), total_assets
|
|
|
|
|
|
def done_files_per_user(db: Session) -> dict[int, int]:
|
|
"""How many finished files each user's Files lists."""
|
|
return dict(
|
|
db.execute(
|
|
select(DownloadJob.user_id, func.count())
|
|
.where(DownloadJob.status == JobStatus.DONE)
|
|
.group_by(DownloadJob.user_id)
|
|
)
|
|
.tuples()
|
|
.all()
|
|
)
|
|
|
|
|
|
def custom_quota(db: Session, user_id: int) -> DownloadQuota | None:
|
|
"""The admin's own limits for this user, if any were set."""
|
|
return db.get(DownloadQuota, user_id)
|