diff --git a/db/database.py b/db/database.py index 89bb41a..fd24cb5 100644 --- a/db/database.py +++ b/db/database.py @@ -152,17 +152,35 @@ CREATE INDEX IF NOT EXISTS idx_source_websites_active ON source_websites(is_acti CREATE INDEX IF NOT EXISTS idx_ai_providers_active ON ai_providers(is_active); +CREATE TABLE IF NOT EXISTS admin_channels ( + id SERIAL PRIMARY KEY, + channel_id BIGINT UNIQUE NOT NULL, + title VARCHAR(255), + username VARCHAR(255), + is_default BOOLEAN DEFAULT FALSE, + is_active BOOLEAN DEFAULT TRUE, + created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_admin_channels_active ON admin_channels(is_active); + -- Migration safety for existing tables ALTER TABLE ai_providers ADD COLUMN IF NOT EXISTS fallback_provider_id BIGINT REFERENCES ai_providers(id) ON DELETE SET NULL; ALTER TABLE ai_providers ADD COLUMN IF NOT EXISTS supports_vision BOOLEAN DEFAULT FALSE; ALTER TABLE sources ADD COLUMN IF NOT EXISTS category_id INT REFERENCES channel_categories(id) ON DELETE SET NULL; ALTER TABLE targets ADD COLUMN IF NOT EXISTS category_id INT REFERENCES channel_categories(id) ON DELETE SET NULL; +ALTER TABLE sources ADD COLUMN IF NOT EXISTS admin_channel_id BIGINT; +ALTER TABLE source_websites ADD COLUMN IF NOT EXISTS admin_channel_id BIGINT; +ALTER TABLE posts ADD COLUMN IF NOT EXISTS review_channel_id BIGINT; ALTER TABLE sources ADD COLUMN IF NOT EXISTS is_active BOOLEAN DEFAULT TRUE; ALTER TABLE targets ADD COLUMN IF NOT EXISTS is_active BOOLEAN DEFAULT TRUE; ALTER TABLE sources ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP; ALTER TABLE targets ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP; CREATE INDEX IF NOT EXISTS idx_sources_category ON sources(category_id); CREATE INDEX IF NOT EXISTS idx_targets_category ON targets(category_id); +CREATE INDEX IF NOT EXISTS idx_sources_admin_channel ON sources(admin_channel_id); +CREATE INDEX IF NOT EXISTS idx_source_websites_admin_channel ON source_websites(admin_channel_id); +CREATE INDEX IF NOT EXISTS idx_posts_review_channel ON posts(review_channel_id); ALTER TABLE targets ADD COLUMN IF NOT EXISTS personality TEXT DEFAULT ''; ALTER TABLE targets ADD COLUMN IF NOT EXISTS custom_footer TEXT DEFAULT ''; ALTER TABLE targets ADD COLUMN IF NOT EXISTS sleep_start_hour INT DEFAULT 0; diff --git a/db/models.py b/db/models.py index f998933..14b8ba1 100644 --- a/db/models.py +++ b/db/models.py @@ -1,6 +1,16 @@ from dataclasses import dataclass, field from typing import Optional, List, Dict, Any +@dataclass +class AdminReviewChannel: + id: Optional[int] + channel_id: int + title: Optional[str] + username: Optional[str] = None + is_default: bool = False + is_active: bool = True + created_at: Optional[str] = None + @dataclass class ChannelCategory: id: Optional[int] @@ -16,6 +26,7 @@ class SourceChannel: username: Optional[str] title: Optional[str] category_id: Optional[int] = None + admin_channel_id: Optional[int] = None context_message_count: int = 0 is_active: bool = True created_at: Optional[str] = None @@ -26,6 +37,7 @@ class SourceWebsite: name: str url: str category_id: Optional[int] = None + admin_channel_id: Optional[int] = None check_interval_min: int = 30 auto_reanalyze_hours: int = 24 last_reanalyzed_at: Optional[str] = None @@ -82,6 +94,7 @@ class Post: rejection_reason: Optional[str] = None published_to: List[Dict[str, Any]] = field(default_factory=list) is_deleted: bool = False + review_channel_id: Optional[int] = None review_message_id: Optional[int] = None scheduled_at: Optional[str] = None published_at: Optional[str] = None diff --git a/db/repository.py b/db/repository.py index 59cbbd1..371ced3 100644 --- a/db/repository.py +++ b/db/repository.py @@ -1,11 +1,17 @@ import json import asyncpg from datetime import datetime, timezone -from typing import List, Optional, Dict, Any, Tuple -from db.models import SourceChannel, SourceWebsite, TargetChannel, Post, Setting, AILog, AIProviderProfile, ChannelCategory +from db.models import ( + SourceChannel, SourceWebsite, TargetChannel, Post, Setting, + AILog, AIProviderProfile, ChannelCategory, AdminReviewChannel +) from db.database import get_db_pool +def _parse_admin_channel_row(row: asyncpg.Record) -> AdminReviewChannel: + return AdminReviewChannel(**dict(row)) + + def _parse_post_row(row: asyncpg.Record) -> Post: data = dict(row) if isinstance(data.get("published_to"), str): @@ -566,13 +572,19 @@ class Repository: ) return [_parse_post_row(r) for r in rows] - async def update_review_message_id(self, post_id: int, review_message_id: int) -> None: + async def update_review_message_id(self, post_id: int, review_message_id: Optional[int], review_channel_id: Optional[int] = None) -> None: pool = await self._get_pool() async with pool.acquire() as conn: - await conn.execute( - "UPDATE posts SET review_message_id = $1 WHERE id = $2;", - review_message_id, post_id, - ) + if review_channel_id is not None: + await conn.execute( + "UPDATE posts SET review_message_id = $1, review_channel_id = $2 WHERE id = $3;", + review_message_id, review_channel_id, post_id, + ) + else: + await conn.execute( + "UPDATE posts SET review_message_id = $1 WHERE id = $2;", + review_message_id, post_id, + ) async def _upsert_published_entry( self, @@ -1044,6 +1056,184 @@ class Repository: description="Global emergency operational pause" ) + # --- Admin Review Channels --- + async def add_admin_channel( + self, + channel_id: int, + title: Optional[str] = None, + username: Optional[str] = None, + is_default: bool = False + ) -> int: + pool = await self._get_pool() + async with pool.acquire() as conn: + async with conn.transaction(): + if is_default: + await conn.execute("UPDATE admin_channels SET is_default = FALSE;") + row = await conn.fetchrow( + """ + INSERT INTO admin_channels (channel_id, title, username, is_default, is_active) + VALUES ($1, $2, $3, $4, TRUE) + ON CONFLICT(channel_id) DO UPDATE SET + title = COALESCE(EXCLUDED.title, admin_channels.title), + username = COALESCE(EXCLUDED.username, admin_channels.username), + is_default = EXCLUDED.is_default, + is_active = TRUE + RETURNING id; + """, + channel_id, title, username, is_default + ) + return row["id"] + + async def get_admin_channels(self) -> List[AdminReviewChannel]: + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch("SELECT * FROM admin_channels WHERE is_active = TRUE ORDER BY is_default DESC, id ASC;") + return [_parse_admin_channel_row(r) for r in rows] + + async def get_admin_channel_by_id(self, admin_id: int) -> Optional[AdminReviewChannel]: + pool = await self._get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT * FROM admin_channels WHERE id = $1;", admin_id) + return _parse_admin_channel_row(row) if row else None + + async def get_admin_channel_by_telegram_id(self, channel_id: int) -> Optional[AdminReviewChannel]: + pool = await self._get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT * FROM admin_channels WHERE channel_id = $1;", channel_id) + return _parse_admin_channel_row(row) if row else None + + async def get_default_admin_channel(self) -> Optional[AdminReviewChannel]: + pool = await self._get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT * FROM admin_channels WHERE is_active = TRUE AND is_default = TRUE LIMIT 1;") + if not row: + row = await conn.fetchrow("SELECT * FROM admin_channels WHERE is_active = TRUE ORDER BY id ASC LIMIT 1;") + return _parse_admin_channel_row(row) if row else None + + async def set_default_admin_channel(self, admin_id: int) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + async with conn.transaction(): + await conn.execute("UPDATE admin_channels SET is_default = FALSE;") + await conn.execute("UPDATE admin_channels SET is_default = TRUE WHERE id = $1;", admin_id) + + async def delete_admin_channel(self, admin_id: int) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT channel_id FROM admin_channels WHERE id = $1;", admin_id) + if row: + ch_id = row["channel_id"] + await conn.execute("UPDATE sources SET admin_channel_id = NULL WHERE admin_channel_id = $1;", ch_id) + await conn.execute("UPDATE source_websites SET admin_channel_id = NULL WHERE admin_channel_id = $1;", ch_id) + await conn.execute("DELETE FROM admin_channels WHERE id = $1;", admin_id) + + # --- Source & Website Admin Channel Setters --- + async def set_source_admin_channel(self, source_id: int, admin_channel_id: Optional[int]) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute("UPDATE sources SET admin_channel_id = $1 WHERE id = $2;", admin_channel_id, source_id) + + async def set_website_admin_channel(self, site_id: int, admin_channel_id: Optional[int]) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute("UPDATE source_websites SET admin_channel_id = $1 WHERE id = $2;", admin_channel_id, site_id) + + # --- Review Purge Helpers --- + async def get_review_posts_by_admin_channel(self, admin_channel_id: int) -> List[Post]: + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT * FROM posts + WHERE review_channel_id = $1 + AND review_message_id IS NOT NULL + AND is_deleted = FALSE + ORDER BY id ASC; + """, + admin_channel_id + ) + return [_parse_post_row(r) for r in rows] + + async def count_review_posts_by_admin_channel(self, admin_channel_id: int) -> int: + pool = await self._get_pool() + async with pool.acquire() as conn: + val = await conn.fetchval( + """ + SELECT COUNT(*) FROM posts + WHERE review_channel_id = $1 + AND review_message_id IS NOT NULL + AND is_deleted = FALSE; + """, + admin_channel_id + ) + return val or 0 + + async def clear_review_messages_for_channel(self, admin_channel_id: int) -> List[Post]: + pool = await self._get_pool() + async with pool.acquire() as conn: + async with conn.transaction(): + rows = await conn.fetch( + """ + SELECT * FROM posts + WHERE review_channel_id = $1 + AND review_message_id IS NOT NULL + AND is_deleted = FALSE; + """, + admin_channel_id + ) + posts = [_parse_post_row(r) for r in rows] + await conn.execute( + """ + UPDATE posts + SET review_message_id = NULL, + is_deleted = TRUE, + status = 'deleted' + WHERE review_channel_id = $1 + AND review_message_id IS NOT NULL + AND is_deleted = FALSE; + """, + admin_channel_id + ) + return posts + + async def get_all_review_posts_with_message_id(self) -> List[Post]: + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT * FROM posts + WHERE review_message_id IS NOT NULL + AND is_deleted = FALSE + ORDER BY id ASC; + """ + ) + return [_parse_post_row(r) for r in rows] + + async def clear_all_review_messages(self) -> List[Post]: + pool = await self._get_pool() + async with pool.acquire() as conn: + async with conn.transaction(): + rows = await conn.fetch( + """ + SELECT * FROM posts + WHERE review_message_id IS NOT NULL + AND is_deleted = FALSE; + """ + ) + posts = [_parse_post_row(r) for r in rows] + await conn.execute( + """ + UPDATE posts + SET review_message_id = NULL, + is_deleted = TRUE, + status = 'deleted' + WHERE review_message_id IS NOT NULL + AND is_deleted = FALSE; + """ + ) + return posts + + diff --git a/services/admin_bot.py b/services/admin_bot.py index 682e874..c348cfe 100644 --- a/services/admin_bot.py +++ b/services/admin_bot.py @@ -69,7 +69,8 @@ def get_source_fetch_buttons(channel_id: int, source_id: Optional[int] = None): ] if source_id is not None: rows.append([ - Button.inline("📜 تنظیم پیام‌های زمینه (Context)", data=f"src_ctx_menu:{source_id}"), + Button.inline("📜 پیام‌های زمینه (Context)", data=f"src_ctx_menu:{source_id}"), + Button.inline("📋 کانال نظارت ادمین", data=f"src_adm:{source_id}"), ]) rows.append([ Button.inline("📁 تعیین دسته‌بندی", data=f"src_cat:{source_id}"), @@ -663,9 +664,31 @@ class AdminBotService: ) return caption + async def get_effective_review_channel_id(self, post: Post) -> Optional[int]: + """Determine which Admin Review Channel this post should be sent to.""" + if post.source_channel_id <= -900000000: + site_id = abs(post.source_channel_id + 900000000) + site = await self.repo.get_source_website_by_id(site_id) + if site and site.admin_channel_id: + return site.admin_channel_id + else: + src = await self.repo.get_source_by_channel_id(post.source_channel_id) + if src and src.admin_channel_id: + return src.admin_channel_id + + default_adm = await self.repo.get_default_admin_channel() + if default_adm and default_adm.channel_id: + return default_adm.channel_id + + return self.review_channel_id or None + async def send_raw_review_post(self, post_id: int, refresh_only: bool = False): post = await self.repo.get_post_by_id(post_id) - if not post or not self.review_channel_id or post.is_deleted: + if not post or post.is_deleted: + return + + target_review_channel_id = post.review_channel_id or await self.get_effective_review_channel_id(post) + if not target_review_channel_id: return targets = await self.repo.get_active_targets() @@ -678,7 +701,7 @@ class AdminBotService: return try: await self.client.edit_message( - self.review_channel_id, post.review_message_id, + target_review_channel_id, post.review_message_id, clamp_for_telegram(caption, bool(post.media_path)), parse_mode="html", buttons=keyboard, ) @@ -692,7 +715,7 @@ class AdminBotService: try: if has_media: msg = await self.client.send_file( - self.review_channel_id, + target_review_channel_id, file=post.media_path, caption=caption, parse_mode="html", @@ -700,14 +723,14 @@ class AdminBotService: ) else: msg = await self.client.send_message( - self.review_channel_id, + target_review_channel_id, caption, parse_mode="html", buttons=keyboard ) - await self.repo.update_review_message_id(post.id, msg.id) + await self.repo.update_review_message_id(post.id, msg.id, review_channel_id=target_review_channel_id) except Exception as e: - await log_exception("admin_bot.send_review_post", e, {"post_id": post.id, "review_channel_id": self.review_channel_id}) + await log_exception("admin_bot.send_review_post", e, {"post_id": post.id, "review_channel_id": target_review_channel_id}) async def refresh_error_metrics(self) -> int: """Republish the open-error gauges from the database. Returns the open count.""" @@ -954,6 +977,14 @@ class AdminBotService: if cat: cat_name = cat.name + adm_name = "پیش‌فرض سیستم" + if getattr(source, "admin_channel_id", None): + adm = await self.repo.get_admin_channel_by_telegram_id(source.admin_channel_id) + if adm: + adm_name = f"{adm.title or adm.channel_id}" + else: + adm_name = f"{source.admin_channel_id}" + auto_targets = await self.repo.get_targets_auto_routed_from(source.channel_id) auto_line = ( "🤖 ارسال خودکار به: " + "، ".join(f"{t.title}" for t in auto_targets) @@ -968,6 +999,7 @@ class AdminBotService: f"• 🆔 شناسه کانال: {source.channel_id}\n" f"• 🔗 یوزرنیم: @{source.username or 'ندارد'}\n" f"• 📁 دسته‌بندی: {cat_name}\n" + f"• 📋 کانال نظارت ادمین: {adm_name}\n" f"• 📜 پیام‌های زمینه (Context): {ctx_label}\n" f"• 📥 پست‌های دریافت‌شده: {collected}\n" f"• {auto_line}\n\n" @@ -1053,12 +1085,21 @@ class AdminBotService: last_re = site.last_reanalyzed_at.strftime("%Y-%m-%d %H:%M") if site.last_reanalyzed_at else "انجام نشده" needs = getattr(site, "custom_instructions", "") or "" + adm_name = "پیش‌فرض سیستم" + if getattr(site, "admin_channel_id", None): + adm = await self.repo.get_admin_channel_by_telegram_id(site.admin_channel_id) + if adm: + adm_name = f"{adm.title or adm.channel_id}" + else: + adm_name = f"{site.admin_channel_id}" + text = ( f"🌐 وبسایت مبدا: {site.name}\n\n" f"• 🔗 آدرس وبسایت: {site.url}\n" f"• 🔌 اندپوینت دریافت: {endpoint}\n" f"• 🏷 نوع ساختار: {parser_type}\n" f"• 📁 دسته‌بندی: {cat_name}\n" + f"• 📋 کانال نظارت ادمین: {adm_name}\n" f"• 📝 نیازمندی‌ها و فیلترهای استخراج:\n" f"{needs or 'ثبت نشده (استخراج تمام اخبار و پست‌ها)'}\n\n" f"• ⏱ فاصله بررسی: هر {site.check_interval_min} دقیقه\n" @@ -1078,11 +1119,14 @@ class AdminBotService: Button.inline("⏱ تغییر فاصله بررسی", data=f"web_intv:{site.id}") ], [ - Button.inline("🔄 دوره تحلیل مجدد AI", data=f"web_reintv:{site.id}"), + Button.inline("📋 کانال نظارت ادمین", data=f"web_adm:{site.id}"), Button.inline("📁 تعیین دسته‌بندی", data=f"web_cat:{site.id}") ], [ - Button.inline("🗑 حذف این وبسایت", data=f"del_web:{site.id}"), + Button.inline("🔄 دوره تحلیل مجدد AI", data=f"web_reintv:{site.id}"), + Button.inline("🗑 حذف این وبسایت", data=f"del_web:{site.id}") + ], + [ Button.inline("🔙 بازگشت به لیست وبسایت‌ها", data="list_web") ] ] @@ -1144,7 +1188,10 @@ class AdminBotService: Button.inline("➕ افزودن مبدا / مقصد جدید", data="hub_add"), Button.inline("📂 دسته‌بندی موضوعی کانال‌ها", data="hub_cats") ], - [Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main")] + [ + Button.inline("🗑 پاکسازی پیام‌های کانال‌های ادمین", data="purge_adm_menu"), + Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main") + ] ] return text, buttons @@ -1152,11 +1199,13 @@ class AdminBotService: sources = await self.repo.get_active_sources() targets = await self.repo.get_active_targets() websites = await self.repo.get_active_source_websites() + adm_channels = await self.repo.get_admin_channels() text = ( f"🤖 بخش کانال‌ها، وبسایت‌ها و ربات‌ها (Bots Hub):\n\n" f"• 📡 کانال‌های مبدا: {len(sources)} کانال\n" f"• 🎯 کانال‌های مقصد: {len(targets)} کانال\n" - f"• 🌐 وبسایت‌های مبدا: {len(websites)} وبسایت\n\n" + f"• 🌐 وبسایت‌های مبدا: {len(websites)} وبسایت\n" + f"• 📋 کانال‌های نظارت ادمین: {len(adm_channels)} کانال ثبت‌شده\n\n" f"بخش مورد نظر را برای مشاهده و مدیریت انتخاب کنید:" ) buttons = [ @@ -1173,10 +1222,13 @@ class AdminBotService: Button.inline("➕ افزودن وبسایت", data="add_web") ], [ - Button.inline("🔑 احراز هویت یوزربات", data="hub_req_code"), + Button.inline("📋 کانال‌های نظارت ادمین", data="list_adm_channels"), Button.inline("📂 مدیریت دسته‌بندی‌ها", data="list_cat") ], - [Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main")] + [ + Button.inline("🔑 احراز هویت یوزربات", data="hub_req_code"), + Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main") + ] ] return text, buttons @@ -1194,13 +1246,198 @@ class AdminBotService: buttons = [ [Button.inline(pause_btn_text, data=pause_cb)], [ - Button.inline("📝 گزارش تغییرات (Release Notes)", data="sys_changes"), - Button.inline("❓ راهنمای سیستم", data="sys_help") + Button.inline("🗑 پاکسازی پیام‌های کانال‌های ادمین", data="purge_adm_menu"), + Button.inline("📝 گزارش تغییرات", data="sys_changes") ], - [Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main")] + [ + Button.inline("❓ راهنمای سیستم", data="sys_help"), + Button.inline("🔙 بازگشت به منوی اصلی", data="back_to_main") + ] ] return text, buttons + async def _render_source_admin_channel_menu(self, source_id: int) -> Tuple[str, List[List[Button]]]: + source = await self.repo.get_source_by_id(source_id) + if not source: + return "❌ کانال مبدا یافت نشد.", [] + + adm_channels = await self.repo.get_admin_channels() + text = ( + f"📋 تعیین کانال نظارت ادمین برای کانال مبدا:\n" + f"📢 {source.title or source.channel_id}\n\n" + f"پست‌های جدید این کانال مبدا برای بررسی به کدام کانال مدیر/نظارت ارسال شوند؟\n\n" + f"یکی از کانال‌های زیر را انتخاب کنید یا گزینه پیش‌فرض را بزنید:" + ) + buttons = [] + for a in adm_channels: + mark = "✅ " if source.admin_channel_id == a.channel_id else "" + def_tag = " (پیش‌فرض)" if a.is_default else "" + buttons.append([Button.inline(f"{mark}📋 {a.title or a.channel_id}{def_tag}", data=f"src_set_adm:{source.id}:{a.channel_id}")]) + + is_def = (source.admin_channel_id is None) + def_mark = "✅ " if is_def else "" + buttons.append([Button.inline(f"{def_mark}⚙️ پیش‌فرض سیستم ({self.review_channel_id})", data=f"src_set_adm:{source.id}:none")]) + buttons.append([Button.inline("➕ افزودن کانال ادمین جدید", data=f"src_add_adm:{source.id}")]) + buttons.append([Button.inline("🔙 بازگشت به تنظیمات کانال مبدا", data=f"src_view:{source.id}")]) + return text, buttons + + async def _render_website_admin_channel_menu(self, site_id: int) -> Tuple[str, List[List[Button]]]: + site = await self.repo.get_source_website_by_id(site_id) + if not site: + return "❌ وبسایت یافت نشد.", [] + + adm_channels = await self.repo.get_admin_channels() + text = ( + f"📋 تعیین کانال نظارت ادمین برای وبسایت مبدا:\n" + f"🌐 {site.name}\n\n" + f"پست‌های استخراج‌شده از این وبسایت برای بررسی به کدام کانال مدیر/نظارت ارسال شوند؟\n\n" + f"یکی از کانال‌های زیر را انتخاب کنید یا گزینه پیش‌فرض را بزنید:" + ) + buttons = [] + for a in adm_channels: + mark = "✅ " if site.admin_channel_id == a.channel_id else "" + def_tag = " (پیش‌فرض)" if a.is_default else "" + buttons.append([Button.inline(f"{mark}📋 {a.title or a.channel_id}{def_tag}", data=f"web_set_adm:{site.id}:{a.channel_id}")]) + + is_def = (site.admin_channel_id is None) + def_mark = "✅ " if is_def else "" + buttons.append([Button.inline(f"{def_mark}⚙️ پیش‌فرض سیستم ({self.review_channel_id})", data=f"web_set_adm:{site.id}:none")]) + buttons.append([Button.inline("➕ افزودن کانال ادمین جدید", data=f"web_add_adm:{site.id}")]) + buttons.append([Button.inline("🔙 بازگشت به تنظیمات وبسایت", data=f"web_view:{site.id}")]) + return text, buttons + + async def _render_admin_channels_menu(self) -> Tuple[str, List[List[Button]]]: + channels = await self.repo.get_admin_channels() + lines = [ + "📋 مدیریت کانال‌های نظارت و بازبینی ادمین (Admin Review Channels)\n", + "شما می‌توانید چندین کانال ادمین مجزا تعریف کنید تا پست‌های منابع مختلف به کانال‌های مدیر متفاوتی هدایت شوند:\n" + ] + buttons = [] + if channels: + for ch in channels: + def_star = " ⭐️ (پیش‌فرض)" if ch.is_default else "" + lines.append(f"• {ch.title or 'کانال ادمین'} ({ch.channel_id}){def_star}") + buttons.append([ + Button.inline(f"📋 {ch.title or ch.channel_id}{def_star}", data=f"adm_view:{ch.id}"), + ]) + else: + lines.append(f"تنها کانال پیش‌فرض محیطی فعال است: {self.review_channel_id}") + + buttons.append([Button.inline("➕ افزودن کانال ادمین جدید", data="add_adm_ch")]) + buttons.append([Button.inline("🔙 بازگشت به منوی ربات‌ها", data="hub_bots")]) + return "\n".join(lines), buttons + + async def _render_admin_channel_detail(self, admin_id: int) -> Tuple[str, List[List[Button]]]: + adm = await self.repo.get_admin_channel_by_id(admin_id) + if not adm: + return "❌ کانال ادمین یافت نشد.", [[Button.inline("🔙 بازگشت", data="list_adm_channels")]] + + pending_count = await self.repo.count_review_posts_by_admin_channel(adm.channel_id) + def_line = "⭐️ وضعیت: کانال پیش‌فرض سیستم" if adm.is_default else "⚪️ وضعیت: کانال فرعی" + + text = ( + f"📋 مشخصات کانال نظارت ادمین: {adm.title or adm.channel_id}\n\n" + f"• 🆔 شناسه عددی: {adm.channel_id}\n" + f"• 🔗 یوزرنیم: @{adm.username or 'ندارد'}\n" + f"• {def_line}\n" + f"• 📨 پیام‌های بررسی‌نشده در این کانال: {pending_count} پیام\n\n" + f"عملیات مورد نظر را انتخاب کنید:" + ) + buttons = [] + if not adm.is_default: + buttons.append([Button.inline("⭐️ تعیین به‌عنوان پیش‌فرض سیستم", data=f"adm_set_def:{adm.id}")]) + buttons.append([Button.inline(f"🗑 پاکسازی پیام‌های این کانال ({pending_count})", data=f"purge_adm_ask:{adm.channel_id}")]) + buttons.append([Button.inline("❌ حذف این کانال ادمین", data=f"adm_del:{adm.id}")]) + buttons.append([Button.inline("🔙 بازگشت به لیست کانال‌های ادمین", data="list_adm_channels")]) + return text, buttons + + async def _render_purge_admin_channels_menu(self) -> Tuple[str, List[List[Button]]]: + adm_channels = await self.repo.get_admin_channels() + all_channels_map: Dict[int, str] = {} + if self.review_channel_id: + all_channels_map[self.review_channel_id] = "کانال پیش‌فرض محیطی" + + for a in adm_channels: + all_channels_map[a.channel_id] = a.title or str(a.channel_id) + + lines = [ + "🗑 پاکسازی پیام‌های کانال‌های ادمین (Purge Review Messages)\n", + "با استفاده از این بخش می‌توانید با یک دکمه، تمام کارت‌ها و پیام‌های ارسال‌شده به کانال‌های مدیران را به‌صورت تفکیک‌شده یا یکجا پاکسازی کنید:\n" + ] + buttons = [] + total_pending = 0 + + for ch_id, title in all_channels_map.items(): + cnt = await self.repo.count_review_posts_by_admin_channel(ch_id) + total_pending += cnt + lines.append(f"• 📋 {title} ({ch_id}): {cnt} پیام بررسی‌نشده") + buttons.append([ + Button.inline(f"🗑 پاکسازی پیام‌های {title[:20]} ({cnt})", data=f"purge_adm_ask:{ch_id}") + ]) + + if total_pending > 0 or len(all_channels_map) > 1: + buttons.append([Button.inline(f"🔥 پاکسازی همه کانال‌های ادمین (مجموع {total_pending})", data="purge_adm_ask:all")]) + + buttons.append([Button.inline("🔙 بازگشت به منوی سیستم", data="hub_system")]) + return "\n".join(lines), buttons + + async def purge_review_messages(self, channel_id_or_all: str) -> Tuple[int, int, str]: + """ + Deletes review messages from the specified admin channel (or all admin channels). + Returns: (deleted_tg_count, cleared_db_count, summary_message) + """ + deleted_tg_count = 0 + cleared_db_count = 0 + + if channel_id_or_all == "all": + posts = await self.repo.clear_all_review_messages() + cleared_db_count = len(posts) + by_channel: Dict[int, List[int]] = {} + for p in posts: + if p.media_path and os.path.exists(p.media_path): + try: + os.remove(p.media_path) + except Exception: + pass + ch_id = p.review_channel_id or self.review_channel_id + if ch_id and p.review_message_id: + by_channel.setdefault(ch_id, []).append(p.review_message_id) + + for ch_id, msg_ids in by_channel.items(): + for chunk_idx in range(0, len(msg_ids), 100): + chunk = msg_ids[chunk_idx:chunk_idx+100] + try: + await self.client.delete_messages(ch_id, chunk) + deleted_tg_count += len(chunk) + except Exception as e: + logger.warning(f"Error deleting chunk from channel {ch_id}: {e}") + summary = f"🔥 تمام پیام‌های کانال‌های نظارت پاکسازی شدند:\n\n• 🗑 پیام‌های حذف‌شده از تلگرام: {deleted_tg_count}\n• 🗂 پست‌های بایگانی‌شده در سیستم: {cleared_db_count}" + else: + ch_id = int(channel_id_or_all) + posts = await self.repo.clear_review_messages_for_channel(ch_id) + cleared_db_count = len(posts) + msg_ids = [] + for p in posts: + if p.media_path and os.path.exists(p.media_path): + try: + os.remove(p.media_path) + except Exception: + pass + if p.review_message_id: + msg_ids.append(p.review_message_id) + + for chunk_idx in range(0, len(msg_ids), 100): + chunk = msg_ids[chunk_idx:chunk_idx+100] + try: + await self.client.delete_messages(ch_id, chunk) + deleted_tg_count += len(chunk) + except Exception as e: + logger.warning(f"Error deleting chunk from channel {ch_id}: {e}") + + summary = f"🗑 پیام‌های کانال نظارت {ch_id} پاکسازی شدند:\n\n• 🗑 پیام‌های حذف‌شده از تلگرام: {deleted_tg_count}\n• 🗂 پست‌های بایگانی‌شده در سیستم: {cleared_db_count}" + + return deleted_tg_count, cleared_db_count, summary + async def _render_monitor_hub(self) -> Tuple[str, List[List[Button]]]: rep = await get_instant_metrics_report() buttons = [ @@ -1597,6 +1834,22 @@ class AdminBotService: text, buttons = await self._render_categories_menu() await event.reply(text, parse_mode="html", buttons=buttons or None) + # --- Admin Review Channels Management --- + @self.client.on(events.NewMessage(pattern=r"(?i)^(/admin_channels|/review_channels|📋 کانال‌های نظارت|📋 کانال‌های نظارت ادمین)$")) + async def cmd_admin_channels(event: events.NewMessage.Event): + if not self.is_admin(event.sender_id): + return + text, buttons = await self._render_admin_channels_menu() + await event.reply(text, parse_mode="html", buttons=buttons or None) + + # --- Purge Review Channels --- + @self.client.on(events.NewMessage(pattern=r"(?i)^(/purge|/clear_reviews|🗑 پاکسازی کانال‌های ادمین|🗑 پاکسازی کانال‌های نظارت)$")) + async def cmd_purge_admin_channels(event: events.NewMessage.Event): + if not self.is_admin(event.sender_id): + return + text, buttons = await self._render_purge_admin_channels_menu() + await event.reply(text, parse_mode="html", buttons=buttons or None) + # --- Unified Add Hub --- @self.client.on(events.NewMessage(pattern=r"(?i)^(/add|➕ افزودن مبدا / مقصد)$")) async def cmd_add_hub(event: events.NewMessage.Event): @@ -1834,6 +2087,33 @@ class AdminBotService: ) await event.reply(card, parse_mode="html", buttons=buttons) + # 2.1. Waiting for Admin Review Channel Forward or Text + elif action == "wait_admin_channel_fwd": + ch_id, title, username, err = await self._resolve_channel(event) + if err or not ch_id: + await event.reply(err or "خطا در دریافت اطلاعات کانال نظارت. لطفا مجدد ارسال کنید:", buttons=get_cancel_button()) + return + + src_id = state.get("src_id") + site_id = state.get("site_id") + self.user_states.pop(event.sender_id, None) + adm_id = await self.repo.add_admin_channel(channel_id=ch_id, title=title, username=username) + + if src_id: + await self.repo.set_source_admin_channel(src_id, ch_id) + card, buttons = await self._render_source_config(src_id) + await event.reply(f"✅ کانال نظارت {title} با شناسه {ch_id} افزوده و به کانال مبدا اختصاص داده شد!", parse_mode="html") + await event.reply(card, parse_mode="html", buttons=buttons) + elif site_id: + await self.repo.set_website_admin_channel(site_id, ch_id) + card, buttons = await self._render_website_config(site_id) + await event.reply(f"✅ کانال نظارت {title} با شناسه {ch_id} افزوده و به وبسایت مبدا اختصاص داده شد!", parse_mode="html") + await event.reply(card, parse_mode="html", buttons=buttons) + else: + card, buttons = await self._render_admin_channel_detail(adm_id) + await event.reply(f"✅ کانال نظارت ادمین {title} با شناسه {ch_id} با موفقیت افزوده شد!", parse_mode="html") + await event.reply(card, parse_mode="html", buttons=buttons) + # 3. Waiting for Target Personality elif action == "wait_personality": target_id = state.get("target_id") @@ -3128,6 +3408,130 @@ class AdminBotService: await event.edit(card, parse_mode="html", buttons=buttons) await event.answer("✅ دسته‌بندی کانال با موفقیت به روز شد!") + # --- Admin Review Channels Management Callbacks --- + elif data == "list_adm_channels": + text, buttons = await self._render_admin_channels_menu() + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer() + + elif data == "add_adm_ch": + self.user_states[event.sender_id] = {"action": "wait_admin_channel_fwd"} + guide = ( + "📋 افزودن کانال نظارت و بازبینی ادمین:\n\n" + "ساده‌ترین روش: یک پیام از کانال نظارت مورد نظر به این ربات فوروارد (Forward) کنید!\n\n" + "(یا می‌توانید آیدی، یوزرنیم یا لینک کانال مثل @admin_review_channel را ارسال نمایید. مطمئن شوید ربات در آن کانال ادمین با دسترسی ارسال و حذف پیام است)" + ) + await event.reply(guide, parse_mode="html", buttons=get_cancel_button()) + await event.answer() + + elif data.startswith("adm_view:"): + adm_id = int(data.split(":")[1]) + text, buttons = await self._render_admin_channel_detail(adm_id) + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer() + + elif data.startswith("adm_set_def:"): + adm_id = int(data.split(":")[1]) + await self.repo.set_default_admin_channel(adm_id) + text, buttons = await self._render_admin_channel_detail(adm_id) + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer("⭐️ کانال به‌عنوان پیش‌فرض سیستم تنظیم شد!") + + elif data.startswith("adm_del:"): + adm_id = int(data.split(":")[1]) + await self.repo.delete_admin_channel(adm_id) + text, buttons = await self._render_admin_channels_menu() + await event.edit(f"🗑 کانال نظارت ادمین حذف شد.\n\n{text}", parse_mode="html", buttons=buttons) + await event.answer("کانال نظارت حذف شد.") + + # --- Source / Website Admin Channel Assignment Callbacks --- + elif data.startswith("src_adm:"): + src_id = int(data.split(":")[1]) + text, buttons = await self._render_source_admin_channel_menu(src_id) + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer() + + elif data.startswith("src_set_adm:"): + _, src_id_str, adm_ch_str = data.split(":") + src_id = int(src_id_str) + adm_ch = None if adm_ch_str == "none" else int(adm_ch_str) + await self.repo.set_source_admin_channel(src_id, adm_ch) + card, buttons = await self._render_source_config(src_id) + await event.edit(card, parse_mode="html", buttons=buttons) + await event.answer("✅ کانال نظارت برای این مبدا تنظیم شد!") + + elif data.startswith("src_add_adm:"): + src_id = int(data.split(":")[1]) + self.user_states[event.sender_id] = {"action": "wait_admin_channel_fwd", "src_id": src_id} + guide = ( + "📋 افزودن کانال نظارت جدید برای این مبدا:\n\n" + "یک پیام از کانال نظارت مورد نظر به ربات فوروارد کنید یا یوزرنیم/آیدی آن را بفرستید:" + ) + await event.reply(guide, parse_mode="html", buttons=get_cancel_button()) + await event.answer() + + elif data.startswith("web_adm:"): + site_id = int(data.split(":")[1]) + text, buttons = await self._render_website_admin_channel_menu(site_id) + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer() + + elif data.startswith("web_set_adm:"): + _, site_id_str, adm_ch_str = data.split(":") + site_id = int(site_id_str) + adm_ch = None if adm_ch_str == "none" else int(adm_ch_str) + await self.repo.set_website_admin_channel(site_id, adm_ch) + card, buttons = await self._render_website_config(site_id) + await event.edit(card, parse_mode="html", buttons=buttons) + await event.answer("✅ کانال نظارت برای این وبسایت تنظیم شد!") + + elif data.startswith("web_add_adm:"): + site_id = int(data.split(":")[1]) + self.user_states[event.sender_id] = {"action": "wait_admin_channel_fwd", "site_id": site_id} + guide = ( + "📋 افزودن کانال نظارت جدید برای این وبسایت:\n\n" + "یک پیام از کانال نظارت مورد نظر به ربات فوروارد کنید یا یوزرنیم/آیدی آن را بفرستید:" + ) + await event.reply(guide, parse_mode="html", buttons=get_cancel_button()) + await event.answer() + + # --- Purge Review Messages Callbacks --- + elif data == "purge_adm_menu": + text, buttons = await self._render_purge_admin_channels_menu() + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer() + + elif data.startswith("purge_adm_ask:"): + target_ch = data.split(":", 1)[1] + if target_ch == "all": + prompt_text = ( + "⚠️ هشدار مهم: پاکسازی تمام کانال‌های نظارت ادمین!\n\n" + "آیا از حذف کامل تمام کارت‌های بازبینی و پیام‌های ارسال‌شده در تمام کانال‌های نظارت مطمئن هستید؟\n\n" + "این عملیات پیام‌ها را از کانال‌های تلگرام حذف کرده و وضعیت پست‌ها را بایگانی می‌کند." + ) + else: + prompt_text = ( + f"⚠️ هشدار: پاکسازی پیام‌های کانال نظارت {target_ch}\n\n" + f"آیا از حذف تمام کارت‌های بازبینی ارسال‌شده به این کانال مطمئن هستید؟" + ) + + confirm_buttons = [ + [Button.inline("🗑 بله، پیام‌ها پاکسازی شوند", data=f"purge_adm_do:{target_ch}")], + [Button.inline("❌ انصراف و بازگشت", data="purge_adm_menu")] + ] + await event.edit(prompt_text, parse_mode="html", buttons=confirm_buttons) + await event.answer() + + elif data.startswith("purge_adm_do:"): + target_ch = data.split(":", 1)[1] + await event.edit("⏳ در حال پاکسازی پیام‌ها از تلگرام و پایگاه‌داده...", parse_mode="html", buttons=None) + _, _, summary = await self.purge_review_messages(target_ch) + back_buttons = [ + [Button.inline("🗑 بازگشت به منوی پاکسازی", data="purge_adm_menu")], + [Button.inline("🔙 منوی سیستم", data="hub_system")] + ] + await event.edit(summary, parse_mode="html", buttons=back_buttons) + await event.answer("✅ پاکسازی با موفقیت انجام شد!") # --- Review Channel Buttons --- elif data.startswith("hist:"):