134 lines
6.8 KiB
Python
134 lines
6.8 KiB
Python
import re
|
|
import json
|
|
import logging
|
|
import httpx
|
|
from typing import Optional, Dict, Any, Tuple
|
|
from core.llm import LLMClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
|
|
WEBSITE_ANALYSIS_SYSTEM_PROMPT = """
|
|
You are an expert web scraping and REST API discovery agent.
|
|
Your job is to analyze website HTML, RSS feeds, or API responses, and configure the optimal extraction schema for pulling the latest articles or news posts.
|
|
|
|
You MUST respond with valid JSON matching this schema:
|
|
{
|
|
"status": "success",
|
|
"endpoint_url": "URL to fetch latest items from (e.g. RSS feed, WordPress JSON endpoint, or HTML page)",
|
|
"parser_type": "rss" | "wordpress_json" | "rest_json" | "html",
|
|
"http_method": "GET",
|
|
"headers": {
|
|
"User-Agent": "Mozilla/5.0 ..."
|
|
},
|
|
"items_path": "JSON key containing the list of items (e.g. 'items', 'articles', 'data') or empty if root is a list or RSS channel.item",
|
|
"field_mappings": {
|
|
"title": "field or selector for article title",
|
|
"content": "field or selector for summary or body text",
|
|
"link": "field or selector for full article URL",
|
|
"image_url": "field or selector for featured image URL (or empty if none)",
|
|
"published_at": "field or selector for date/time (or empty)"
|
|
},
|
|
"summary": "Brief Persian explanation of the discovered API/source (e.g. 'از طریق API رسمی وردپرس شناسایی شد')."
|
|
}
|
|
"""
|
|
|
|
class WebsiteAnalyzer:
|
|
def __init__(self, llm: LLMClient):
|
|
self.llm = llm
|
|
|
|
async def analyze_website(self, url: str, custom_instructions: str = "") -> Tuple[bool, Dict[str, Any], str]:
|
|
"""Discover endpoints and prompt AI to build the extractor schema."""
|
|
clean_url = url.strip()
|
|
if not clean_url.startswith("http://") and not clean_url.startswith("https://"):
|
|
clean_url = "https://" + clean_url
|
|
|
|
discovered_info = []
|
|
sample_data = ""
|
|
|
|
# 1. Fetch homepage and look for RSS/JSON endpoints
|
|
headers = {"User-Agent": USER_AGENT, "Accept": "*/*"}
|
|
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
|
|
try:
|
|
resp = await client.get(clean_url, headers=headers)
|
|
resp.raise_for_status()
|
|
html_text = resp.text
|
|
|
|
# Check RSS / Atom links in HTML
|
|
rss_links = re.findall(r'<link[^>]+type=["\']application/(?:rss|atom)\+xml["\'][^>]+href=["\']([^"\']+)["\']', html_text, re.I)
|
|
if not rss_links:
|
|
rss_links = re.findall(r'<link[^>]+href=["\']([^"\']+)["\'][^>]+type=["\']application/(?:rss|atom)\+xml["\']', html_text, re.I)
|
|
|
|
# Check for WordPress REST API link
|
|
wp_api = re.findall(r'<link[^>]+rel=["\']https://api\.w\.org/["\'][^>]+href=["\']([^"\']+)["\']', html_text, re.I)
|
|
|
|
if wp_api:
|
|
discovered_info.append(f"Discovered WordPress REST API base: {wp_api[0]}")
|
|
try:
|
|
wp_posts_url = wp_api[0].rstrip('/') + '/wp/v2/posts?per_page=5'
|
|
wp_res = await client.get(wp_posts_url, headers=headers)
|
|
if wp_res.status_code == 200:
|
|
discovered_info.append(f"WordPress Posts API returned 200 OK: {wp_posts_url}")
|
|
sample_data = f"Sample from WordPress API ({wp_posts_url}):\n" + wp_res.text[:2500]
|
|
except Exception as e:
|
|
logger.debug(f"WP test fetch error: {e}")
|
|
|
|
if not sample_data and rss_links:
|
|
feed_url = rss_links[0]
|
|
if not feed_url.startswith("http"):
|
|
from urllib.parse import urljoin
|
|
feed_url = urljoin(clean_url, feed_url)
|
|
discovered_info.append(f"Discovered RSS feed link: {feed_url}")
|
|
try:
|
|
feed_res = await client.get(feed_url, headers=headers)
|
|
if feed_res.status_code == 200:
|
|
sample_data = f"Sample from RSS Feed ({feed_url}):\n" + feed_res.text[:2500]
|
|
except Exception as e:
|
|
logger.debug(f"RSS test fetch error: {e}")
|
|
|
|
# If no direct feed found, test common paths: /feed, /rss
|
|
if not sample_data:
|
|
for common_path in ["/feed", "/rss", "/feed.xml", "/rss.xml", "/wp-json/wp/v2/posts?per_page=5"]:
|
|
try:
|
|
from urllib.parse import urljoin
|
|
test_url = urljoin(clean_url, common_path)
|
|
test_res = await client.get(test_url, headers=headers)
|
|
if test_res.status_code == 200 and ("xml" in test_res.headers.get("content-type", "") or "json" in test_res.headers.get("content-type", "")):
|
|
discovered_info.append(f"Found working feed at: {test_url}")
|
|
sample_data = f"Sample from {test_url}:\n" + test_res.text[:2500]
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
# If still no feed, provide HTML snippet
|
|
if not sample_data:
|
|
discovered_info.append("No standard RSS/API feed found; analyzing HTML structure.")
|
|
sample_data = f"HTML head/body sample from {clean_url}:\n" + html_text[:3000]
|
|
|
|
except Exception as e:
|
|
return False, {}, f"خطا در برقراری ارتباط با وبسایت: {e}"
|
|
|
|
custom_block = f"\nUser Extraction Needs / Filtering Instructions:\n{custom_instructions.strip()}\n" if custom_instructions and custom_instructions.strip() else ""
|
|
|
|
prompt = (
|
|
f"Website Target URL: {clean_url}\n"
|
|
f"{custom_block}"
|
|
f"Discovery Notes:\n" + "\n".join(discovered_info) + "\n\n"
|
|
f"Data / Feed Sample:\n{sample_data}\n\n"
|
|
f"Determine the best endpoint_url, parser_type, and field mappings to regularly extract the latest news/articles from this site matching the user requirements."
|
|
)
|
|
|
|
try:
|
|
res = await self.llm.generate_json(
|
|
prompt=prompt,
|
|
system_prompt=WEBSITE_ANALYSIS_SYSTEM_PROMPT,
|
|
action_name="analyze_source_website"
|
|
)
|
|
if not isinstance(res, dict) or not res.get("endpoint_url"):
|
|
return False, {}, "پاسخ هوش مصنوعی شامل ساختار معتبر API برای این سایت نبود."
|
|
|
|
return True, res, res.get("summary", "سایت با موفقیت تحلیل شد.")
|
|
except Exception as e:
|
|
return False, {}, f"خطا در تحلیل ساختار توسط هوش مصنوعی: {e}"
|