diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index dfc24af..dfd66a3 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -15,6 +15,7 @@ api_bp.add_url_rule("/health", view_func=health.get_health, methods=["GET"]) def all_blueprints() -> list[Blueprint]: from .gallery import gallery_bp + from .import_admin import import_admin_bp from .settings import settings_bp from .tags import tags_bp - return [api_bp, gallery_bp, tags_bp, settings_bp] + return [api_bp, gallery_bp, tags_bp, settings_bp, import_admin_bp] diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py new file mode 100644 index 0000000..7eb5884 --- /dev/null +++ b/backend/app/api/import_admin.py @@ -0,0 +1,151 @@ +"""Import admin API: trigger scan, list tasks, retry, clear.""" + +from datetime import datetime, timedelta, timezone + +from quart import Blueprint, jsonify, request +from sqlalchemy import delete, select, update + +from ..extensions import make_engine, make_session_factory +from ..models import ImportBatch, ImportTask + +import_admin_bp = Blueprint("import_admin", __name__, url_prefix="/api/import") + +_engine = None +_Session = None + + +def _session_factory(): + global _engine, _Session + if _engine is None: + _engine = make_engine() + _Session = make_session_factory(_engine) + return _Session + + +@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(): + Session = _session_factory() + async with 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 + + Session = _session_factory() + async with 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(): + Session = _session_factory() + async with 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(timezone.utc) - timedelta(days=int(age_days)) if age_days else None + ) + + Session = _session_factory() + async with 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}) diff --git a/tests/test_api_import_admin.py b/tests/test_api_import_admin.py new file mode 100644 index 0000000..62461a4 --- /dev/null +++ b/tests/test_api_import_admin.py @@ -0,0 +1,90 @@ +from datetime import datetime, timedelta, timezone + +import pytest + +from backend.app import create_app +from backend.app.celery_app import celery +from backend.app.models import ImportBatch, ImportTask + + +@pytest.fixture(autouse=True) +def eager(): + celery.conf.task_always_eager = True + yield + celery.conf.task_always_eager = False + + +@pytest.fixture +async def app(): + return create_app() + + +@pytest.fixture +async def client(app): + async with app.test_client() as c: + yield c + + +@pytest.mark.asyncio +async def test_trigger_rejects_unknown_mode(client): + resp = await client.post("/api/import/trigger", json={"mode": "deep"}) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_status_when_idle(client): + resp = await client.get("/api/import/status") + body = await resp.get_json() + assert body["active_batch"] is None + + +@pytest.mark.asyncio +async def test_list_tasks_with_status_filter(client, db): + batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick") + db.add(batch) + await db.flush() + for status in ("complete", "failed", "complete"): + db.add(ImportTask( + batch_id=batch.id, source_path="/x", task_type="media", status=status, + finished_at=datetime.now(timezone.utc), + )) + await db.commit() + + resp = await client.get("/api/import/tasks?status=failed") + body = await resp.get_json() + assert len(body["tasks"]) == 1 + assert body["tasks"][0]["status"] == "failed" + + +@pytest.mark.asyncio +async def test_retry_failed(client, db): + batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick") + db.add(batch) + await db.flush() + db.add(ImportTask(batch_id=batch.id, source_path="/x", task_type="media", status="failed")) + await db.commit() + + resp = await client.post("/api/import/retry-failed") + body = await resp.get_json() + assert body["retried"] == 1 + + +@pytest.mark.asyncio +async def test_clear_completed(client, db): + batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick") + db.add(batch) + await db.flush() + long_ago = datetime.now(timezone.utc) - timedelta(days=30) + db.add(ImportTask( + batch_id=batch.id, source_path="/x", task_type="media", + status="complete", finished_at=long_ago, + )) + db.add(ImportTask( + batch_id=batch.id, source_path="/y", task_type="media", + status="complete", finished_at=datetime.now(timezone.utc), + )) + await db.commit() + + resp = await client.post("/api/import/clear-completed", json={"age_days": 7}) + body = await resp.get_json() + assert body["deleted"] == 1