feat: add source websites with automated endpoint analysis, target queue dispatch ordering, and markdown styling
This commit is contained in:
mamad
@@ -0,0 +1,263 @@
|
||||
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
|
||||
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, "وبسایت یافت نشد."
|
||||
|
||||
ok, api_cfg, summary = await self.analyzer.analyze_website(site.url)
|
||||
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
|
||||
ok, new_cfg, _ = await self.analyzer.analyze_website(site.url)
|
||||
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
|
||||
Reference in New Issue
Block a user