266 lines
11 KiB
Python
266 lines
11 KiB
Python
import os
|
|
import re
|
|
import html
|
|
import asyncio
|
|
import logging
|
|
import hashlib
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional, List, Dict, Any, Callable, Awaitable, Tuple
|
|
import httpx
|
|
from db.models import SourceWebsite
|
|
from db.repository import Repository
|
|
from services.website_analyzer import WebsiteAnalyzer, USER_AGENT
|
|
from core.error_logger import log_exception
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def _clean_html(raw_html: str) -> str:
|
|
if not raw_html:
|
|
return ""
|
|
text = re.sub(r'<[^>]+>', '', raw_html)
|
|
text = html.unescape(text)
|
|
return re.sub(r'\s+', ' ', text).strip()
|
|
|
|
def _extract_nested(data: Any, path: str) -> Any:
|
|
if not path or not data:
|
|
return data
|
|
parts = path.split(".")
|
|
curr = data
|
|
for p in parts:
|
|
if isinstance(curr, dict):
|
|
curr = curr.get(p)
|
|
elif isinstance(curr, list) and p.isdigit():
|
|
idx = int(p)
|
|
curr = curr[idx] if idx < len(curr) else None
|
|
else:
|
|
return None
|
|
return curr
|
|
|
|
class WebsiteCollectorService:
|
|
def __init__(
|
|
self,
|
|
repo: Repository,
|
|
analyzer: WebsiteAnalyzer,
|
|
ai_processor = None,
|
|
on_post_received: Optional[Callable[[int], Awaitable[None]]] = None,
|
|
on_error_alert: Optional[Callable[[str, List[Any]], Awaitable[None]]] = None,
|
|
):
|
|
self.repo = repo
|
|
self.analyzer = analyzer
|
|
self.ai_processor = ai_processor
|
|
self.on_post_received = on_post_received
|
|
self.on_error_alert = on_error_alert
|
|
self._running = False
|
|
self._task: Optional[asyncio.Task] = None
|
|
|
|
async def start(self):
|
|
logger.info("Starting Website Collector Service...")
|
|
self._running = True
|
|
self._task = asyncio.create_task(self._collector_loop())
|
|
|
|
async def stop(self):
|
|
self._running = False
|
|
if self._task:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
logger.info("Website Collector Service stopped.")
|
|
|
|
async def _collector_loop(self):
|
|
while self._running:
|
|
try:
|
|
if not await self.repo.is_system_paused():
|
|
await self._process_websites()
|
|
except Exception as e:
|
|
await log_exception("website_collector.loop", e)
|
|
await asyncio.sleep(60)
|
|
|
|
async def _process_websites(self):
|
|
websites = await self.repo.get_active_source_websites()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
for site in websites:
|
|
try:
|
|
# 1. Check Periodic Auto Re-analysis
|
|
if site.auto_reanalyze_hours > 0 and site.last_reanalyzed_at:
|
|
last_re = site.last_reanalyzed_at
|
|
if last_re.tzinfo is None:
|
|
last_re = last_re.replace(tzinfo=timezone.utc)
|
|
if (now - last_re).total_seconds() / 3600.0 >= site.auto_reanalyze_hours:
|
|
logger.info(f"Auto re-analyzing website #{site.id} ({site.name})...")
|
|
await self.reanalyze_website(site.id)
|
|
|
|
# 2. Check Fetch Interval
|
|
if site.last_fetched_at:
|
|
last_fe = site.last_fetched_at
|
|
if last_fe.tzinfo is None:
|
|
last_fe = last_fe.replace(tzinfo=timezone.utc)
|
|
if (now - last_fe).total_seconds() / 60.0 < site.check_interval_min:
|
|
continue
|
|
|
|
await self.fetch_website_posts(site.id)
|
|
except Exception as e:
|
|
logger.error(f"Error processing website {site.name}: {e}", exc_info=True)
|
|
|
|
async def reanalyze_website(self, site_id: int) -> Tuple[bool, str]:
|
|
site = await self.repo.get_source_website_by_id(site_id)
|
|
if not site:
|
|
return False, "وبسایت یافت نشد."
|
|
|
|
custom_inst = getattr(site, "custom_instructions", "") or ""
|
|
ok, api_cfg, summary = await self.analyzer.analyze_website(site.url, custom_instructions=custom_inst)
|
|
if ok and api_cfg:
|
|
await self.repo.update_source_website_api_config(site_id, api_cfg)
|
|
return True, f"✅ وبسایت «{site.name}» با موفقیت تحلیل شد:\n{summary}"
|
|
else:
|
|
err_msg = summary or "خطا در تحلیل وبسایت"
|
|
await self.repo.update_source_website_fetch_status(site_id, error=err_msg)
|
|
if self.on_error_alert:
|
|
await self.on_error_alert(
|
|
f"⚠️ <b>خطا در تحلیل مجدد هوشمند وبسایت «{site.name}»:</b>\n<code>{err_msg}</code>",
|
|
site_id
|
|
)
|
|
return False, err_msg
|
|
|
|
async def fetch_website_posts(self, site_id: int) -> int:
|
|
site = await self.repo.get_source_website_by_id(site_id)
|
|
if not site or not site.is_active:
|
|
return 0
|
|
|
|
api_cfg = site.api_config or {}
|
|
endpoint_url = api_cfg.get("endpoint_url")
|
|
if not endpoint_url:
|
|
# Trigger initial analysis
|
|
custom_inst = getattr(site, "custom_instructions", "") or ""
|
|
ok, new_cfg, _ = await self.analyzer.analyze_website(site.url, custom_instructions=custom_inst)
|
|
if ok and new_cfg:
|
|
await self.repo.update_source_website_api_config(site_id, new_cfg)
|
|
api_cfg = new_cfg
|
|
endpoint_url = api_cfg.get("endpoint_url")
|
|
else:
|
|
await self.repo.update_source_website_fetch_status(site_id, error="بدون اندپوینت معتبر")
|
|
return 0
|
|
|
|
headers = api_cfg.get("headers") or {"User-Agent": USER_AGENT, "Accept": "*/*"}
|
|
parser_type = api_cfg.get("parser_type", "rss")
|
|
mappings = api_cfg.get("field_mappings", {})
|
|
|
|
articles = []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=20.0, follow_redirects=True) as client:
|
|
resp = await client.get(endpoint_url, headers=headers)
|
|
resp.raise_for_status()
|
|
|
|
if parser_type == "wordpress_json" or "json" in resp.headers.get("content-type", "") or parser_type == "rest_json":
|
|
raw_data = resp.json()
|
|
items = _extract_nested(raw_data, api_cfg.get("items_path", ""))
|
|
if isinstance(items, list):
|
|
for item in items[:15]:
|
|
title = _clean_html(str(_extract_nested(item, mappings.get("title", "title.rendered")) or ""))
|
|
content = _clean_html(str(_extract_nested(item, mappings.get("content", "content.rendered")) or ""))
|
|
link = str(_extract_nested(item, mappings.get("link", "link")) or "")
|
|
img_url = str(_extract_nested(item, mappings.get("image_url", "yoast_head_json.og_image.0.url")) or "")
|
|
pub_at = _extract_nested(item, mappings.get("published_at", "date_gmt"))
|
|
if title and link:
|
|
articles.append({
|
|
"title": title,
|
|
"content": content,
|
|
"link": link,
|
|
"image_url": img_url if img_url.startswith("http") else None,
|
|
"published_at": pub_at
|
|
})
|
|
|
|
elif parser_type == "rss" or "xml" in resp.headers.get("content-type", ""):
|
|
root = ET.fromstring(resp.content)
|
|
channel = root.find("channel") or root
|
|
items = channel.findall("item") or root.findall(".//item")
|
|
for item in items[:15]:
|
|
t_node = item.find("title")
|
|
l_node = item.find("link")
|
|
d_node = item.find("description")
|
|
p_node = item.find("pubDate")
|
|
enc_node = item.find("enclosure")
|
|
|
|
title = _clean_html(t_node.text if t_node is not None else "")
|
|
link = (l_node.text or "").strip() if l_node is not None else ""
|
|
desc = _clean_html(d_node.text if d_node is not None else "")
|
|
img_url = enc_node.attrib.get("url") if enc_node is not None else None
|
|
|
|
if title and link:
|
|
articles.append({
|
|
"title": title,
|
|
"content": desc,
|
|
"link": link,
|
|
"image_url": img_url,
|
|
"published_at": p_node.text if p_node is not None else None
|
|
})
|
|
|
|
await self.repo.update_source_website_fetch_status(site_id, error=None)
|
|
|
|
except Exception as e:
|
|
err_msg = str(e)
|
|
logger.warning(f"Error fetching website #{site.id} ({site.name}): {err_msg}")
|
|
await self.repo.update_source_website_fetch_status(site_id, error=err_msg)
|
|
if self.on_error_alert:
|
|
await self.on_error_alert(
|
|
f"⚠️ <b>خطا در دریافت اطلاعات از وبسایت «{site.name}»:</b>\n"
|
|
f"🌐 URL: <code>{endpoint_url}</code>\n"
|
|
f"❌ خطا: <code>{err_msg}</code>",
|
|
site_id
|
|
)
|
|
return 0
|
|
|
|
# Ingest new articles into database
|
|
new_count = 0
|
|
pseudo_channel_id = -900000000 - site.id
|
|
|
|
for art in reversed(articles):
|
|
link = art["link"]
|
|
text_body = f"📌 <b>{art['title']}</b>\n\n{art['content']}\n\n🔗 منبع: {link}"
|
|
msg_id = int(hashlib.md5(link.encode("utf-8")).hexdigest()[:8], 16) % 1000000000
|
|
|
|
# Tags & Semantic deduplication
|
|
tags, subject = [], art["title"]
|
|
if self.ai_processor:
|
|
try:
|
|
tags, subject = await self.ai_processor.extract_tags_and_subject(art['title'] + "\n" + art['content'])
|
|
except Exception:
|
|
pass
|
|
|
|
is_duplicate = False
|
|
duplicate_of_id = None
|
|
similarity_reason = None
|
|
if tags and self.ai_processor:
|
|
try:
|
|
candidates = await self.repo.find_candidate_posts_by_tags(tags, limit=10)
|
|
if candidates:
|
|
is_dup, dup_id, reason = await self.ai_processor.check_semantic_duplicate(text_body, candidates)
|
|
if is_dup:
|
|
is_duplicate = True
|
|
duplicate_of_id = dup_id
|
|
similarity_reason = reason
|
|
except Exception:
|
|
pass
|
|
|
|
post_id = await self.repo.create_raw_post(
|
|
source_channel_id=pseudo_channel_id,
|
|
source_message_id=msg_id,
|
|
raw_text=text_body,
|
|
tags=tags,
|
|
subject=subject,
|
|
is_duplicate=is_duplicate,
|
|
duplicate_of_id=duplicate_of_id,
|
|
similarity_reason=similarity_reason,
|
|
)
|
|
|
|
if post_id:
|
|
new_count += 1
|
|
logger.info(f"Ingested post ID {post_id} from source website '{site.name}'")
|
|
if self.on_post_received:
|
|
await self.on_post_received(post_id)
|
|
|
|
return new_count
|