Spaces:

sitcod3
/

modul-x

Running

modul-x / app /tasks /celery_tasks.py

Upload folder using huggingface_hub

978c21c verified 11 days ago

7.13 kB

from celery import Celery
from ..config import settings
import asyncio

celery_app = Celery(
    "emailvalidator",
    broker=settings.celery_broker_url,
    backend=settings.celery_result_backend,
    include=["app.tasks.celery_tasks"]
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    worker_concurrency=10,
    task_soft_time_limit=300,
    task_time_limit=600,
    # Fail fast when Redis/broker is down instead of retrying forever
    broker_connection_max_retries=2,
    broker_connection_retry_on_startup=False,
    task_publish_retry=False,
    task_publish_retry_policy={"max_retries": 1},
    result_backend_transport_options={
        "retry_policy": {"max_retries": 1, "interval_start": 0, "interval_step": 1},
    },
    # Improve worker acknowledgment and task reliability
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_reject_on_worker_lost=True,
    # Drain the greylist retry queue periodically. Without a beat schedule the
    # queue would only ever fill — a slow leak of unretried addresses.
    beat_schedule={
        "drain-greylist-retries": {
            "task": "app.tasks.celery_tasks.drain_retries_task",
            "schedule": 300.0,  # every 5 minutes
        }
    },
)


@celery_app.task(bind=True, max_retries=1, rate_limit="12/m")
def drain_retries_task(self, limit: int = 50):
    """Re-probe addresses that were greylisted earlier.

    Scheduled by beat every 5 minutes. Without this the retry queue would only
    fill, leaving valid mailboxes on greylisting domains stranded as "risky".
    """
    from ..validator.engine import EmailValidationEngine
    from ..validator.retry_queue import drain_retry_queue, retry_queue
    import sys

    async def run():
        eng = EmailValidationEngine()

        async def prober(email: str, mx_host: str):
            return await eng.mailbox.verify(email, mx_host)

        stats = await drain_retry_queue(prober, queue=retry_queue, limit=limit)

        # Cache the resolved verdicts so History/Results and future runs see the
        # retry's outcome (a valid mailbox stops being reported as risky).
        cached = 0
        for email, result in stats.get("results", []):
            if result.verified is None:
                continue
            try:
                await eng.cache.put_verdict(email, True, result, None)
                cached += 1
            except Exception as exc:
                print(f"[RETRY] could not cache {email}: {exc}", file=sys.stderr)

        return {
            "checked": stats["checked"],
            "resolved": stats["resolved"],
            "requeued": stats["requeued"],
            "errors": stats["errors"],
            "cached": cached,
        }

    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            return loop.run_until_complete(run())
        finally:
            loop.close()
    except Exception as exc:
        print(f"[RETRY] drain failed: {exc}", file=sys.stderr)
        return {"error": str(exc)}


