137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
"""Import admin API: trigger scan, list tasks, retry, clear."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from quart import Blueprint, jsonify, request
|
|
from sqlalchemy import delete, select, update
|
|
|
|
from ..extensions import get_session
|
|
from ..models import ImportBatch, ImportTask
|
|
|
|
import_admin_bp = Blueprint("import_admin", __name__, url_prefix="/api/import")
|
|
|
|
|
|
@import_admin_bp.route("/trigger", methods=["POST"])
|
|
async def trigger_scan():
|
|
body = await request.get_json(silent=True) or {}
|
|
mode = body.get("mode", "quick")
|
|
if mode != "quick":
|
|
# FC-2d will accept 'deep'. For now, reject anything else.
|
|
return jsonify({"error": f"mode {mode!r} not supported in FC-2a; use 'quick'"}), 400
|
|
|
|
# Enqueue the scan task in Celery. The task creates the batch itself.
|
|
from ..tasks.scan import scan_directory
|
|
|
|
async_result = scan_directory.delay(triggered_by="manual")
|
|
return jsonify({"celery_task_id": async_result.id, "mode": mode}), 202
|
|
|
|
|
|
@import_admin_bp.route("/status", methods=["GET"])
|
|
async def status():
|
|
async with get_session() as session:
|
|
active = (
|
|
await session.execute(
|
|
select(ImportBatch)
|
|
.where(ImportBatch.status == "running")
|
|
.order_by(ImportBatch.started_at.desc())
|
|
.limit(1)
|
|
)
|
|
).scalar_one_or_none()
|
|
payload = {"active_batch": None}
|
|
if active:
|
|
payload["active_batch"] = {
|
|
"id": active.id,
|
|
"total_files": active.total_files,
|
|
"imported": active.imported,
|
|
"skipped": active.skipped,
|
|
"failed": active.failed,
|
|
"started_at": active.started_at.isoformat(),
|
|
}
|
|
return jsonify(payload)
|
|
|
|
|
|
@import_admin_bp.route("/tasks", methods=["GET"])
|
|
async def list_tasks():
|
|
status_filter = request.args.get("status")
|
|
try:
|
|
limit = min(int(request.args.get("limit", "50")), 200)
|
|
except ValueError:
|
|
return jsonify({"error": "limit must be an integer"}), 400
|
|
cursor_raw = request.args.get("cursor")
|
|
cursor_id = int(cursor_raw) if cursor_raw else None
|
|
|
|
async with get_session() as session:
|
|
stmt = select(ImportTask).order_by(ImportTask.created_at.desc(), ImportTask.id.desc())
|
|
if status_filter:
|
|
stmt = stmt.where(ImportTask.status == status_filter)
|
|
if cursor_id is not None:
|
|
stmt = stmt.where(ImportTask.id < cursor_id)
|
|
stmt = stmt.limit(limit + 1)
|
|
rows = (await session.execute(stmt)).scalars().all()
|
|
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
return jsonify({
|
|
"tasks": [
|
|
{
|
|
"id": t.id,
|
|
"batch_id": t.batch_id,
|
|
"source_path": t.source_path,
|
|
"task_type": t.task_type,
|
|
"status": t.status,
|
|
"result_image_id": t.result_image_id,
|
|
"error": t.error,
|
|
"size_bytes": t.size_bytes,
|
|
"created_at": t.created_at.isoformat(),
|
|
"started_at": t.started_at.isoformat() if t.started_at else None,
|
|
"finished_at": t.finished_at.isoformat() if t.finished_at else None,
|
|
}
|
|
for t in rows
|
|
],
|
|
"next_cursor": rows[-1].id if has_more and rows else None,
|
|
})
|
|
|
|
|
|
@import_admin_bp.route("/retry-failed", methods=["POST"])
|
|
async def retry_failed():
|
|
async with get_session() as session:
|
|
failed_ids = (
|
|
await session.execute(select(ImportTask.id).where(ImportTask.status == "failed"))
|
|
).scalars().all()
|
|
if not failed_ids:
|
|
return jsonify({"retried": 0})
|
|
await session.execute(
|
|
update(ImportTask)
|
|
.where(ImportTask.id.in_(failed_ids))
|
|
.values(status="queued", error=None, started_at=None, finished_at=None)
|
|
)
|
|
await session.commit()
|
|
|
|
from ..tasks.import_file import import_media_file
|
|
for tid in failed_ids:
|
|
import_media_file.delay(tid)
|
|
|
|
return jsonify({"retried": len(failed_ids)})
|
|
|
|
|
|
@import_admin_bp.route("/clear-completed", methods=["POST"])
|
|
async def clear_completed():
|
|
body = await request.get_json(silent=True) or {}
|
|
age_days = body.get("age_days", 0)
|
|
status_filter = body.get("status", ["complete", "skipped"])
|
|
if not isinstance(status_filter, list):
|
|
return jsonify({"error": "status must be a list"}), 400
|
|
|
|
cutoff = (
|
|
datetime.now(UTC) - timedelta(days=int(age_days)) if age_days else None
|
|
)
|
|
|
|
async with get_session() as session:
|
|
stmt = delete(ImportTask).where(ImportTask.status.in_(status_filter))
|
|
if cutoff is not None:
|
|
stmt = stmt.where(ImportTask.finished_at < cutoff)
|
|
result = await session.execute(stmt)
|
|
await session.commit()
|
|
|
|
return jsonify({"deleted": result.rowcount or 0})
|