Add celery routing and update procfile

Author: trecouvr

Procfile

@@ -1,4 +1,4 @@
 web: gunicorn --config gunicorn_conf.py docia.wsgi
-worker: celery --app docia worker -l INFO --concurrency=2
+worker: celery --app docia worker -l INFO -Q celery -n celery@%h --concurrency=2
+workerheavycpu: celery --app docia worker -l INFO Q heavy_cpu -n heavy_cpu@%h --concurrency=1
 postdeploy: python manage.py migrate
-

docia/file_processing/analyze.py

@@ -41,7 +41,7 @@ def process(self, job):
         document.save(update_fields=["llm_response", "json_error"])
 
 
-@shared_task
+@shared_task(name="docia.analyze_document")
 def task_analyze_document(job_id: str):
     runner = AnalyzeDocumentJobWorker()
     return runner.run(job_id)

docia/file_processing/classification.py

@@ -40,7 +40,7 @@ def process(self, job):
         document.save(update_fields=["classification", "classification_type"])
 
 
-@shared_task
+@shared_task(name="docia.classify_document")
 def task_classify_document(job_id: str):
     runner = DocumentClassificationJobWorker()
     return runner.run(job_id)

docia/file_processing/text_extraction.py

@@ -35,7 +35,7 @@ def process(self, job):
         document.save(update_fields=["text", "is_ocr", "nb_mot"])
 
 
-@shared_task
+@shared_task(name="docia.extract_text", queue="heavy_cpu")
 def task_extract_text(job_id: str) -> JobStatus:
     worker = ExtractTextJobWorker()
     return worker.run(job_id)

docia/file_processing/utils.py

@@ -50,7 +50,7 @@ def run(self, job_id: str) -> JobStatus:
     def process(self, job): ...
 
 
-@shared_task
+@shared_task(name="docia.finalize_batch")
 def task_finalize_batch(job_results: list[JobStatus], batch_id: str) -> JobStatus:
     """
     Celery task to finalize a batch text extraction process.

docia/tasks.py

@@ -10,7 +10,7 @@
 logger = logging.getLogger(__name__)
 
 
-@shared_task
+@shared_task(name="docia.add")
 def add(x, y):
     logger.info("add %s + %s", x, y)
     return x + y