import os
import re
import json
import html
import logging
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
logger = logging.getLogger(__name__)
# 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")
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")
def extract_and_parse_json(text: str):
"""
Cleans raw LLM text (removing markdown code fences, preambles, thinking process)
and extracts a valid JSON object or list.
"""
if not text:
raise ValueError("Empty response from LLM")
text = text.strip()
# 1. Check for markdown code fences
fence_match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text, re.IGNORECASE)
if fence_match:
text = fence_match.group(1).strip()
# 2. Try direct json.loads
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# 3. Find outermost '{' ... '}' or '[' ... ']'
start_dict = text.find('{')
end_dict = text.rfind('}')
start_list = text.find('[')
end_list = text.rfind(']')
if start_dict != -1 and end_dict != -1 and end_dict > start_dict:
try:
return json.loads(text[start_dict:end_dict + 1])
except json.JSONDecodeError:
pass
if start_list != -1 and end_list != -1 and end_list > start_list:
try:
return json.loads(text[start_list:end_list + 1])
except json.JSONDecodeError:
pass
raise ValueError(f"Could not extract valid JSON from LLM response: {text[:200]}")
def call_llm_with_structured_fallback(client, model_name, system_instruction, user_content, pydantic_cls, max_tokens, timeout):
"""
Attempts to call LLM using OpenAI beta parse first, then falls back to standard completions
with json_object response format and robust markdown-stripping JSON parsing.
"""
# 1. First attempt: beta structured parse (ideal for official OpenAI models / compatible endpoints)
try:
completion = client.beta.chat.completions.parse(
model=model_name,
messages=[
{"role": "system", "content": system_instruction},
{"role": "user", "content": user_content}
],
response_format=pydantic_cls,
max_tokens=max_tokens,
timeout=timeout
)
if completion.choices and completion.choices[0].message.parsed:
return completion.choices[0].message.parsed
except Exception:
# Fallback to chat completions if beta parse fails (e.g. OpenRouter returning markdown wrappers or non-strict JSON)
pass
# 2. Second attempt: standard chat completions with json_object response format
system_instruction_json = (
f"{system_instruction}\n"
"CRITICAL REQUIREMENT: You MUST reply strictly with a valid JSON object. "
"Do NOT wrap the JSON in markdown code blocks like ```json ... ```. "
"Do NOT include any introduction, thinking process, or explanatory text before or after the JSON."
)
try:
completion = client.chat.completions.create(
model=model_name,
messages=[
{"role": "system", "content": system_instruction_json},
{"role": "user", "content": user_content}
],
response_format={"type": "json_object"},
max_tokens=max_tokens,
timeout=timeout
)
raw_text = completion.choices[0].message.content or ""
json_data = extract_and_parse_json(raw_text)
return pydantic_cls.model_validate(json_data)
except Exception:
pass
# 3. Third attempt: standard chat completions without response_format constraint
completion = client.chat.completions.create(
model=model_name,
messages=[
{"role": "system", "content": system_instruction_json},
{"role": "user", "content": user_content}
],
max_tokens=max_tokens,
timeout=timeout
)
raw_text = completion.choices[0].message.content or ""
json_data = extract_and_parse_json(raw_text)
return pydantic_cls.model_validate(json_data)
@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 / OpenRouter 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 any(k in api_key for k in ['placeholder', 'your_openai_api_key', 'your-openai-api-key-here']):
evaluation.is_flagged = False
evaluation.reason = "AI Evaluation skipped: OpenAI API key is not configured."
evaluation.confidence = 0.0
evaluation.save()
return
try:
# Support OpenRouter keys seamlessly
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 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}"
)
default_model = "openrouter/free" if api_key.startswith("sk-or-") else "gpt-4o-mini"
model_name = os.getenv("OPENAI_MODEL", default_model)
result = call_llm_with_structured_fallback(
client=client,
model_name=model_name,
system_instruction=system_instruction,
user_content=user_content,
pydantic_cls=AdEvaluationResult,
max_tokens=2500,
timeout=35
)
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=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:\n{ev.ad.description[:2500]}\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)
parsed = call_llm_with_structured_fallback(
client=client,
model_name=model_name,
system_instruction=system_instruction,
user_content=user_content,
pydantic_cls=BatchAdEvaluationResult,
max_tokens=8000,
timeout=90
)
res_map = {}
for idx, res in enumerate(parsed.results):
eval_id_str = str(res.evaluation_id).strip()
res_map[eval_id_str] = res
res_map[str(idx)] = res
res_map[f"Ad #{idx + 1}"] = res
res_map[f"#{idx + 1}"] = res
flagged_count = 0
for idx, ev in enumerate(evaluations):
ev_id_str = str(ev.id).strip()
res = res_map.get(ev_id_str)
if not res and idx < len(parsed.results):
res = parsed.results[idx]
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):
"""
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)
)