JakeBreath
99f617d296
Footer: - Left is now 'Backend Storage:' with a capacity bar (blue, peach at 80%, red at 95% per DESIGN.md) and a used/total/free tooltip; the watched folder path is no longer printed. /api/status/ returns a compact storage summary instead of the path (the full storage page still shows paths to authenticated users). - Centre shows the backend API origin (empty = same origin). Staff get a link to /setup to point the browser elsewhere; everyone else sees it as plain text. The Account 'Backend connection' card is gone — this is installation plumbing, not a per-user setting. - Design spec updated to match. Staff role: - The custom role did nothing on several endpoints that only accepted Django's is_staff/is_superuser. One canonical check now exists: User.is_app_staff (superuser, Django staff, or the staff role), used by the stats/users APIs, item object permissions, can_delete, upload/ similarity/download/match querysets, and the management commands (which also pick staff-role accounts for e621 sync/match and file ownership). Verified with a role-only staff account (is_staff/is_superuser false): stats/users 200, all 32 downloads + 2 scans visible, others' items editable; the same account as role=user gets 403 for all of those.
114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.management.base import BaseCommand
|
|
from django.db.models import Q
|
|
|
|
from apps.library.models import MediaLocation
|
|
from apps.library.services import ALLOWED_EXTENSIONS, index_file
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Scan the watched folder and index media files into the library."
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
"--prune",
|
|
action="store_true",
|
|
help="Remove locations whose files no longer exist on disk.",
|
|
)
|
|
parser.add_argument(
|
|
"--user",
|
|
help="Username that owns newly indexed files (default: first superuser).",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
folder = Path(settings.WATCHED_FOLDER)
|
|
if not folder.is_dir():
|
|
self.stderr.write(
|
|
self.style.ERROR(f"Watched folder does not exist: {folder}")
|
|
)
|
|
return
|
|
|
|
owner = None
|
|
if options["user"]:
|
|
owner = User.objects.filter(username=options["user"]).first()
|
|
if owner is None:
|
|
self.stderr.write(
|
|
self.style.ERROR(f"User not found: {options['user']}")
|
|
)
|
|
return
|
|
else:
|
|
owner = (
|
|
User.objects.filter(
|
|
Q(is_superuser=True)
|
|
| Q(is_staff=True)
|
|
| Q(role=User.ROLE_STAFF)
|
|
)
|
|
.order_by("-is_superuser", "id")
|
|
.first()
|
|
)
|
|
if owner is None:
|
|
self.stderr.write(
|
|
self.style.WARNING(
|
|
"No staff user found; scanned files will have no owner. "
|
|
"Use --user <name> to assign one."
|
|
)
|
|
)
|
|
if owner is not None:
|
|
self.stdout.write(f"New files will be owned by {owner.username}.")
|
|
|
|
self.stdout.write(f"Scanning {folder} ...")
|
|
seen = set()
|
|
new_items = 0
|
|
new_locations = 0
|
|
updated_locations = 0
|
|
skipped = 0
|
|
|
|
for root, dirs, files in os.walk(folder):
|
|
dirs.sort()
|
|
for name in sorted(files):
|
|
if Path(name).suffix.lower() not in ALLOWED_EXTENSIONS:
|
|
continue
|
|
path = Path(root) / name
|
|
try:
|
|
item, created_item, _location, created_location = index_file(
|
|
path, folder
|
|
)
|
|
except OSError as exc:
|
|
skipped += 1
|
|
self.stderr.write(f"Skipped {path}: {exc}")
|
|
continue
|
|
if owner is not None and item.uploaded_by_id is None:
|
|
item.uploaded_by = owner
|
|
item.save(update_fields=["uploaded_by", "updated_at"])
|
|
seen.add(str(path))
|
|
new_items += int(created_item)
|
|
if created_location:
|
|
new_locations += 1
|
|
else:
|
|
updated_locations += 1
|
|
processed = len(seen)
|
|
if processed % 200 == 0:
|
|
self.stdout.write(f"Processed {processed} files ...")
|
|
|
|
if options["prune"]:
|
|
stale = MediaLocation.objects.exclude(path__in=seen)
|
|
pruned = stale.count()
|
|
stale.delete()
|
|
self.stdout.write(f"Pruned {pruned} missing location(s).")
|
|
|
|
self.stdout.write(
|
|
self.style.SUCCESS(
|
|
f"Done. {len(seen)} file(s) on disk, "
|
|
f"{new_items} new item(s), "
|
|
f"{new_locations} new location(s), "
|
|
f"{updated_locations} refreshed, "
|
|
f"{skipped} skipped."
|
|
)
|
|
)
|