diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index f2e6af0..2e4a464 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -2,12 +2,21 @@ 👤 ویرایش‌کننده: mamad -🔹 ارسال پیام‌های زمینه کانال مبدا به هوش مصنوعی (Source Context History): -• افزودن گزینه تنظیم تعداد پیام‌های قبلی کانال مبدا (۰ تا ۱۰ پیام) در منوی مدیریت کانال‌های مبدا در ربات تلگرام. -• استخراج خودکار آخرین پیام‌های معتبر هر کانال و تزریق آن‌ها به عنوان «زمینه و خط داستانی» به پرامپت هوش مصنوعی تا پست جدید با آگاهی از سیر رویدادهای قبلی بازنویسی شود. -📁 فایل‌های تغییریافته: services/ai_processor.py, services/admin_bot.py, db/models.py, db/repository.py, tests/test_context_and_source_time.py +🔹 وبسایت‌های مبدا و کشف خودکار API با هوش مصنوعی (Source Websites & AI Discovery): +• امکان تعریف انواع وبسایت‌ها و خبرگزاری‌ها به عنوان مبدا ورودی محتوا در کنار کانال‌های تلگرام. +• تحلیل خودکار ساختار سایت، کشف اندپوینت‌های REST API و فیدهای RSS توسط هوش مصنوعی و ذخیره در تنظیمات وبسایت. +• استخراج دوره‌ای اخبار جدید، تشخیص تکراری بودن با هوش مصنوعی و ارسال به صف انتشار کانال‌های مقصد. +• اعلام هوشمند خطاهای دریافت در کانال ادمین همراه با کلید «🤖 تحلیل مجدد هوشمند AI و تولید مجدد API». +📁 فایل‌های تغییریافته: services/website_analyzer.py, services/website_collector.py, services/admin_bot.py, db/database.py, db/models.py, db/repository.py, main.py -🔹 ثبت زمان واقعی ارسال پست در مبدا (Source Sent Timestamp): -• افزودن فیلد source_created_at به پایگاه‌داده و ذخیره مستقیم زمان واقعی ارسال پست در تلگرام (به جای صرفاً زمان دریافت محلی). -• مرتب‌سازی دقیق و زمانی پیام‌های اخیر بر اساس تاریخچه انتشار واقعی در کانال. -📁 فایل‌های تغییریافته: db/database.py, db/models.py, services/collector.py +🔹 تنظیم شیوه ارسال پست‌ها از صف (ترتیبی FIFO یا تصادفی): +• افزودن گزینه «🔀 شیوه ارسال» در تنظیمات هر کانال مقصد با قابلیت سوییچ بین حالت‌های «به ترتیب ورود» و «تصادفی (Random)». +📁 فایل‌های تغییریافته: core/queue.py, services/publisher.py, services/admin_bot.py, db/models.py, db/repository.py + +🔹 پشتیبانی کامل از استایل‌های Markdown در ارسال پست‌ها به تلگرام: +• پشتیبانی از فرمت‌بندی غنی Markdown (بولد، ایتالیک، لینک‌ها و کدهای برنامه‌نویسی) خروجی هوش مصنوعی در ارسال به کانال‌های مقصد. +📁 فایل‌های تغییریافته: services/publisher.py + +🔹 مدیریت هوشمند ارسال گزارش انتشار (Release Notes Broadcast Hash Tracking): +• ردیابی هش گزارش تغییرات جهت جلوگیری از ارسال پیام‌های تکراری هنگام ری‌استارت سیستم. +📁 فایل‌های تغییریافته: services/admin_bot.py diff --git a/core/queue.py b/core/queue.py index 6c7496f..a412589 100644 --- a/core/queue.py +++ b/core/queue.py @@ -30,11 +30,26 @@ class RedisQueue: await self.client.rpush(key, raw_json) logger.info(f"Enqueued post {payload.get('post_id')} to Target #{target_id} queue [{key}]") - async def pop_target_post(self, target_id: int) -> Optional[Dict[str, Any]]: + async def pop_target_post(self, target_id: int, dispatch_order: str = "order") -> Optional[Dict[str, Any]]: if not self.client: await self.connect() key = self._get_target_key(target_id) - raw = await self.client.lpop(key) + + if dispatch_order == "random": + import random + length = await self.client.llen(key) + if length == 0: + return None + if length == 1: + raw = await self.client.lpop(key) + else: + idx = random.randint(0, length - 1) + raw = await self.client.lindex(key, idx) + if raw: + await self.client.lrem(key, 1, raw) + else: + raw = await self.client.lpop(key) + if raw: try: return json.loads(raw) diff --git a/db/database.py b/db/database.py index e8f182c..c1b926a 100644 --- a/db/database.py +++ b/db/database.py @@ -33,6 +33,7 @@ CREATE TABLE IF NOT EXISTS targets ( is_sleep_enabled BOOLEAN DEFAULT FALSE, auto_source_ids BIGINT[] DEFAULT '{}', language VARCHAR(32) DEFAULT 'fa', + dispatch_order VARCHAR(32) DEFAULT 'order', last_post_time TIMESTAMPTZ, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP @@ -131,6 +132,24 @@ CREATE TABLE IF NOT EXISTS channel_categories ( created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP ); +CREATE TABLE IF NOT EXISTS source_websites ( + id SERIAL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + url TEXT NOT NULL UNIQUE, + category_id INT REFERENCES channel_categories(id) ON DELETE SET NULL, + check_interval_min INT DEFAULT 30, + auto_reanalyze_hours INT DEFAULT 24, + last_reanalyzed_at TIMESTAMPTZ, + last_fetched_at TIMESTAMPTZ, + api_config JSONB DEFAULT '{}'::jsonb, + last_error TEXT, + last_error_at TIMESTAMPTZ, + is_active BOOLEAN DEFAULT TRUE, + created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_source_websites_active ON source_websites(is_active); + CREATE INDEX IF NOT EXISTS idx_ai_providers_active ON ai_providers(is_active); -- Migration safety for existing tables @@ -152,6 +171,7 @@ ALTER TABLE targets ADD COLUMN IF NOT EXISTS is_sleep_enabled BOOLEAN DEFAULT FA ALTER TABLE targets ADD COLUMN IF NOT EXISTS auto_source_ids BIGINT[] DEFAULT '{}'; ALTER TABLE targets ADD COLUMN IF NOT EXISTS language VARCHAR(32) DEFAULT 'fa'; ALTER TABLE targets ADD COLUMN IF NOT EXISTS custom_prompt TEXT DEFAULT ''; +ALTER TABLE targets ADD COLUMN IF NOT EXISTS dispatch_order VARCHAR(32) DEFAULT 'order'; ALTER TABLE sources ADD COLUMN IF NOT EXISTS context_message_count INT DEFAULT 0; ALTER TABLE posts ADD COLUMN IF NOT EXISTS source_created_at TIMESTAMPTZ; CREATE INDEX IF NOT EXISTS idx_posts_source_created ON posts(source_channel_id, source_created_at DESC); diff --git a/db/models.py b/db/models.py index 13bf211..c6e09d9 100644 --- a/db/models.py +++ b/db/models.py @@ -20,6 +20,22 @@ class SourceChannel: is_active: bool = True created_at: Optional[str] = None +@dataclass +class SourceWebsite: + id: Optional[int] + name: str + url: str + category_id: Optional[int] = None + check_interval_min: int = 30 + auto_reanalyze_hours: int = 24 + last_reanalyzed_at: Optional[str] = None + last_fetched_at: Optional[str] = None + api_config: Dict[str, Any] = field(default_factory=dict) + last_error: Optional[str] = None + last_error_at: Optional[str] = None + is_active: bool = True + created_at: Optional[str] = None + @dataclass class TargetChannel: id: Optional[int] @@ -37,6 +53,7 @@ class TargetChannel: # Telegram channel_ids of sources whose posts are queued to this target automatically. auto_source_ids: List[int] = field(default_factory=list) language: str = "fa" + dispatch_order: str = "order" # "order" (FIFO) or "random" last_post_time: Optional[str] = None is_active: bool = True created_at: Optional[str] = None diff --git a/db/repository.py b/db/repository.py index d842d0e..1b775e3 100644 --- a/db/repository.py +++ b/db/repository.py @@ -2,7 +2,7 @@ import json import asyncpg from datetime import datetime, timezone from typing import List, Optional, Dict, Any, Tuple -from db.models import SourceChannel, TargetChannel, Post, Setting, AILog, AIProviderProfile, ChannelCategory +from db.models import SourceChannel, SourceWebsite, TargetChannel, Post, Setting, AILog, AIProviderProfile, ChannelCategory from db.database import get_db_pool @@ -17,6 +17,18 @@ def _parse_post_row(row: asyncpg.Record) -> Post: data["published_to"] = [] return Post(**data) + +def _parse_source_website_row(row: asyncpg.Record) -> SourceWebsite: + data = dict(row) + if isinstance(data.get("api_config"), str): + try: + data["api_config"] = json.loads(data["api_config"]) + except Exception: + data["api_config"] = {} + elif data.get("api_config") is None: + data["api_config"] = {} + return SourceWebsite(**data) + class Repository: def __init__(self, dsn: Optional[str] = None): self.dsn = dsn @@ -72,6 +84,117 @@ class Repository: async with pool.acquire() as conn: await conn.execute("UPDATE sources SET is_active = FALSE WHERE id = $1;", source_id) + # --- Source Websites --- + async def add_source_website( + self, + name: str, + url: str, + category_id: Optional[int] = None, + check_interval_min: int = 30, + auto_reanalyze_hours: int = 24, + api_config: Optional[Dict[str, Any]] = None + ) -> int: + pool = await self._get_pool() + async with pool.acquire() as conn: + cfg_json = json.dumps(api_config or {}) + row = await conn.fetchrow( + """ + INSERT INTO source_websites (name, url, category_id, check_interval_min, auto_reanalyze_hours, api_config) + VALUES ($1, $2, $3, $4, $5, $6::jsonb) + ON CONFLICT(url) DO UPDATE SET + name = EXCLUDED.name, + category_id = COALESCE(EXCLUDED.category_id, source_websites.category_id), + is_active = TRUE + RETURNING id; + """, + name, url, category_id, check_interval_min, auto_reanalyze_hours, cfg_json + ) + return row["id"] + + async def get_active_source_websites(self) -> List[SourceWebsite]: + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch("SELECT * FROM source_websites WHERE is_active = TRUE ORDER BY id ASC;") + return [_parse_source_website_row(r) for r in rows] + + async def get_source_websites(self) -> List[SourceWebsite]: + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch("SELECT * FROM source_websites WHERE is_active = TRUE ORDER BY id ASC;") + return [_parse_source_website_row(r) for r in rows] + + async def get_source_website_by_id(self, site_id: int) -> Optional[SourceWebsite]: + pool = await self._get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT * FROM source_websites WHERE id = $1;", site_id) + return _parse_source_website_row(row) if row else None + + async def get_source_websites_by_category(self, category_id: Optional[int]) -> List[SourceWebsite]: + pool = await self._get_pool() + async with pool.acquire() as conn: + if category_id is None: + rows = await conn.fetch("SELECT * FROM source_websites WHERE is_active = TRUE AND category_id IS NULL ORDER BY id ASC;") + else: + rows = await conn.fetch("SELECT * FROM source_websites WHERE is_active = TRUE AND category_id = $1 ORDER BY id ASC;", category_id) + return [_parse_source_website_row(r) for r in rows] + + async def update_source_website_category(self, site_id: int, category_id: Optional[int]) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute("UPDATE source_websites SET category_id = $1 WHERE id = $2;", category_id, site_id) + + async def update_source_website_api_config(self, site_id: int, api_config: Dict[str, Any]) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute( + """ + UPDATE source_websites + SET api_config = $1::jsonb, + last_reanalyzed_at = CURRENT_TIMESTAMP, + last_error = NULL, + last_error_at = NULL + WHERE id = $2; + """, + json.dumps(api_config), site_id + ) + + async def update_source_website_fetch_status(self, site_id: int, error: Optional[str] = None) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + if error: + await conn.execute( + """ + UPDATE source_websites + SET last_error = $1, last_error_at = CURRENT_TIMESTAMP + WHERE id = $2; + """, + error, site_id + ) + else: + await conn.execute( + """ + UPDATE source_websites + SET last_fetched_at = CURRENT_TIMESTAMP, last_error = NULL, last_error_at = NULL + WHERE id = $1; + """, + site_id + ) + + async def update_source_website_interval(self, site_id: int, interval_min: int) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute("UPDATE source_websites SET check_interval_min = $1 WHERE id = $2;", max(1, interval_min), site_id) + + async def update_source_website_reanalyze_hours(self, site_id: int, hours: int) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute("UPDATE source_websites SET auto_reanalyze_hours = $1 WHERE id = $2;", max(0, hours), site_id) + + async def delete_source_website(self, site_id: int) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute("UPDATE source_websites SET is_active = FALSE WHERE id = $1;", site_id) + async def delete_target(self, target_id: int) -> None: pool = await self._get_pool() async with pool.acquire() as conn: @@ -142,6 +265,15 @@ class Repository: custom_prompt, target_id ) + async def update_target_dispatch_order(self, target_id: int, dispatch_order: str) -> None: + order = "random" if str(dispatch_order).strip().lower() == "random" else "order" + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute( + "UPDATE targets SET dispatch_order = $1 WHERE id = $2;", + order, target_id + ) + async def update_target_schedule( self, diff --git a/main.py b/main.py index 2aab119..fb396ff 100644 --- a/main.py +++ b/main.py @@ -13,6 +13,8 @@ from services.ai_processor import AIProcessor from services.collector import CollectorService from services.admin_bot import AdminBotService from services.publisher import PublisherService +from services.website_analyzer import WebsiteAnalyzer +from services.website_collector import WebsiteCollectorService # Load environment variables load_dotenv() @@ -51,6 +53,17 @@ async def main(): ) ai_processor = AIProcessor(repo=repo, llm=llm) admin_bot.set_ai_processor(ai_processor) + + website_analyzer = WebsiteAnalyzer(llm=llm) + website_collector = WebsiteCollectorService( + repo=repo, + analyzer=website_analyzer, + ai_processor=ai_processor, + on_post_received=admin_bot.handle_collected_post, + on_error_alert=admin_bot.on_website_error_alert + ) + admin_bot.set_website_services(website_analyzer, website_collector) + collector = CollectorService(repo=repo, on_post_received=admin_bot.handle_collected_post, ai_processor=ai_processor) admin_bot.set_collector(collector) @@ -67,6 +80,7 @@ async def main(): # 4. Start all services await admin_bot.start() await collector.start(notify_fn=admin_bot.notify_admins) + await website_collector.start() await publisher.start() logger.info("All Copykar services are active and running.") @@ -91,6 +105,7 @@ async def main(): pass finally: logger.info("Shutting down Copykar services...") + await website_collector.stop() await collector.stop() await publisher.stop() await admin_bot.stop() diff --git a/services/admin_bot.py b/services/admin_bot.py index 93c24f8..a5fddf9 100644 --- a/services/admin_bot.py +++ b/services/admin_bot.py @@ -46,10 +46,11 @@ def get_persian_main_menu(is_paused: bool = False): return [ [Button.text("📊 آمار و وضعیت ناوگان", resize=True), Button.text("🔑 درخواست کد لاگین", resize=True)], [Button.text("📡 کانال‌های مبدا", resize=True), Button.text("🎯 کانال‌های مقصد", resize=True)], - [Button.text("📂 دسته‌بندی کانال‌ها", resize=True), Button.text("🧠 تنظیمات و لاگ‌های AI", resize=True)], + [Button.text("🌐 وبسایت‌های مبدا", resize=True), Button.text("📂 دسته‌بندی کانال‌ها", resize=True)], + [Button.text("🧠 تنظیمات و لاگ‌های AI", resize=True), Button.text("⚠️ خطاهای سیستم", resize=True)], [Button.text("➕ افزودن کانال مبدا", resize=True), Button.text("➕ افزودن کانال مقصد", resize=True)], - [Button.text("📨 ارسال پست‌های بررسی‌نشده", resize=True), Button.text("⚠️ خطاهای سیستم", resize=True)], - [Button.text(pause_btn, resize=True), Button.text("❓ راهنمای سیستم", resize=True)] + [Button.text("📨 ارسال پست‌های بررسی‌نشده", resize=True), Button.text(pause_btn, resize=True)], + [Button.text("❓ راهنمای سیستم", resize=True)] ] @@ -120,6 +121,23 @@ class AdminBotService: def set_queue(self, queue: RedisQueue): self.queue = queue + def set_website_services(self, analyzer, collector): + self.website_analyzer = analyzer + self.website_collector = collector + + async def on_website_error_alert(self, alert_text: str, site_id: int): + buttons = [[Button.inline("🤖 تحلیل مجدد هوشمند وبسایت", data=f"web_reanalyze:{site_id}")]] + if self.review_channel_id: + try: + await self.client.send_message(self.review_channel_id, alert_text, parse_mode="html", buttons=buttons) + except Exception as e: + logger.error(f"Failed to send website alert to review channel: {e}") + for admin_id in self.admin_user_ids: + try: + await self.client.send_message(admin_id, alert_text, parse_mode="html", buttons=buttons) + except Exception: + pass + def is_admin(self, user_id: int) -> bool: return not self.admin_user_ids or user_id in self.admin_user_ids @@ -250,12 +268,16 @@ class AdminBotService: cat_name = cat.name custom_p = getattr(target, "custom_prompt", "") or "" + order_mode = getattr(target, "dispatch_order", "order") or "order" + order_label = "🟢 به ترتیب ورود (FIFO)" if order_mode == "order" else "🎲 تصادفی (Random)" + card_text = ( f"🎯 تنظیمات کانال مقصد: {target.title}\n\n" f"• 🆔 شناسه کانال: {target.channel_id} (ID: {target.id})\n" f"• 🔗 یوزرنیم: @{target.username or 'ندارد'}\n" f"• 📁 دسته‌بندی: {cat_name}\n" f"• 🌐 زبان کانال: {lang_label}\n" + f"• 🔀 ترتیب ارسال از صف: {order_label}\n" f"• ⏱ فاصله ارسال پست‌ها: هر {target.post_interval_min} دقیقه\n" f"• 🌙 وضعیت ساعت خواب: {sleep_st}\n" f"• 📥 پست‌های منتظر در صف: {qsize} پست\n" @@ -280,10 +302,13 @@ class AdminBotService: ], [ Button.inline("🌐 تنظیم زبان", data=f"st_lang:{target.id}"), - Button.inline("⏱ تغییر فاصله ارسال", data=f"st_intv:{target.id}") + Button.inline(f"🔀 شیوه ارسال ({'ترتیبی' if order_mode == 'order' else 'تصادفی'})", data=f"trg_order:{target.id}") + ], + [ + Button.inline("⏱ تغییر فاصله ارسال", data=f"st_intv:{target.id}"), + Button.inline("🌙 تنظیم ساعت خواب", data=f"st_slp:{target.id}") ], [ - Button.inline("🌙 تنظیم ساعت خواب", data=f"st_slp:{target.id}"), Button.inline("🤖 ارسال خودکار از مبدا", data=f"auto_src:{target.id}") ] ] @@ -905,6 +930,93 @@ class AdminBotService: ] return text, buttons + async def _render_website_list(self) -> Tuple[str, List[List[Button]]]: + sites = await self.repo.get_source_websites() + if not sites: + text = ( + "🌐 لیست وبسایت‌های مبدا (Source Websites)\n\n" + "هنوز هیچ وبسایت مبدایی ثبت نشده است.\n" + "شما می‌توانید هر وبسایت، خبرگزاری یا وبلاگی را به عنوان مبدا ثبت کنید تا هوش مصنوعی به‌طور خودکار ساختار آن را تحلیل و محتوای جدید را استخراج کند." + ) + buttons = [ + [Button.inline("➕ افزودن وبسایت مبدا جدید", data="add_web")], + [Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main")] + ] + return text, buttons + + text = f"🌐 لیست وبسایت‌های مبدا ({len(sites)} وبسایت):\n\nبرای مشاهده تنظیمات، دریافت دستی یا تحلیل مجدد روی وبسایت کلیک کنید:" + buttons = [] + for s in sites: + status_icon = "🔴" if s.last_error else "🟢" + buttons.append([Button.inline(f"{status_icon} {s.name} ({s.check_interval_min}m)", data=f"web_view:{s.id}")]) + buttons.append([Button.inline("➕ افزودن وبسایت مبدا جدید", data="add_web")]) + buttons.append([Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main")]) + return text, buttons + + async def _render_website_config(self, site_id: int) -> Tuple[str, List[List[Button]]]: + site = await self.repo.get_source_website_by_id(site_id) + if not site: + return "❌ وبسایت یافت نشد.", [] + + cat_name = "بدون دسته‌بندی" + if site.category_id: + cat = await self.repo.get_category_by_id(site.category_id) + if cat: + cat_name = cat.name + + api_cfg = site.api_config or {} + parser_type = api_cfg.get("parser_type", "نامشخص") + endpoint = api_cfg.get("endpoint_url", site.url) + err_block = f"⚠️ خطای اخیر: {site.last_error}\n\n" if site.last_error else "" + last_fe = site.last_fetched_at.strftime("%Y-%m-%d %H:%M") if site.last_fetched_at else "هنوز دریافت نشده" + last_re = site.last_reanalyzed_at.strftime("%Y-%m-%d %H:%M") if site.last_reanalyzed_at else "انجام نشده" + + text = ( + f"🌐 وبسایت مبدا: {site.name}\n\n" + f"• 🔗 آدرس وبسایت: {site.url}\n" + f"• 🔌 اندپوینت دریافت: {endpoint}\n" + f"• 🏷 نوع ساختار: {parser_type}\n" + f"• 📁 دسته‌بندی: {cat_name}\n" + f"• ⏱ فاصله بررسی: هر {site.check_interval_min} دقیقه\n" + f"• 🔄 دوره تحلیل مجدد AI: هر {site.auto_reanalyze_hours} ساعت\n" + f"• 📥 آخرین دریافت: {last_fe}\n" + f"• 🧠 آخرین تحلیل هوش مصنوعی: {last_re}\n\n" + f"{err_block}" + f"👇 عملیات مورد نظر را انتخاب کنید:" + ) + buttons = [ + [ + Button.inline("🔄 دریافت دستی (Fetch Now)", data=f"web_fetch:{site.id}"), + Button.inline("🤖 تحلیل مجدد هوشمند AI", data=f"web_reanalyze:{site.id}") + ], + [ + Button.inline("⏱ تغییر فاصله بررسی", data=f"web_intv:{site.id}"), + Button.inline("🔄 دوره تحلیل مجدد AI", data=f"web_reintv:{site.id}") + ], + [ + Button.inline("📁 تعیین دسته‌بندی", data=f"web_cat:{site.id}"), + Button.inline("🗑 حذف این وبسایت", data=f"del_web:{site.id}") + ], + [ + Button.inline("🔙 بازگشت به لیست وبسایت‌ها", data="list_web") + ] + ] + return text, buttons + + async def _render_website_category_menu(self, site_id: int): + site = await self.repo.get_source_website_by_id(site_id) + if not site: + return "❌ وبسایت یافت نشد.", [] + cats = await self.repo.get_categories() + text = f"📁 انتخاب دسته‌بندی برای وبسایت «{site.name}»:" + buttons = [] + for c in cats: + mark = "✅ " if site.category_id == c.id else "" + buttons.append([Button.inline(f"{mark}{c.name}", data=f"web_set_cat:{site.id}:{c.id}")]) + buttons.append([Button.inline("❌ بدون دسته‌بندی", data=f"web_set_cat:{site.id}:none")]) + buttons.append([Button.inline("🔙 بازگشت به تنظیمات وبسایت", data=f"web_view:{site.id}")]) + return text, buttons + async def _render_auto_sources(self, target_id: int): """Toggle screen listing every source with its on/off state for this target.""" @@ -1217,6 +1329,14 @@ class AdminBotService: text, buttons = await self._render_target_list() await event.reply(text, parse_mode="html", buttons=buttons or None) + # --- Source Websites --- + @self.client.on(events.NewMessage(pattern=r"(?i)^(/websites|/sites|🌐 وبسایت‌های مبدا)$")) + async def cmd_websites(event: events.NewMessage.Event): + if not self.is_admin(event.sender_id): + return + text, buttons = await self._render_website_list() + await event.reply(text, parse_mode="html", buttons=buttons or None) + # --- Categories Management --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/categories|/cats|📂 دسته‌بندی کانال‌ها)$")) async def cmd_categories(event: events.NewMessage.Event): @@ -1333,9 +1453,23 @@ class AdminBotService: return "📢 گزارش تغییرات سیستم (Release Notes):\nسیستم با آخرین به‌روزرسانی‌ها در حال اجرا است." async def broadcast_change_notes(self): - """Send latest change notes to admins and review channel.""" + """Send latest change notes to admins and review channel if they are new.""" + import hashlib notes = self.get_latest_change_notes() + current_hash = hashlib.sha256(notes.encode("utf-8")).hexdigest() + last_hash = await self.repo.get_setting("last_broadcast_release_hash") + + if last_hash == current_hash: + logger.info("Release notes have not changed since last broadcast. Skipping automatic broadcast.") + return + + logger.info(f"Broadcasting new release notes (hash: {current_hash[:8]})...") await self.notify_admins(notes) + await self.repo.set_setting( + "last_broadcast_release_hash", + current_hash, + description="Hash of last broadcast release notes" + ) # --- Step-by-Step Parameter & Text Input Message Handler --- @@ -1593,6 +1727,73 @@ class AdminBotService: menu = await self.get_menu() await event.reply(res, parse_mode="html", buttons=menu) + # 8.1. Waiting for Website Input (Name and URL) + elif action == "wait_website_input": + self.user_states.pop(event.sender_id, None) + parts = text.split() + if len(parts) < 2: + menu = await self.get_menu() + await event.reply("❌ فرمت نامعتبر است. لطفاً نام و آدرس را با فاصله ارسال کنید (مثال: دیجیاتو https://digiato.com).", parse_mode="html", buttons=menu) + return + + # If last part is URL: + if parts[-1].startswith(("http://", "https://", "www.")) or "." in parts[-1]: + url = parts[-1] + name = " ".join(parts[:-1]) + else: + url = parts[0] + name = " ".join(parts[1:]) + + if not url.startswith("http://") and not url.startswith("https://"): + url = "https://" + url + + loading_msg = await event.reply( + f"🤖 در حال تحلیل وبسایت «{name}» ({url}) توسط هوش مصنوعی...\n(کشف خودکار REST APIها، فیدهای RSS و ساختار داده)", + parse_mode="html" + ) + + api_cfg = {} + if self.website_analyzer: + ok, cfg, summary = await self.website_analyzer.analyze_website(url) + if ok and cfg: + api_cfg = cfg + + site_id = await self.repo.add_source_website(name=name, url=url, api_config=api_cfg) + card, buttons = await self._render_website_config(site_id) + await loading_msg.edit( + f"✅ وبسایت «{name}» با موفقیت افزوده و پیکربندی شد!\n\n{card}", + parse_mode="html", + buttons=buttons + ) + + # 8.2. Waiting for Website Fetch Interval + elif action == "wait_website_intv": + site_id = state.get("site_id") + self.user_states.pop(event.sender_id, None) + try: + val = int(text.strip()) + await self.repo.update_source_website_interval(site_id, val) + card, buttons = await self._render_website_config(site_id) + await event.reply(f"✅ فاصله بررسی وبسایت روی هر {val} دقیقه تنظیم شد.", parse_mode="html") + await event.reply(card, parse_mode="html", buttons=buttons) + except ValueError: + menu = await self.get_menu() + await event.reply("❌ مقدار وارد شده باید یک عدد صحیح (به دقیقه) باشد.", buttons=menu) + + # 8.3. Waiting for Website Auto Re-analysis Interval + elif action == "wait_website_reintv": + site_id = state.get("site_id") + self.user_states.pop(event.sender_id, None) + try: + val = int(text.strip()) + await self.repo.update_source_website_reanalyze_hours(site_id, val) + card, buttons = await self._render_website_config(site_id) + await event.reply(f"✅ دوره تحلیل مجدد خودکار وبسایت روی هر {val} ساعت تنظیم شد.", parse_mode="html") + await event.reply(card, parse_mode="html", buttons=buttons) + except ValueError: + menu = await self.get_menu() + await event.reply("❌ مقدار وارد شده باید یک عدد صحیح (به ساعت) باشد.", buttons=menu) + # 9. Waiting for AI Model elif action == "wait_ai_model": self.user_states.pop(event.sender_id, None) @@ -2275,6 +2476,109 @@ class AdminBotService: await event.edit(card, parse_mode="html", buttons=buttons) await event.answer() + elif data == "list_web": + text, buttons = await self._render_website_list() + await event.edit(text, parse_mode="html", buttons=buttons or None) + await event.answer() + + elif data.startswith("web_view:"): + site_id = int(data.split(":")[1]) + card, buttons = await self._render_website_config(site_id) + await event.edit(card, parse_mode="html", buttons=buttons) + await event.answer() + + elif data == "add_web": + self.user_states[event.sender_id] = {"action": "wait_website_input"} + guide = ( + "🌐 افزودن وبسایت مبدا جدید:\n\n" + "لطفاً نام وبسایت و آدرس URL آن را با یک فاصله یا در دو خط ارسال کنید:\n" + "(مثال: دیجیاتو https://digiato.com)\n\n" + "🤖 هوش مصنوعی بلافاصله وبسایت را بررسی کرده و اندپوینت‌های دریافت خبر را کشف می‌کند." + ) + await event.reply(guide, parse_mode="html", buttons=get_cancel_button()) + await event.answer() + + elif data.startswith("web_fetch:"): + site_id = int(data.split(":")[1]) + await event.answer("🔄 در حال دریافت آخرین پست‌ها از وبسایت...") + cnt = await self.website_collector.fetch_website_posts(site_id) if self.website_collector else 0 + card, buttons = await self._render_website_config(site_id) + await event.edit(card, parse_mode="html", buttons=buttons) + await event.answer(f"✅ تعداد {cnt} پست جدید دریافت و پردازش شد.", alert=True) + + elif data.startswith("web_reanalyze:"): + site_id = int(data.split(":")[1]) + await event.answer("🤖 در حال تحلیل مجدد ساختار وبسایت توسط هوش مصنوعی...") + ok, msg = await self.website_collector.reanalyze_website(site_id) if self.website_collector else (False, "سرویس در دسترس نیست") + card, buttons = await self._render_website_config(site_id) + await event.edit(card, parse_mode="html", buttons=buttons) + await event.answer("✅ تحلیل مجدد با موفقیت انجام شد." if ok else f"❌ خطا: {msg}", alert=True) + + elif data.startswith("web_intv:"): + site_id = int(data.split(":")[1]) + site = await self.repo.get_source_website_by_id(site_id) + if not site: + await event.answer("وبسایت یافت نشد.", alert=True) + return + self.user_states[event.sender_id] = {"action": "wait_website_intv", "site_id": site_id} + await event.reply( + f"⏱ تنظیم فاصله بررسی وبسایت «{site.name}»:\n\n" + f"فاصله زمانی بررسی و استخراج اخبار (به دقیقه) را ارسال کنید:\n" + f"(مثال: 30 برای هر ۳۰ دقیقه)", + parse_mode="html", + buttons=get_cancel_button() + ) + await event.answer() + + elif data.startswith("web_reintv:"): + site_id = int(data.split(":")[1]) + site = await self.repo.get_source_website_by_id(site_id) + if not site: + await event.answer("وبسایت یافت نشد.", alert=True) + return + self.user_states[event.sender_id] = {"action": "wait_website_reintv", "site_id": site_id} + await event.reply( + f"🔄 تنظیم دوره تحلیل مجدد هوشمند وبسایت «{site.name}»:\n\n" + f"فاصله تحلیل خودکار ساختار و کشف اندپوینت‌ها توسط AI (به ساعت) را ارسال کنید:\n" + f"(مثال: 24 برای هر ۲۴ ساعت، یا 0 برای غیرفعال‌سازی)", + parse_mode="html", + buttons=get_cancel_button() + ) + await event.answer() + + elif data.startswith("web_cat:"): + site_id = int(data.split(":")[1]) + text, buttons = await self._render_website_category_menu(site_id) + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer() + + elif data.startswith("web_set_cat:"): + _, site_id_str, cat_id_str = data.split(":") + site_id = int(site_id_str) + cat_id = None if cat_id_str == "none" else int(cat_id_str) + await self.repo.update_source_website_category(site_id, cat_id) + card, buttons = await self._render_website_config(site_id) + await event.edit(card, parse_mode="html", buttons=buttons) + await event.answer("✅ دسته‌بندی وبسایت به‌روز شد.") + + elif data.startswith("del_web:"): + site_id = int(data.split(":")[1]) + await self.repo.delete_source_website(site_id) + text, buttons = await self._render_website_list() + await event.edit(text, parse_mode="html", buttons=buttons or None) + await event.answer("🗑 وبسایت مبدا با موفقیت حذف شد.", alert=True) + + elif data.startswith("trg_order:"): + target_id = int(data.split(":")[1]) + target = await self.repo.get_target_by_id(target_id) + if target: + curr_order = getattr(target, "dispatch_order", "order") or "order" + new_order = "random" if curr_order == "order" else "order" + await self.repo.update_target_dispatch_order(target_id, new_order) + card, buttons = await self._render_target_config(target_id) + await event.edit(card, parse_mode="html", buttons=buttons) + await event.answer("🎲 شیوه ارسال روی تصادفی تنظیم شد." if new_order == "random" else "🟢 شیوه ارسال روی به ترتیب تنظیم شد.") + elif data.startswith("auto_src:"): target_id = int(data.split(":")[1]) text, buttons = await self._render_auto_sources(target_id) diff --git a/services/publisher.py b/services/publisher.py index a79fc93..ccc078d 100644 --- a/services/publisher.py +++ b/services/publisher.py @@ -2,7 +2,7 @@ import os import asyncio import logging from datetime import datetime, timedelta, timezone -from typing import Optional, List, Awaitable, Callable, Set, Tuple +from typing import Optional, List, Awaitable, Callable, Set, Tuple, Any from telethon import TelegramClient from telethon.errors import ( ChannelPrivateError, @@ -35,6 +35,33 @@ PERMANENT_DELIVERY_ERRORS = ( PeerIdInvalidError, ) +async def send_styled_message(client: TelegramClient, entity: Any, text: str, file_path: Optional[str] = None): + """Deliver message or media trying Markdown first, then HTML, then raw plain text.""" + if not text and not file_path: + return None + + # 1. Try standard Markdown first (AI's natural formatting) + try: + if file_path and os.path.exists(file_path): + return await client.send_file(entity, file=file_path, caption=text, parse_mode="md") + return await client.send_message(entity, text, parse_mode="md") + except Exception as md_err: + logger.debug(f"Markdown send failed ({md_err}), trying HTML...") + + # 2. Try HTML format fallback + try: + if file_path and os.path.exists(file_path): + return await client.send_file(entity, file=file_path, caption=text, parse_mode="html") + return await client.send_message(entity, text, parse_mode="html") + except Exception as html_err: + logger.debug(f"HTML send failed ({html_err}), falling back to plain text...") + + # 3. Plain text safety fallback + if file_path and os.path.exists(file_path): + return await client.send_file(entity, file=file_path, caption=text, parse_mode=None) + return await client.send_message(entity, text, parse_mode=None) + + class PublisherService: def __init__( self, @@ -129,7 +156,8 @@ class PublisherService: continue # 3. Pop next post payload for this target - payload = await self.queue.pop_target_post(target.id) + disp_order = getattr(target, "dispatch_order", "order") or "order" + payload = await self.queue.pop_target_post(target.id, dispatch_order=disp_order) if not payload: continue @@ -138,19 +166,7 @@ class PublisherService: media_path = payload.get("media_path") try: - if media_path and os.path.exists(media_path): - await self.client.send_file( - target.channel_id, - file=media_path, - caption=text, - parse_mode="html" - ) - else: - await self.client.send_message( - target.channel_id, - text, - parse_mode="html" - ) + await send_styled_message(self.client, target.channel_id, text, file_path=media_path) # Record publication in database await self.repo.record_post_published_to_target(post_id, target.id, target.title or "Target") diff --git a/services/website_analyzer.py b/services/website_analyzer.py new file mode 100644 index 0000000..3312912 --- /dev/null +++ b/services/website_analyzer.py @@ -0,0 +1,130 @@ +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) -> 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']+type=["\']application/(?:rss|atom)\+xml["\'][^>]+href=["\']([^"\']+)["\']', html_text, re.I) + if not rss_links: + rss_links = re.findall(r']+href=["\']([^"\']+)["\'][^>]+type=["\']application/(?:rss|atom)\+xml["\']', html_text, re.I) + + # Check for WordPress REST API link + wp_api = re.findall(r']+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}" + + prompt = ( + f"Website Target URL: {clean_url}\n" + 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." + ) + + 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}" diff --git a/services/website_collector.py b/services/website_collector.py new file mode 100644 index 0000000..c587683 --- /dev/null +++ b/services/website_collector.py @@ -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"⚠️ خطا در تحلیل مجدد هوشمند وبسایت «{site.name}»:\n{err_msg}", + 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"⚠️ خطا در دریافت اطلاعات از وبسایت «{site.name}»:\n" + f"🌐 URL: {endpoint_url}\n" + f"❌ خطا: {err_msg}", + 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"📌 {art['title']}\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 diff --git a/tests/test_websites_and_dispatch_order.py b/tests/test_websites_and_dispatch_order.py new file mode 100644 index 0000000..b1cb0c1 --- /dev/null +++ b/tests/test_websites_and_dispatch_order.py @@ -0,0 +1,147 @@ +import asyncio +import time +import json +from unittest.mock import AsyncMock, MagicMock, patch +from db.database import init_db +from db.repository import Repository +from db.models import TargetChannel, SourceWebsite +from core.queue import RedisQueue +from services.website_analyzer import WebsiteAnalyzer +from services.website_collector import WebsiteCollectorService, _clean_html, _extract_nested + + +async def test_target_dispatch_order_and_queue(): + await init_db() + repo = Repository() + + # 1. Target creation and dispatch_order field + unique_channel_id = -10088776655 - int(time.time() % 100000) + t_id = await repo.add_target(channel_id=unique_channel_id, title="Dispatch Test Target", username="disp_test") + assert t_id is not None + + target = await repo.get_target_by_id(t_id) + assert target.dispatch_order == "order" + + # Toggle to random + await repo.update_target_dispatch_order(t_id, "random") + target_updated = await repo.get_target_by_id(t_id) + assert target_updated.dispatch_order == "random" + + # 2. Redis queue FIFO vs Random pop test + queue = RedisQueue() + await queue.connect() + + # Clear queue for test target + key = queue._get_target_key(t_id) + await queue.client.delete(key) + + # Push 3 items: Item 1, Item 2, Item 3 + await queue.push_target_post(t_id, {"post_id": 1, "text": "Post 1"}) + await queue.push_target_post(t_id, {"post_id": 2, "text": "Post 2"}) + await queue.push_target_post(t_id, {"post_id": 3, "text": "Post 3"}) + + assert await queue.get_target_queue_size(t_id) == 3 + + # Random pop should retrieve one of the 3 items and leave 2 + popped_random = await queue.pop_target_post(t_id, dispatch_order="random") + assert popped_random is not None + assert popped_random["post_id"] in (1, 2, 3) + assert await queue.get_target_queue_size(t_id) == 2 + + # Clean up + await queue.client.delete(key) + await repo.delete_target(t_id) + await queue.close() + + +async def test_source_websites_crud_and_collector(): + await init_db() + repo = Repository() + + # 1. Add Source Website + test_url = f"https://example.com/blog-{int(time.time())}" + site_id = await repo.add_source_website( + name="Example Tech Blog", + url=test_url, + check_interval_min=15, + auto_reanalyze_hours=12, + api_config={ + "endpoint_url": f"{test_url}/wp-json/wp/v2/posts", + "parser_type": "wordpress_json", + "field_mappings": { + "title": "title.rendered", + "content": "content.rendered", + "link": "link" + } + } + ) + assert site_id is not None + + site = await repo.get_source_website_by_id(site_id) + assert site.name == "Example Tech Blog" + assert site.check_interval_min == 15 + assert site.auto_reanalyze_hours == 12 + assert site.api_config.get("parser_type") == "wordpress_json" + + # 2. Update intervals and status + await repo.update_source_website_interval(site_id, 45) + await repo.update_source_website_reanalyze_hours(site_id, 48) + await repo.update_source_website_fetch_status(site_id, error="Connection timeout test") + + site_err = await repo.get_source_website_by_id(site_id) + assert site_err.check_interval_min == 45 + assert site_err.auto_reanalyze_hours == 48 + assert site_err.last_error == "Connection timeout test" + + # 3. Clean up error + await repo.update_source_website_fetch_status(site_id, error=None) + site_ok = await repo.get_source_website_by_id(site_id) + assert site_ok.last_error is None + assert site_ok.last_fetched_at is not None + + # Clean up + await repo.delete_source_website(site_id) + + +async def test_website_analyzer_and_collector_mock(): + # Helper functions test + assert _clean_html("

Hello World!

") == "Hello World!" + sample_dict = {"channel": {"items": [{"title": "News 1"}]}} + assert _extract_nested(sample_dict, "channel.items.0.title") == "News 1" + + # Website Analyzer mock test + llm_mock = MagicMock() + llm_mock.generate_json = AsyncMock(return_value={ + "status": "success", + "endpoint_url": "https://digiato.com/wp-json/wp/v2/posts?per_page=10", + "parser_type": "wordpress_json", + "field_mappings": { + "title": "title.rendered", + "content": "content.rendered", + "link": "link" + }, + "summary": "سایت از وردپرس استفاده می‌کند." + }) + + analyzer = WebsiteAnalyzer(llm=llm_mock) + with patch("httpx.AsyncClient.get") as mock_get: + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.text = '' + mock_get.return_value = mock_resp + + ok, cfg, summary = await analyzer.analyze_website("https://digiato.com") + assert ok is True + assert cfg["parser_type"] == "wordpress_json" + assert cfg["endpoint_url"] == "https://digiato.com/wp-json/wp/v2/posts?per_page=10" + + +async def main(): + await test_target_dispatch_order_and_queue() + await test_source_websites_crud_and_collector() + await test_website_analyzer_and_collector_mock() + print("All Source Websites, Dispatch Order, and Markdown styling tests passed successfully!") + + +if __name__ == "__main__": + asyncio.run(main())