@celery_app.task(bind=True, max_retries=3, rate_limit="100/m")
def bulk_validate_task(self, emails: list, webhook_url: str = None, deep: bool = True):
    """
    Celery task for async bulk email validation.
    Runs synchronously inside Celery worker.

    `deep` defaults to True: bulk must run the full 7-layer pipeline (catch-all
    + SMTP) to produce a real verdict, matching single verification.
    """
    from ..validator.engine import EmailValidationEngine
    from ..db import repo
    import sys

    CHUNK = 200

    async def run():
        eng = EmailValidationEngine()
        results = []
        total = len(emails)

        # Process in chunks through the engine's batched path: DNS/MX once per
        # domain and one mailbox call per chunk, so the SMTP engine reuses a
        # session per receiving host instead of a TCP connection per address.
        for start in range(0, total, CHUNK):
            chunk = emails[start:start + CHUNK]
            try:
                chunk_results = await eng.validate_batch(chunk, deep=deep)
                results.extend(r.model_dump() for r in chunk_results)
            except Exception as e:
                print(
                    f"[CELERY TASK] Validation failed for chunk "
                    f"{start + 1}-{start + len(chunk)}/{total}: {e.__class__.__name__}",
                    file=sys.stderr,
                )
                results.extend(
                    {"email": email, "error": str(e), "status": "error"}
                    for email in chunk
                )

            done = min(start + len(chunk), total)
            self.update_state(state="PROGRESS", meta={"current": done, "total": total})

            # Persist incremental progress to Supabase (best-effort)
            try:
                repo.update_bulk_job_progress(self.request.id, done, total)
            except Exception as e:
                print(f"[CELERY TASK] Failed to update progress: {str(e)}", file=sys.stderr)

        # Run-quality monitor (report item 10): a run that is mostly
        # risky/unknown is a systemic failure — page via the signed webhook.
        try:
            from ..utils.rate_alerts import maybe_alert_risky_rate

            await maybe_alert_risky_rate(
                results,
                webhook_url=webhook_url,
                job_id=self.request.id,
                threshold=settings.risky_rate_alert_threshold,
            )
        except Exception as e:
            print(f"[CELERY TASK] Rate alert check failed: {str(e)}", file=sys.stderr)

        return results

    try:
        print(f"[CELERY TASK] Starting bulk validation for {len(emails)} emails, task_id: {self.request.id}")
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        results = loop.run_until_complete(run())
        loop.close()

        print(f"[CELERY TASK] Completed bulk validation for {len(emails)} emails, task_id: {self.request.id}")

        # Persist finished job (best-effort)
        try:
            repo.finish_bulk_job(self.request.id, results)
        except Exception as e:
            print(f"[CELERY TASK] Failed to finish job: {str(e)}", file=sys.stderr)

        # Webhook notification
        if webhook_url:
            import httpx
            try:
                httpx.post(webhook_url, json={"results": results}, timeout=10)
                print(f"[CELERY TASK] Webhook notification sent to {webhook_url}")
            except Exception as e:
                print(f"[CELERY TASK] Webhook notification failed: {str(e)}", file=sys.stderr)

        return results
    except Exception as exc:
        print(f"[CELERY TASK] Fatal error in bulk validation: {str(exc)}", file=sys.stderr)
        raise self.retry(exc=exc, countdown=5)

	from celery import Celery
	from ..config import settings
	import asyncio

	celery_app = Celery(
	"emailvalidator",
	broker=settings.celery_broker_url,
	backend=settings.celery_result_backend,
	include=["app.tasks.celery_tasks"]
	)

	celery_app.conf.update(
	task_serializer="json",
	accept_content=["json"],
	result_serializer="json",
	timezone="UTC",
	enable_utc=True,
	task_track_started=True,
	worker_concurrency=10,
	task_soft_time_limit=300,
	task_time_limit=600,
	# Fail fast when Redis/broker is down instead of retrying forever
	broker_connection_max_retries=2,
	broker_connection_retry_on_startup=False,
	task_publish_retry=False,
	task_publish_retry_policy={"max_retries": 1},
	result_backend_transport_options={
	"retry_policy": {"max_retries": 1, "interval_start": 0, "interval_step": 1},
	},
	# Improve worker acknowledgment and task reliability
	task_acks_late=True,
	worker_prefetch_multiplier=1,
	task_reject_on_worker_lost=True,
	# Drain the greylist retry queue periodically. Without a beat schedule the
	# queue would only ever fill — a slow leak of unretried addresses.
	beat_schedule={
	"drain-greylist-retries": {
	"task": "app.tasks.celery_tasks.drain_retries_task",
	"schedule": 300.0, # every 5 minutes
	}
	},
	)


	@celery_app.task(bind=True, max_retries=1, rate_limit="12/m")
	def drain_retries_task(self, limit: int = 50):
	"""Re-probe addresses that were greylisted earlier.

	Scheduled by beat every 5 minutes. Without this the retry queue would only
	fill, leaving valid mailboxes on greylisting domains stranded as "risky".
	"""
	from ..validator.engine import EmailValidationEngine
	from ..validator.retry_queue import drain_retry_queue, retry_queue
	import sys

	async def run():
	eng = EmailValidationEngine()

	async def prober(email: str, mx_host: str):
	return await eng.mailbox.verify(email, mx_host)

	stats = await drain_retry_queue(prober, queue=retry_queue, limit=limit)

	# Cache the resolved verdicts so History/Results and future runs see the
	# retry's outcome (a valid mailbox stops being reported as risky).
	cached = 0
	for email, result in stats.get("results", []):
	if result.verified is None:
	continue
	try:
	await eng.cache.put_verdict(email, True, result, None)
	cached += 1
	except Exception as exc:
	print(f"[RETRY] could not cache {email}: {exc}", file=sys.stderr)

	return {
	"checked": stats["checked"],
	"resolved": stats["resolved"],
	"requeued": stats["requeued"],
	"errors": stats["errors"],
	"cached": cached,
	}

	try:
	loop = asyncio.new_event_loop()
	asyncio.set_event_loop(loop)
	try:
	return loop.run_until_complete(run())
	finally:
	loop.close()
	except Exception as exc:
	print(f"[RETRY] drain failed: {exc}", file=sys.stderr)
	return {"error": str(exc)}


	@celery_app.task(bind=True, max_retries=3, rate_limit="100/m")
	def bulk_validate_task(self, emails: list, webhook_url: str = None, deep: bool = True):
	"""
	Celery task for async bulk email validation.
	Runs synchronously inside Celery worker.

	`deep` defaults to True: bulk must run the full 7-layer pipeline (catch-all
	+ SMTP) to produce a real verdict, matching single verification.
	"""
	from ..validator.engine import EmailValidationEngine
	from ..db import repo
	import sys

	CHUNK = 200

	async def run():
	eng = EmailValidationEngine()
	results = []
	total = len(emails)

	# Process in chunks through the engine's batched path: DNS/MX once per
	# domain and one mailbox call per chunk, so the SMTP engine reuses a
	# session per receiving host instead of a TCP connection per address.
	for start in range(0, total, CHUNK):
	chunk = emails[start:start + CHUNK]
	try:
	chunk_results = await eng.validate_batch(chunk, deep=deep)
	results.extend(r.model_dump() for r in chunk_results)
	except Exception as e:
	print(
	f"[CELERY TASK] Validation failed for chunk "
	f"{start + 1}-{start + len(chunk)}/{total}: {e.__class__.__name__}",
	file=sys.stderr,
	)
	results.extend(
	{"email": email, "error": str(e), "status": "error"}
	for email in chunk
	)

	done = min(start + len(chunk), total)
	self.update_state(state="PROGRESS", meta={"current": done, "total": total})

	# Persist incremental progress to Supabase (best-effort)
	try:
	repo.update_bulk_job_progress(self.request.id, done, total)
	except Exception as e:
	print(f"[CELERY TASK] Failed to update progress: {str(e)}", file=sys.stderr)

	# Run-quality monitor (report item 10): a run that is mostly
	# risky/unknown is a systemic failure — page via the signed webhook.
	try:
	from ..utils.rate_alerts import maybe_alert_risky_rate

	await maybe_alert_risky_rate(
	results,
	webhook_url=webhook_url,
	job_id=self.request.id,
	threshold=settings.risky_rate_alert_threshold,
	)
	except Exception as e:
	print(f"[CELERY TASK] Rate alert check failed: {str(e)}", file=sys.stderr)

	return results

	try:
	print(f"[CELERY TASK] Starting bulk validation for {len(emails)} emails, task_id: {self.request.id}")
	loop = asyncio.new_event_loop()
	asyncio.set_event_loop(loop)
	results = loop.run_until_complete(run())
	loop.close()

	print(f"[CELERY TASK] Completed bulk validation for {len(emails)} emails, task_id: {self.request.id}")

	# Persist finished job (best-effort)
	try:
	repo.finish_bulk_job(self.request.id, results)
	except Exception as e:
	print(f"[CELERY TASK] Failed to finish job: {str(e)}", file=sys.stderr)

	# Webhook notification
	if webhook_url:
	import httpx
	try:
	httpx.post(webhook_url, json={"results": results}, timeout=10)
	print(f"[CELERY TASK] Webhook notification sent to {webhook_url}")
	except Exception as e:
	print(f"[CELERY TASK] Webhook notification failed: {str(e)}", file=sys.stderr)

	return results
	except Exception as exc:
	print(f"[CELERY TASK] Fatal error in bulk validation: {str(exc)}", file=sys.stderr)
	raise self.retry(exc=exc, countdown=5)