diff --git a/backend/ads/migrations/0002_alter_adevaluation_crawl_task.py b/backend/ads/migrations/0002_alter_adevaluation_crawl_task.py new file mode 100644 index 0000000..ebd327d --- /dev/null +++ b/backend/ads/migrations/0002_alter_adevaluation_crawl_task.py @@ -0,0 +1,20 @@ +# Generated by Django 6.0.8 on 2026-08-08 10:04 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('ads', '0001_initial'), + ('crawler', '0002_alter_crawlrun_crawl_task'), + ] + + operations = [ + migrations.AlterField( + model_name='adevaluation', + name='crawl_task', + field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='evaluations', to='crawler.crawltask'), + ), + ] diff --git a/backend/ads/models.py b/backend/ads/models.py index 78240df..9670670 100644 --- a/backend/ads/models.py +++ b/backend/ads/models.py @@ -23,7 +23,7 @@ class Ad(BaseModel): class AdEvaluation(BaseModel): - crawl_task = models.ForeignKey(CrawlTask, on_delete=models.PROTECT, related_name='evaluations') + crawl_task = models.ForeignKey(CrawlTask, on_delete=models.CASCADE, related_name='evaluations') ad = models.ForeignKey(Ad, on_delete=models.PROTECT, related_name='evaluations') is_flagged = models.BooleanField(default=False) reason = models.TextField(null=True, blank=True) diff --git a/backend/ads/tasks.py b/backend/ads/tasks.py new file mode 100644 index 0000000..b387836 --- /dev/null +++ b/backend/ads/tasks.py @@ -0,0 +1,160 @@ +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) + ) diff --git a/backend/crawler/migrations/0002_alter_crawlrun_crawl_task.py b/backend/crawler/migrations/0002_alter_crawlrun_crawl_task.py new file mode 100644 index 0000000..b73db9e --- /dev/null +++ b/backend/crawler/migrations/0002_alter_crawlrun_crawl_task.py @@ -0,0 +1,19 @@ +# Generated by Django 6.0.8 on 2026-08-08 10:04 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('crawler', '0001_initial'), + ] + + operations = [ + migrations.AlterField( + model_name='crawlrun', + name='crawl_task', + field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='runs', to='crawler.crawltask'), + ), + ] diff --git a/backend/crawler/models.py b/backend/crawler/models.py index c2fdfda..0b4c33d 100644 --- a/backend/crawler/models.py +++ b/backend/crawler/models.py @@ -48,7 +48,7 @@ class CrawlRun(BaseModel): ('FAILED', 'Failed'), ] - crawl_task = models.ForeignKey(CrawlTask, on_delete=models.PROTECT, related_name='runs') + crawl_task = models.ForeignKey(CrawlTask, on_delete=models.CASCADE, related_name='runs') started_at = models.DateTimeField(auto_now_add=True) finished_at = models.DateTimeField(null=True, blank=True) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='RUNNING') diff --git a/backend/crawler/tasks.py b/backend/crawler/tasks.py new file mode 100644 index 0000000..10c57b2 --- /dev/null +++ b/backend/crawler/tasks.py @@ -0,0 +1,198 @@ +import re +import json +import requests +import urllib.parse +from urllib.parse import urlparse +from django.utils import timezone +from celery import shared_task +from crawler.models import CrawlRun, CrawlTask +from ads.models import Ad, AdEvaluation + +def parse_divar_url(divar_url): + """ + Parses a public Divar search link to extract city, category path, + and query parameters. + """ + parsed = urlparse(divar_url) + path_parts = [p for p in parsed.path.split('/') if p] + + city = 'tehran' + category = '' + + if len(path_parts) >= 2 and path_parts[0] == 's': + city = path_parts[1] + if len(path_parts) > 2: + category = '/'.join(path_parts[2:]) + elif len(path_parts) >= 1: + city = path_parts[0] + if len(path_parts) > 1: + category = '/'.join(path_parts[1:]) + + query_params = dict(urllib.parse.parse_qsl(parsed.query)) + return city, category, query_params + +def extract_widgets_from_dict(d, found_ads): + """ + Recursively searches a JSON structure for dictionaries matching + the properties of an ad widget, extracting token, title, description, + price, and image details. + """ + if isinstance(d, dict): + # Retrieve token + token = d.get('token') + if not token: + token = d.get('action', {}).get('payload', {}).get('token') + if not token: + token = d.get('data', {}).get('token') + + # Retrieve title + title = d.get('title') or d.get('data', {}).get('title') + + # Verify it represents an ad widget (token length >= 6 and has a title) + if token and title and isinstance(token, str) and len(token) >= 6: + desc = ( + d.get('description') or + d.get('data', {}).get('description') or + d.get('data', {}).get('top_description_text') or + d.get('data', {}).get('middle_description_text') or + d.get('data', {}).get('bottom_description_text') or + '' + ) + + price = ( + d.get('price') or + d.get('data', {}).get('price_text') or + d.get('data', {}).get('price') or + '' + ) + + image_url = '' + image_data = d.get('image') or d.get('data', {}).get('image') or d.get('data', {}).get('image_url') + if isinstance(image_data, str): + image_url = image_data + elif isinstance(image_data, dict): + image_url = image_data.get('url') or image_data.get('src') or '' + + ad_info = { + 'token': token, + 'title': title, + 'description': desc, + 'price': price, + 'image_url': image_url, + 'category': d.get('category') or d.get('data', {}).get('category') or '' + } + found_ads[token] = ad_info + + for v in d.values(): + extract_widgets_from_dict(v, found_ads) + + elif isinstance(d, list): + for item in d: + extract_widgets_from_dict(item, found_ads) + +@shared_task +def run_crawl_pipeline(run_id): + """ + Background Celery task that executes a crawling run. + Parses Divar listing data, registers new Ads, and queues AI evaluations. + """ + try: + run = CrawlRun.objects.get(pk=run_id) + except CrawlRun.DoesNotExist: + return + + task = run.crawl_task + + # 1. Verification checks + if not task.is_active: + run.status = 'FAILED' + run.finished_at = timezone.now() + run.error_log = "CrawlTask is inactive." + run.save() + return + + now = timezone.localtime(timezone.now()).time() + if not (task.start_hour <= now <= task.end_hour): + run.status = 'SUCCESS' + run.finished_at = timezone.now() + run.error_log = f"Skipped: current time {now.strftime('%H:%M')} is outside allowed window {task.start_hour.strftime('%H:%M')} to {task.end_hour.strftime('%H:%M')}" + run.save() + return + + # Update run status to RUNNING + run.status = 'RUNNING' + run.save() + + headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36' + } + + city, category, query_params = parse_divar_url(task.divar_url) + found_ads = {} + + # 2. Try fetching from Divar public web search API + api_url = f"https://api.divar.ir/v8/web-search/{city}/{category}" + try: + res = requests.get(api_url, params=query_params, headers=headers, timeout=15) + if res.status_code == 200: + extract_widgets_from_dict(res.json(), found_ads) + except Exception as api_err: + # Log error but proceed to HTML scraper fallback + pass + + # 3. Fallback: Parse HTML state if API failed/returned nothing + if not found_ads: + try: + res = requests.get(task.divar_url, headers=headers, timeout=15) + if res.status_code == 200: + match = re.search(r'window\.__PRELOADED_STATE__\s*=\s*(\{[\s\S]*?\});', res.text) + if not match: + match = re.search(r'window\.__INITIAL_STATE__\s*=\s*(\{[\s\S]*?\});', res.text) + if match: + state_data = json.loads(match.group(1)) + extract_widgets_from_dict(state_data, found_ads) + except Exception as html_err: + run.status = 'FAILED' + run.finished_at = timezone.now() + run.error_log = f"Scraping failed. API and HTML fallbacks both failed.\nAPI Error: {str(api_err) if 'api_err' in locals() else 'None'}\nHTML Error: {str(html_err)}" + run.save() + return + + # 4. Save listings & trigger AI evaluations + ads_fetched = len(found_ads) + ads_evaluated = 0 + + for token, ad_info in found_ads.items(): + try: + # Create or get Ad metadata + ad, created = Ad.objects.get_or_create( + divar_token=token, + defaults={ + 'title': ad_info['title'], + 'description': ad_info['description'], + 'price': ad_info['price'], + 'category': ad_info['category'], + 'images': [ad_info['image_url']] if ad_info['image_url'] else [], + 'url': f"https://divar.ir/v/{token}" + } + ) + + # Check if an evaluation already exists for this crawler on this ad + 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) + except Exception: + # If a single ad save fails, continue processing other ads + pass + + # 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()