PouyaKhajavi
/
divar-crawler
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

feat(backend): implement divar scraper pipeline and openai/telegram tasks

master
PouyaKhajavi 1 day ago
parent
87c99dfe71
commit
4626482238
6 changed files with 399 additions and 2 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 20
      backend/ads/migrations/0002_alter_adevaluation_crawl_task.py
  2. 2
      backend/ads/models.py
  3. 160
      backend/ads/tasks.py
  4. 19
      backend/crawler/migrations/0002_alter_crawlrun_crawl_task.py
  5. 2
      backend/crawler/models.py
  6. 198
      backend/crawler/tasks.py

20
backend/ads/migrations/0002_alter_adevaluation_crawl_task.py
View File

@ -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'),
),
]

2
backend/ads/models.py
View File

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

160
backend/ads/tasks.py
View File

@ -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"🔔 <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)
)

19
backend/crawler/migrations/0002_alter_crawlrun_crawl_task.py
View File

@ -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'),
),
]

2
backend/crawler/models.py
View File

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

198
backend/crawler/tasks.py
View File

@ -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()
Write Preview
Loading…
Cancel
Save