backend/app/worker/tasks.py

"""
Celery tasks for scraping, freshness checks, and cleanup.
"""
import asyncio
import logging

from app.worker.celery_config import celery_app

logger = logging.getLogger(__name__)


@celery_app.task(name="app.worker.tasks.run_scheduled_scrapes")
def run_scheduled_scrapes():
    """Run scrapes for all active sources that are due."""
    asyncio.run(_run_scheduled_scrapes())


async def _run_scheduled_scrapes():
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import select
    from app.core.database import async_session_factory
    from app.models.scraper import JobSource, SourceStatus
    from app.scraper.engine import ScraperEngine

    async with async_session_factory() as db:
        # Find sources due for scraping
        result = await db.execute(
            select(JobSource).where(JobSource.status == SourceStatus.ACTIVE)
        )
        sources = result.scalars().all()

        engine = ScraperEngine(db)
        try:
            for source in sources:
                # Check if enough time has passed since last crawl
                if source.last_crawled_at:
                    hours_since = (
                        datetime.now(timezone.utc) - source.last_crawled_at
                    ).total_seconds() / 3600
                    if hours_since < source.crawl_frequency_hours:
                        continue

                logger.info(f"Starting scrape for source: {source.name}")
                run = await engine.run_source(source)
                logger.info(
                    f"Completed scrape for {source.name}: "
                    f"found={run.jobs_found}, new={run.jobs_new}, "
                    f"updated={run.jobs_updated}, status={run.status}"
                )

            await db.commit()
        finally:
            await engine.close()


@celery_app.task(name="app.worker.tasks.scrape_single_source")
def scrape_single_source(source_id: str):
    """Scrape a single source on demand."""
    asyncio.run(_scrape_single_source(source_id))


async def _scrape_single_source(source_id: str):
    import uuid
    from sqlalchemy import select
    from app.core.database import async_session_factory
    from app.models.scraper import JobSource
    from app.scraper.engine import ScraperEngine

    async with async_session_factory() as db:
        result = await db.execute(
            select(JobSource).where(JobSource.id == uuid.UUID(source_id))
        )
        source = result.scalar_one_or_none()
        if not source:
            logger.error(f"Source not found: {source_id}")
            return

        engine = ScraperEngine(db)
        try:
            run = await engine.run_source(source)
            await db.commit()
            logger.info(f"Single scrape completed for {source.name}: status={run.status}")
        finally:
            await engine.close()


@celery_app.task(name="app.worker.tasks.check_job_freshness")
def check_job_freshness():
    """Check active jobs for staleness and mark expired ones."""
    asyncio.run(_check_job_freshness())


async def _check_job_freshness():
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import select, update
    from app.core.database import async_session_factory
    from app.models.job import Job, JobStatus

    async with async_session_factory() as db:
        # Mark jobs as expired if not seen in 7 days
        cutoff = datetime.now(timezone.utc) - timedelta(days=7)
        await db.execute(
            update(Job)
            .where(Job.status == JobStatus.ACTIVE, Job.date_seen_last < cutoff)
            .values(status=JobStatus.EXPIRED)
        )
        await db.commit()
        logger.info("Job freshness check completed")


@celery_app.task(name="app.worker.tasks.cleanup_expired_jobs")
def cleanup_expired_jobs():
    """Remove very old expired jobs to keep DB size manageable."""
    asyncio.run(_cleanup_expired_jobs())


async def _cleanup_expired_jobs():
    from datetime import datetime, timezone, timedelta
    from sqlalchemy import delete
    from app.core.database import async_session_factory
    from app.models.job import Job, JobStatus

    async with async_session_factory() as db:
        # Delete jobs expired for more than 90 days
        cutoff = datetime.now(timezone.utc) - timedelta(days=90)
        result = await db.execute(
            delete(Job).where(
                Job.status == JobStatus.EXPIRED,
                Job.updated_at < cutoff,
            )
        )
        await db.commit()
        logger.info(f"Cleaned up {result.rowcount} expired jobs")