import os import html import requests from openai import OpenAI, AuthenticationError from pydantic import BaseModel, Field from celery import shared_task from django.utils import timezone from ads.models import Ad, AdEvaluation, NotificationLog # Pydantic model for strict OpenAI structured output format class AdEvaluationResult(BaseModel): is_flagged: bool = Field(description="True if the ad matches the criteria in the prompt, otherwise False") 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") @shared_task(bind=True, max_retries=3, default_retry_delay=10) def evaluate_ad_with_ai(self, evaluation_id): """ Asynchronously evaluates a single ad using OpenAI's gpt-4o-mini model against the specific CrawlTask search criteria. """ try: evaluation = AdEvaluation.objects.get(pk=evaluation_id) except AdEvaluation.DoesNotExist: return ad = evaluation.ad task = evaluation.crawl_task api_key = os.getenv('OPENAI_API_KEY') # Fallback if OpenAI API Key is missing if not api_key or 'placeholder' in api_key or 'your_openai_api_key' in api_key or 'your-openai-api-key-here' in api_key: evaluation.is_flagged = False evaluation.reason = "AI Evaluation skipped: OpenAI API key is not configured." evaluation.confidence = 0.0 evaluation.save() return try: client = OpenAI(api_key=api_key) system_instruction = ( "You are an expert Iranian market analyst. Your job is to read listing descriptions " "and decide if they match specific target criteria. You MUST reply using the structured JSON response format " "with Persian strings." ) user_content = ( f"User Search Criteria: {task.detection_prompt}\n\n" f"Ad Title: {ad.title}\n" f"Ad Price: {ad.price or 'Not specified'}\n" f"Ad Category: {ad.category or 'Not specified'}\n" f"Ad Description:\n{ad.description}" ) completion = client.beta.chat.completions.parse( model="gpt-4o-mini", messages=[ {"role": "system", "content": system_instruction}, {"role": "user", "content": user_content} ], response_format=AdEvaluationResult, timeout=25 ) result = completion.choices[0].message.parsed evaluation.is_flagged = result.is_flagged evaluation.reason = result.reason evaluation.confidence = result.confidence evaluation.save() # Update run stats on successful AI flagging if result.is_flagged: 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') + 1 latest_run.save(update_fields=['ads_flagged_count']) # Send telegram channel notification if configured if task.telegram_channel_id: send_telegram_notification.delay(evaluation.id) except AuthenticationError as auth_err: evaluation.is_flagged = False evaluation.reason = "AI Evaluation failed: Invalid or incorrect OpenAI API key." evaluation.confidence = 0.0 evaluation.save() except Exception as e: # Retry in case of API rate limits or network issues try: self.retry(exc=e) except self.MaxRetriesExceededError: evaluation.is_flagged = False evaluation.reason = f"AI Evaluation failed: {str(e)}" evaluation.confidence = 0.0 evaluation.save() @shared_task(bind=True, max_retries=3, default_retry_delay=15) def send_telegram_notification(self, evaluation_id): """ Sends an HTML formatted alert message to the target Telegram Channel notifying them of a flagged ad. """ try: evaluation = AdEvaluation.objects.get(pk=evaluation_id) except AdEvaluation.DoesNotExist: return ad = evaluation.ad task = evaluation.crawl_task token = os.getenv('TELEGRAM_BOT_TOKEN') channel = task.telegram_channel_id if not token or not channel: return # Escape HTML to prevent telegram parsing errors title_esc = html.escape(ad.title) price_esc = html.escape(ad.price or 'مشخص نشده') cat_esc = html.escape(ad.category or 'مشخص نشده') reason_esc = html.escape(evaluation.reason or '') message_html = ( f"🔔 آگهی پرچم‌گذاری شده دیوار\n\n" f"📌 عنوان: {title_esc}\n" f"💰 قیمت: {price_esc}\n" f"🗂 دسته‌بندی: {cat_esc}\n\n" f"🤖 علت انتخاب AI:\n{reason_esc}\n\n" f"🔗 مشاهده آگهی در دیوار" ) url = f"https://api.telegram.org/bot{token}/sendMessage" payload = { 'chat_id': channel, 'text': message_html, 'parse_mode': 'HTML' } try: res = requests.post(url, json=payload, timeout=10) if res.status_code == 200: NotificationLog.objects.create( evaluation=evaluation, channel_id=channel, status='SENT' ) else: raise Exception(f"Telegram API responded with code {res.status_code}: {res.text}") except Exception as e: try: self.retry(exc=e) except self.MaxRetriesExceededError: NotificationLog.objects.create( evaluation=evaluation, channel_id=channel, status='FAILED', error_message=str(e) )