FabledCurator/backend/app/tasks/admin.py

"""FC-3k: admin destructive Celery tasks.

Two long-running ops on the maintenance queue. task_run lifecycle is
captured automatically by FC-3i signals — these tasks just return
their summary dict so it lands in task_run.metadata (via Celery's
result backend) for the dashboard to surface.

Soft/hard time limits inherit the FC-3i recovery sweep: a runaway
task gets killed and flipped to status='timeout' by
recover_stalled_task_runs.
"""
from __future__ import annotations

import logging
from pathlib import Path

from sqlalchemy.exc import DBAPIError, OperationalError

from ..celery_app import celery
from ..services import cleanup_service
from ._sync_engine import sync_session_factory as _sync_session_factory

log = logging.getLogger(__name__)
IMAGES_ROOT = Path("/images")


@celery.task(
    name="backend.app.tasks.admin.delete_artist_cascade_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
)
def delete_artist_cascade_task(self, *, artist_id: int) -> dict:
    """Wraps cleanup_service.delete_artist_cascade. Returns the
    service's summary dict for FC-3i task_run.metadata capture."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        return cleanup_service.delete_artist_cascade(
            session, artist_id=artist_id, images_root=IMAGES_ROOT,
        )


@celery.task(
    name="backend.app.tasks.admin.bulk_delete_images_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=900, time_limit=1200,  # 15 min / 20 min
)
def bulk_delete_images_task(self, *, image_ids: list[int]) -> dict:
    """Wraps cleanup_service.delete_images."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        return cleanup_service.delete_images(
            session, image_ids=image_ids, images_root=IMAGES_ROOT,
        )


@celery.task(
    name="backend.app.tasks.admin.reextract_archive_attachments_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
)
def reextract_archive_attachments_task(self) -> dict:
    """Wraps cleanup_service.reextract_archive_attachments (#713 part 2):
    re-extract PostAttachments that are actually archives but were filed
    opaquely before the magic-byte gate, and link their members to the post."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        return cleanup_service.reextract_archive_attachments(
            session, images_root=IMAGES_ROOT,
        )


@celery.task(
    name="backend.app.tasks.admin.normalize_tags_task",
    bind=True,
    autoretry_for=(OperationalError, DBAPIError),
    retry_backoff=15, retry_backoff_max=180, max_retries=1,
    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
)
def normalize_tags_task(self) -> dict:
    """Wraps tag_service.normalize_existing_tags (#714): Title-Case the
    back-catalog and merge case/whitespace-variant duplicate tags via the
    tested async merge path. Runs under its own asyncio loop + per-task async
    engine (NullPool, disposed when the loop ends), mirroring download_source."""
    import asyncio

    from ..services.tag_service import normalize_existing_tags
    from ._async_session import async_session_factory

    async def _run() -> dict:
        async_factory, async_engine = async_session_factory()
        try:
            async with async_factory() as session:
                # normalize_existing_tags commits per group internally.
                return await normalize_existing_tags(session, dry_run=False)
        finally:
            await async_engine.dispose()

    return asyncio.run(_run())