divar-crawler/backend/ads/tasks.py

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"🔔 <b>آگهی پرچم‌گذاری شده دیوار</b>\n\n"
        f"📌 <b>عنوان:</b> {title_esc}\n"
        f"💰 <b>قیمت:</b> {price_esc}\n"
        f"🗂 <b>دسته‌بندی:</b> {cat_esc}\n\n"
        f"🤖 <b>علت انتخاب AI:</b>\n{reason_esc}\n\n"
        f"🔗 <a href='{ad.url}'>مشاهده آگهی در دیوار</a>"
    )

    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)
            )