feat(api): add batch AI evaluation task and stop active crawl run endpoint

backend/ads/tasks.py

@@ -13,6 +13,16 @@ class AdEvaluationResult(BaseModel):
     reason: str = Field(description="A concise explanation in Persian describing why the ad matches or doesn't match the criteria")
     confidence: float = Field(description="Confidence score between 0.0 and 1.0")
 
+class SingleBatchAdResult(BaseModel):
+    evaluation_id: str = Field(description="The exact UUID evaluation_id provided for the ad")
+    is_flagged: bool = Field(description="True if the ad matches the target criteria, otherwise False")
+    reason: str = Field(description="A concise explanation in Persian describing why the ad matches or doesn't match")
+    confidence: float = Field(description="Confidence score between 0.0 and 1.0")
+
+class BatchAdEvaluationResult(BaseModel):
+    results: list[SingleBatchAdResult] = Field(description="List of evaluation results corresponding to each ad in the batch")
+
+
 @shared_task(bind=True, max_retries=3, default_retry_delay=10)
 def evaluate_ad_with_ai(self, evaluation_id):
     """
@@ -106,6 +116,107 @@ def evaluate_ad_with_ai(self, evaluation_id):
             evaluation.confidence = 0.0
             evaluation.save()
 
+@shared_task(bind=True, max_retries=3, default_retry_delay=10)
+def evaluate_ad_batch_with_ai(self, evaluation_ids):
+    """
+    Asynchronously evaluates a batch of ads in a single LLM API call.
+    """
+    if not evaluation_ids:
+        return
+
+    evaluations = list(AdEvaluation.objects.filter(id__in=evaluation_ids).select_related('ad', 'crawl_task'))
+    if not evaluations:
+        return
+
+    api_key = os.getenv('OPENAI_API_KEY')
+    if not api_key or any(k in api_key for k in ['placeholder', 'your_openai_api_key', 'your-openai-api-key-here']):
+        for ev in evaluations:
+            ev.is_flagged = False
+            ev.reason = "AI Evaluation skipped: OpenAI API key is not configured."
+            ev.confidence = 0.0
+            ev.save()
+        return
+
+    task = evaluations[0].crawl_task
+    try:
+        client_kwargs = {"api_key": api_key}
+        if api_key.startswith("sk-or-"):
+            client_kwargs["base_url"] = "https://openrouter.ai/api/v1"
+
+        client = OpenAI(**client_kwargs)
+
+        system_instruction = (
+            "You are an expert Iranian market analyst. Your job is to evaluate a batch of ad listings "
+            "against the target search criteria. You MUST reply with a structured JSON object containing "
+            "evaluations for each ad ID provided, using Persian text for the reasons."
+        )
+
+        ads_text = []
+        for index, ev in enumerate(evaluations, 1):
+            ads_text.append(
+                f"--- Ad #{index} ---\n"
+                f"Evaluation ID: {ev.id}\n"
+                f"Title: {ev.ad.title}\n"
+                f"Price: {ev.ad.price or 'Not specified'}\n"
+                f"Category: {ev.ad.category or 'Not specified'}\n"
+                f"Description: {ev.ad.description[:600]}\n"
+            )
+
+        user_content = (
+            f"Target Search Criteria: {task.detection_prompt}\n\n"
+            "Evaluate each of the following ads:\n"
+            + "\n".join(ads_text)
+        )
+
+        default_model = "openrouter/free" if api_key.startswith("sk-or-") else "gpt-4o-mini"
+        model_name = os.getenv("OPENAI_MODEL", default_model)
+
+        completion = client.beta.chat.completions.parse(
+            model=model_name,
+            messages=[
+                {"role": "system", "content": system_instruction},
+                {"role": "user", "content": user_content}
+            ],
+            response_format=BatchAdEvaluationResult,
+            max_tokens=1500,
+            timeout=45
+        )
+
+        parsed = completion.choices[0].message.parsed
+        res_map = {str(res.evaluation_id): res for res in parsed.results}
+
+        flagged_count = 0
+        for ev in evaluations:
+            res = res_map.get(str(ev.id))
+            if res:
+                ev.is_flagged = res.is_flagged
+                ev.reason = res.reason
+                ev.confidence = res.confidence
+            else:
+                ev.is_flagged = False
+                ev.reason = "Evaluation result omitted in batch response."
+                ev.confidence = 0.0
+            ev.save()
+
+            if ev.is_flagged:
+                flagged_count += 1
+                if task.telegram_channel_id:
+                    send_telegram_notification.delay(ev.id)
+
+        if flagged_count > 0:
+            from crawler.models import CrawlRun
+            from django.db.models import F
+            latest_run = CrawlRun.objects.filter(crawl_task=task).order_by('-started_at').first()
+            if latest_run:
+                latest_run.ads_flagged_count = F('ads_flagged_count') + flagged_count
+                latest_run.save(update_fields=['ads_flagged_count'])
+
+    except Exception as e:
+        # Fallback to individual evaluations if batch call fails
+        for ev in evaluations:
+            evaluate_ad_with_ai.delay(ev.id)
+
+
 @shared_task(bind=True, max_retries=3, default_retry_delay=15)
 def send_telegram_notification(self, evaluation_id):
     """

