162 lines
5.4 KiB
Python
162 lines
5.4 KiB
Python
"""scan tasks:
|
|
|
|
* ``scan_directory`` — walks the configured import path, creates an
|
|
ImportBatch, enumerates files into ImportTasks, and enqueues
|
|
import_media_file for each.
|
|
* ``tick_due_sources`` (FC-3d) — periodic Beat-fired scan that fires
|
|
``download_source.delay()`` for sources due for a check.
|
|
"""
|
|
|
|
import asyncio
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from ..celery_app import celery
|
|
from ..config import get_config
|
|
from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask
|
|
from ..services.scheduler_service import select_due_sources
|
|
|
|
|
|
def _iter_import_files(import_root: Path):
|
|
"""Every regular file except sidecar .json, dotfiles, and .partial
|
|
temp files. The Importer dispatches by kind (media / archive /
|
|
other) — scan no longer filters to media (FC-2d-iii)."""
|
|
for entry in import_root.rglob("*"):
|
|
if not entry.is_file():
|
|
continue
|
|
if entry.name.startswith("."):
|
|
continue
|
|
if entry.suffix.lower() in (".json", ".partial"):
|
|
continue
|
|
yield entry
|
|
|
|
|
|
def _sync_session_factory():
|
|
cfg = get_config()
|
|
engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True)
|
|
return sessionmaker(engine, expire_on_commit=False)
|
|
|
|
|
|
@celery.task(name="backend.app.tasks.scan.scan_directory", bind=True)
|
|
def scan_directory(self, triggered_by: str = "manual",
|
|
mode: str = "quick") -> int:
|
|
"""Walks the import root and creates ImportTasks. `mode` is 'quick'
|
|
or 'deep'; deep also enqueues a library-wide phash backfill and (via
|
|
Importer.deep) re-derives on already-imported files. Returns the
|
|
batch id."""
|
|
SessionLocal = _sync_session_factory()
|
|
with SessionLocal() as session:
|
|
settings = session.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
import_root = Path(settings.import_scan_path)
|
|
|
|
batch = ImportBatch(
|
|
triggered_by=triggered_by,
|
|
source_path=str(import_root),
|
|
scan_mode=mode,
|
|
status="running",
|
|
)
|
|
session.add(batch)
|
|
session.flush()
|
|
batch_id = batch.id
|
|
|
|
# Walk and enumerate.
|
|
files_seen = 0
|
|
for entry in _iter_import_files(import_root):
|
|
try:
|
|
size = entry.stat().st_size
|
|
except OSError:
|
|
size = None
|
|
task = ImportTask(
|
|
batch_id=batch_id,
|
|
source_path=str(entry),
|
|
task_type="media",
|
|
status="pending",
|
|
size_bytes=size,
|
|
)
|
|
session.add(task)
|
|
files_seen += 1
|
|
|
|
batch.total_files = files_seen
|
|
session.commit()
|
|
|
|
# Now enqueue import_media_file for each pending task.
|
|
for task in session.execute(
|
|
select(ImportTask).where(ImportTask.batch_id == batch_id)
|
|
).scalars():
|
|
task.status = "queued"
|
|
session.add(task)
|
|
from .import_file import import_media_file
|
|
|
|
import_media_file.delay(task.id)
|
|
session.commit()
|
|
|
|
if mode == "deep":
|
|
from .maintenance import backfill_phash
|
|
|
|
backfill_phash.delay()
|
|
|
|
return batch_id
|
|
|
|
|
|
# --- FC-3d: periodic source-check tick ------------------------------------
|
|
|
|
|
|
def _async_session_factory():
|
|
cfg = get_config()
|
|
engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
|
|
return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine
|
|
|
|
|
|
async def _tick_due_sources_async() -> dict:
|
|
factory, engine = _async_session_factory()
|
|
try:
|
|
async with factory() as session:
|
|
due = await select_due_sources(session)
|
|
if not due:
|
|
return {
|
|
"due": 0, "queued": 0, "skipped_in_flight": 0,
|
|
"tick_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
due_ids = [s.id for s in due]
|
|
in_flight_rows = (await session.execute(
|
|
select(DownloadEvent.source_id).where(
|
|
DownloadEvent.source_id.in_(due_ids),
|
|
DownloadEvent.status.in_(["pending", "running"]),
|
|
)
|
|
)).scalars().all()
|
|
in_flight = set(in_flight_rows)
|
|
|
|
# Local import: keep top of module light, avoid circular import
|
|
# at celery task discovery time.
|
|
from .download import download_source
|
|
|
|
queued = 0
|
|
for s in due:
|
|
if s.id in in_flight:
|
|
continue
|
|
session.add(DownloadEvent(source_id=s.id, status="pending"))
|
|
await session.commit()
|
|
download_source.delay(s.id)
|
|
queued += 1
|
|
return {
|
|
"due": len(due),
|
|
"queued": queued,
|
|
"skipped_in_flight": len(in_flight),
|
|
"tick_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@celery.task(name="backend.app.tasks.scan.tick_due_sources", bind=True)
|
|
def tick_due_sources(self) -> dict:
|
|
"""FC-3d: scan for sources due for a periodic check and fire
|
|
``download_source.delay()`` for each."""
|
|
return asyncio.run(_tick_due_sources_async())
|