From b47bfe6bb47b5a00d0c6cf59c9f654c7ba732550 Mon Sep 17 00:00:00 2001 From: PouyaKhajavi Date: Sun, 9 Aug 2026 11:29:11 +0330 Subject: [PATCH] feat(api): add batch AI evaluation task and stop active crawl run endpoint --- backend/ads/tasks.py | 111 +++++++++++++++++++++++++++++++++++++++ backend/crawler/tasks.py | 34 ++++++++---- backend/crawler/tests.py | 13 +++++ backend/crawler/views.py | 16 +++++- 4 files changed, 162 insertions(+), 12 deletions(-) diff --git a/backend/ads/tasks.py b/backend/ads/tasks.py index 9d04874..aa16f9e 100644 --- a/backend/ads/tasks.py +++ b/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): """ diff --git a/backend/crawler/tasks.py b/backend/crawler/tasks.py index bfe6c2e..f3de09f 100644 --- a/backend/crawler/tasks.py +++ b/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() + diff --git a/backend/crawler/tests.py b/backend/crawler/tests.py index 1bf5666..4697ed2 100644 --- a/backend/crawler/tests.py +++ b/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): diff --git a/backend/crawler/views.py b/backend/crawler/views.py index 86608a8..0362cfe 100644 --- a/backend/crawler/views.py +++ b/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) + +