backend/crawler/tasks.py

@@ -161,9 +161,14 @@ def run_crawl_pipeline(run_id, force=False):
 
     # 4. Save listings & trigger AI evaluations
     ads_fetched = len(found_ads)
-    ads_evaluated = 0
+    new_eval_ids = []
     
     for token, ad_info in found_ads.items():
+        # Check if run was cancelled/stopped by user during loop
+        run.refresh_from_db()
+        if run.status != 'RUNNING':
+            return
+
         try:
             # Create or get Ad metadata
             ad, created = Ad.objects.get_or_create(
@@ -182,18 +187,25 @@ def run_crawl_pipeline(run_id, force=False):
             eval_exists = AdEvaluation.objects.filter(ad=ad, crawl_task=task).exists()
             if not eval_exists:
                 eval_obj = AdEvaluation.objects.create(ad=ad, crawl_task=task)
-                ads_evaluated += 1
-                
-                # Trigger Celery AI evaluation task
-                from ads.tasks import evaluate_ad_with_ai
-                evaluate_ad_with_ai.delay(eval_obj.id)
+                new_eval_ids.append(str(eval_obj.id))
         except Exception:
             # If a single ad save fails, continue processing other ads
             pass
 
+    # Batch process AI evaluations (5 ads per API request)
+    if new_eval_ids:
+        from ads.tasks import evaluate_ad_batch_with_ai
+        batch_size = 5
+        for i in range(0, len(new_eval_ids), batch_size):
+            batch = new_eval_ids[i:i + batch_size]
+            evaluate_ad_batch_with_ai.delay(batch)
+
     # Update runs counters and status
-    run.status = 'SUCCESS'
-    run.finished_at = timezone.now()
-    run.ads_fetched_count = ads_fetched
-    run.ads_evaluated_count = ads_evaluated
-    run.save()
+    run.refresh_from_db()
+    if run.status == 'RUNNING':
+        run.status = 'SUCCESS'
+        run.finished_at = timezone.now()
+        run.ads_fetched_count = ads_fetched
+        run.ads_evaluated_count = len(new_eval_ids)
+        run.save()
+

backend/crawler/tests.py

@@ -138,6 +138,19 @@ class CrawlTaskAPITests(APITestCase):
         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
         self.assertFalse(CrawlRun.objects.filter(id=run.id).exists())
 
+    def test_stop_crawl_run(self):
+        run = CrawlRun.objects.create(
+            crawl_task=self.task,
+            status='RUNNING'
+        )
+        stop_url = f"/api/crawlers/runs/{run.id}/stop/"
+        response = self.client.post(stop_url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        run.refresh_from_db()
+        self.assertEqual(run.status, 'FAILED')
+        self.assertIn("متوقف شد", run.error_log)
+
+
 
 class CrawlTaskCeleryTests(APITestCase):
     def setUp(self):

backend/crawler/views.py

@@ -49,10 +49,24 @@ class CrawlTaskViewSet(viewsets.ModelViewSet):
         return Response(serializer.data)
 
 
+from django.utils import timezone
+
 class CrawlRunViewSet(viewsets.ModelViewSet):
     """
-    ViewSet for CrawlRun models. Supports CRUD (including DELETE).
+    ViewSet for CrawlRun models. Supports CRUD (including DELETE) and stopping active runs.
     """
     queryset = CrawlRun.objects.all().order_by('-started_at')
     serializer_class = CrawlRunSerializer
 
+    @action(detail=True, methods=['post'], url_path='stop')
+    def stop_run(self, request, pk=None):
+        run = self.get_object()
+        if run.status == 'RUNNING':
+            run.status = 'FAILED'
+            run.finished_at = timezone.now()
+            run.error_log = "اجرا توسط کاربر متوقف شد."
+            run.save()
+            return Response({"status": "stopped", "run_id": str(run.id)}, status=status.HTTP_200_OK)
+        return Response({"status": "not_running", "run_id": str(run.id)}, status=status.HTTP_400_BAD_REQUEST)
+
+