diff --git a/bot/keyboards.py b/bot/keyboards.py deleted file mode 100644 index ee4c4fe..0000000 --- a/bot/keyboards.py +++ /dev/null @@ -1,18 +0,0 @@ -from typing import List -from telethon import Button -from db.models import TargetChannel - -def get_review_keyboard(post_id: int, targets: List[TargetChannel]) -> List[List[Button]]: - keyboard = [] - # Add a button for each target channel - for target in targets: - title = target.title or f"Target #{target.id}" - keyboard.append([ - Button.inline(f"🚀 Send to {title}", data=f"appr:{post_id}:{target.id}") - ]) - - # Add Reject button - keyboard.append([ - Button.inline("❌ Reject Post", data=f"rej:{post_id}") - ]) - return keyboard diff --git a/services/collector.py b/services/collector.py index 7b6e141..836b485 100644 --- a/services/collector.py +++ b/services/collector.py @@ -1,5 +1,6 @@ import os import logging +from dataclasses import dataclass from typing import Optional, Callable, Awaitable from telethon import TelegramClient, events from telethon.errors import SessionPasswordNeededError @@ -7,7 +8,7 @@ from telethon.tl.types import MessageMediaPhoto, MessageMediaDocument from db.repository import Repository from core.dedup import compute_content_hash, compute_file_hash from core.queue import RedisQueue -from core.metrics import COLLECTED_POSTS_TOTAL, SOURCE_ACTIVITY_TOTAL +from core.metrics import SOURCE_ACTIVITY_TOTAL, DUPLICATES_DETECTED_TOTAL from core.proxy import get_telegram_proxy from core.error_logger import log_exception @@ -16,6 +17,40 @@ logger = logging.getLogger(__name__) SESSION_DIR = os.getenv("SESSION_DIR", "/app/sessions" if os.path.exists("/app") else "/projects/telegram-bots/copykar/sessions") MEDIA_DIR = os.getenv("MEDIA_DIR", "/app/data/media" if os.path.exists("/app") else "/projects/telegram-bots/copykar/data/media") +# Bounds for the "custom count" prompt in the admin bot. +MIN_FETCH_LIMIT = 1 +MAX_FETCH_LIMIT = 500 + + +@dataclass +class ScrapeResult: + """Outcome of a history scrape. + + 'collected' alone is misleading: a repeat scrape of the same window legitimately + adds nothing, which reads as a broken button unless the other counters are shown. + """ + scanned: int = 0 + collected: int = 0 + already_stored: int = 0 + duplicates: int = 0 + error: Optional[str] = None + + def summary_fa(self, channel_id: int) -> str: + if self.error: + return f"❌ خطا در دریافت پست‌های کانال {channel_id}: {self.error}" + header = ( + f"✅ {self.collected} پست جدید از {channel_id} دریافت و به کانال ادمین ارسال شد." + if self.collected + else f"ℹ️ هیچ پست جدیدی در {channel_id} پیدا نشد." + ) + return ( + f"{header}\n\n" + f"• 🔍 پیام بررسی‌شده: {self.scanned}\n" + f"• 🆕 پست جدید: {self.collected}\n" + f"• 🗂 قبلا ذخیره شده: {self.already_stored}\n" + f"• ♻️ محتوای تکراری: {self.duplicates}" + ) + class CollectorService: def __init__( self, @@ -136,6 +171,10 @@ class CollectorService: async def _handle_message(self, event: events.NewMessage.Event): try: + if await self.repo.is_system_paused(): + logger.info(f"[collector] System is paused by admin. Ignoring incoming post from chat {event.chat_id}") + return + chat_id = event.chat_id source = await self.repo.get_source_by_channel_id(chat_id) if not source or not source.is_active: @@ -162,6 +201,7 @@ class CollectorService: media_hash = compute_file_hash(downloaded_file) content_hash = compute_content_hash(raw_text, media_hash) + duplicate_of = await self.repo.find_duplicate_post(content_hash) post_id = await self.repo.create_raw_post( source_channel_id=chat_id, @@ -170,10 +210,15 @@ class CollectorService: media_path=media_path, media_type=media_type, content_hash=content_hash, + is_duplicate=duplicate_of is not None, + duplicate_of_id=duplicate_of.id if duplicate_of else None, + similarity_reason=f"content hash matches post #{duplicate_of.id}" if duplicate_of else None, ) + if post_id and duplicate_of: + DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc() + if post_id: - COLLECTED_POSTS_TOTAL.labels(source_channel_id=str(chat_id)).inc() SOURCE_ACTIVITY_TOTAL.labels(channel_id=str(chat_id), title=source.title or 'Unknown').inc() logger.info(f"Collected raw post ID {post_id} from source channel {chat_id}") @@ -187,20 +232,21 @@ class CollectorService: channel_id: int, limit: int = 20, progress_callback: Optional[Callable[[str], Awaitable[None]]] = None - ) -> int: + ) -> ScrapeResult: """Scrape historical messages from a source channel.""" + result = ScrapeResult() + if not self.client.is_connected() or not await self.client.is_user_authorized(): if progress_callback: await progress_callback("❌ ربات متصل نیست. لطفا ابتدا لاگین کنید.") - return 0 - - source = await self.repo.get_source_by_channel_id(channel_id) - source_title = source.title if source else str(channel_id) - username = source.username if source else None - collected_count = 0 - skipped_count = 0 + result.error = "not connected" + return result try: + source = await self.repo.get_source_by_channel_id(channel_id) + source_title = source.title if source else str(channel_id) + username = source.username if source else None + entity = await self._resolve_channel_entity(channel_id, username) messages = [] async for msg in self.client.iter_messages(entity, limit=limit): @@ -208,6 +254,8 @@ class CollectorService: messages.reverse() + result.scanned = len(messages) + for message in messages: raw_text = message.raw_text or "" if not raw_text and not message.media: @@ -233,6 +281,10 @@ class CollectorService: media_hash = compute_file_hash(downloaded_file) content_hash = compute_content_hash(raw_text, media_hash) + duplicate_of = await self.repo.find_duplicate_post(content_hash) + if duplicate_of: + result.duplicates += 1 + DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc() post_id = await self.repo.create_raw_post( source_channel_id=channel_id, @@ -241,29 +293,30 @@ class CollectorService: media_path=media_path, media_type=media_type, content_hash=content_hash, + is_duplicate=duplicate_of is not None, + duplicate_of_id=duplicate_of.id if duplicate_of else None, + similarity_reason=f"content hash matches post #{duplicate_of.id}" if duplicate_of else None, ) if post_id: - collected_count += 1 - COLLECTED_POSTS_TOTAL.labels(source_channel_id=str(channel_id)).inc() + result.collected += 1 SOURCE_ACTIVITY_TOTAL.labels(channel_id=str(channel_id), title=source_title).inc() logger.info(f"Backfilled raw post ID {post_id} from {channel_id}") if self.on_post_received: await self.on_post_received(post_id) else: - skipped_count += 1 + result.already_stored += 1 if progress_callback: - await progress_callback( - f"✅ تعداد {collected_count} پست جدید از {channel_id} دریافت و در کانال ادمین قرار گرفت! (رد شده تکراری: {skipped_count})." - ) - return collected_count + await progress_callback(result.summary_fa(channel_id)) + return result except Exception as e: await log_exception("collector.scrape_history", e, {"channel_id": channel_id, "limit": limit}) + result.error = str(e) if progress_callback: - await progress_callback(f"❌ خطا در دریافت پست‌های کانال {channel_id}: {e}") - return collected_count + await progress_callback(result.summary_fa(channel_id)) + return result async def stop(self): if self.client.is_connected(): diff --git a/services/publisher.py b/services/publisher.py index 502d66e..a79fc93 100644 --- a/services/publisher.py +++ b/services/publisher.py @@ -1,13 +1,20 @@ import os import asyncio import logging -from datetime import datetime, timezone -from typing import Optional, List +from datetime import datetime, timedelta, timezone +from typing import Optional, List, Awaitable, Callable, Set, Tuple from telethon import TelegramClient +from telethon.errors import ( + ChannelPrivateError, + ChatAdminRequiredError, + ChatWriteForbiddenError, + PeerIdInvalidError, + UserBannedInChannelError, +) from db.models import TargetChannel from db.repository import Repository from core.queue import RedisQueue -from core.metrics import TARGET_ACTIVITY_TOTAL, QUEUE_POSTS_GAUGE, REDIS_QUEUE_SIZE_GAUGE +from core.metrics import TARGET_ACTIVITY_TOTAL, QUEUE_POSTS_GAUGE from core.proxy import get_telegram_proxy from core.error_logger import log_exception @@ -15,6 +22,19 @@ logger = logging.getLogger(__name__) SESSION_DIR = os.getenv("SESSION_DIR", "/app/sessions" if os.path.exists("/app") else "/projects/telegram-bots/copykar/sessions") +# Sleep windows are configured in the operator's wall-clock time, not UTC. +TIMEZONE_OFFSET_HOURS = float(os.getenv("TIMEZONE_OFFSET_HOURS", "3.5")) + +# Failures that will never resolve by retrying: the account simply cannot post there. +# Re-queueing these would spin the same post through the loop forever. +PERMANENT_DELIVERY_ERRORS = ( + ChatAdminRequiredError, + ChatWriteForbiddenError, + ChannelPrivateError, + UserBannedInChannelError, + PeerIdInvalidError, +) + class PublisherService: def __init__( self, @@ -25,10 +45,15 @@ class PublisherService: api_hash: Optional[str] = None, bot_token: Optional[str] = None, session_name: Optional[str] = None, + notify_fn: Optional[Callable[[str], Awaitable[None]]] = None, ): self.repo = repo self.queue = queue self.client = client + self.notify_fn = notify_fn + # (target_id, error type) pairs already reported, so a broken target is + # announced once instead of every polling cycle. + self._reported_failures: Set[Tuple[int, str]] = set() self.api_id = api_id or int(os.getenv("API_ID", "0")) self.api_hash = api_hash or os.getenv("API_HASH", "") self.bot_token = bot_token or os.getenv("BOT_TOKEN") @@ -42,11 +67,13 @@ class PublisherService: async def start(self): logger.info("Starting Paced Target Publisher Service...") if not self.client.is_connected(): + # Only owns the connection when it built its own client; a shared client + # (the admin bot's) is already connected by its owner. if self.bot_token: await self.client.start(bot_token=self.bot_token) else: await self.client.start() - logger.info("Paced Target Publisher Service connected.") + logger.info("Paced Target Publisher Service connected (delivering as the bot account).") self._running = True self._task = asyncio.create_task(self._publisher_loop()) @@ -66,7 +93,10 @@ class PublisherService: async def _publisher_loop(self): while self._running: try: - await self._process_all_target_queues() + if await self.repo.is_system_paused(): + logger.debug("[publisher] System is paused by admin. Skipping queue processing.") + else: + await self._process_all_target_queues() except Exception as e: logger.error(f"Error in target publisher loop: {e}", exc_info=True) await asyncio.sleep(15) @@ -74,13 +104,11 @@ class PublisherService: async def _process_all_target_queues(self): targets = await self.repo.get_active_targets() now = datetime.now(timezone.utc) - current_hour_local = (now.hour + 3) % 24 # UTC+3:30 approx hour - - total_queued = 0 + local_now = now + timedelta(hours=TIMEZONE_OFFSET_HOURS) + current_hour_local = local_now.hour for target in targets: qsize = await self.queue.get_target_queue_size(target.id) - total_queued += qsize QUEUE_POSTS_GAUGE.labels(status=f"target_{target.id}").set(qsize) if qsize == 0: @@ -130,9 +158,52 @@ class PublisherService: TARGET_ACTIVITY_TOTAL.labels(channel_id=str(target.channel_id), title=target.title or '').inc() logger.info(f"Published post ID {post_id} to Target {target.title} ({target.channel_id})") except Exception as e: - await log_exception("publisher.publish", e, {"post_id": post_id, "target_id": target.id, "channel_id": target.channel_id}) + permanent = isinstance(e, PERMANENT_DELIVERY_ERRORS) + if permanent: + # Dropping the payload is deliberate: the post stays 'pending_review' + # so an admin can re-send it once the permission problem is resolved. + await self._report_broken_target(target, e) + else: + # The payload was already popped; putting it back keeps the post from + # being silently lost on a transient Telegram failure. + try: + await self.queue.push_target_post(target.id, payload) + except Exception as requeue_err: + logger.critical(f"Failed to requeue post {post_id} for target {target.id}: {requeue_err}") - REDIS_QUEUE_SIZE_GAUGE.set(total_queued) + await log_exception("publisher.publish", e, { + "post_id": post_id, + "target_id": target.id, + "channel_id": target.channel_id, + "permanent": permanent, + }) + + async def _report_broken_target(self, target: TargetChannel, error: Exception) -> None: + """Tell the admins once that a target channel is unreachable for this account.""" + key = (target.id, type(error).__name__) + if key in self._reported_failures: + return + self._reported_failures.add(key) + + logger.error( + f"Target #{target.id} ({target.title}) rejected delivery permanently: " + f"{type(error).__name__}. Queue drained for this target until it is fixed." + ) + if not self.notify_fn: + return + + try: + await self.notify_fn( + f"⛔ ارسال به کانال مقصد «{target.title}» ممکن نیست!\n\n" + f"• 🆔 شناسه کانال: {target.channel_id}\n" + f"• ❗️ خطا: {type(error).__name__}\n\n" + "ربات در این کانال عضو یا ادمین با دسترسی ارسال پیام نیست.\n" + "لطفا ربات را در کانال ادمین کنید و دسترسی ارسال پیام (Post Messages) بدهید، " + "سپس پست را دوباره به صف بفرستید.\n\n" + "تا رفع این مشکل، پست‌های این کانال ارسال نمی‌شوند و در وضعیت بررسی باقی می‌مانند." + ) + except Exception as notify_err: + logger.error(f"Could not notify admins about broken target {target.id}: {notify_err}") async def stop(self): self._running = False diff --git a/services/queue_consumer.py b/services/queue_consumer.py deleted file mode 100644 index 1a9d0f0..0000000 --- a/services/queue_consumer.py +++ /dev/null @@ -1,50 +0,0 @@ -import os -import asyncio -import logging -from typing import Optional -from core.queue import RedisQueue -from core.metrics import QUEUE_POSTS_GAUGE, REDIS_QUEUE_SIZE_GAUGE - -logger = logging.getLogger(__name__) - -FETCH_INTERVAL_SECONDS = int(os.getenv("AI_PROCESSING_INTERVAL_SECONDS", "120")) - -class QueueConsumerService: - def __init__(self, queue: RedisQueue, on_post_popped = None, fetch_interval: int = FETCH_INTERVAL_SECONDS): - self.queue = queue - self.on_post_popped = on_post_popped - self.fetch_interval = fetch_interval - self._running = False - self._task: Optional[asyncio.Task] = None - - async def start(self): - logger.info(f"Starting Redis Queue Consumer Service (interval: {self.fetch_interval}s / {self.fetch_interval // 60}m)...") - await self.queue.connect() - self._running = True - self._task = asyncio.create_task(self._consumer_loop()) - - async def _consumer_loop(self): - while self._running: - try: - qsize = await self.queue.qsize() - QUEUE_POSTS_GAUGE.labels(status="redis_incoming").set(qsize) - REDIS_QUEUE_SIZE_GAUGE.set(qsize) - - if qsize > 0: - post_id = await self.queue.pop() - if post_id and self.on_post_popped: - logger.info(f"Dispatching raw post ID {post_id} to Admin Review channel (remaining in Redis: {qsize - 1})") - await self.on_post_popped(post_id) - new_qsize = await self.queue.qsize() - QUEUE_POSTS_GAUGE.labels(status="redis_incoming").set(new_qsize) - REDIS_QUEUE_SIZE_GAUGE.set(new_qsize) - except Exception as e: - logger.error(f"Error in queue consumer loop: {e}", exc_info=True) - - await asyncio.sleep(self.fetch_interval) - - async def stop(self): - self._running = False - if self._task: - self._task.cancel() - logger.info("Redis Queue Consumer Service stopped.") diff --git a/tests/test_auto_routing.py b/tests/test_auto_routing.py new file mode 100644 index 0000000..d5d42a2 --- /dev/null +++ b/tests/test_auto_routing.py @@ -0,0 +1,99 @@ +"""Auto-routing must queue to subscribed targets WITHOUT replacing the admin review card.""" +import asyncio +import sys + +sys.path.insert(0, "/app") + +from db.database import init_db, close_db_pool +from db.repository import Repository +from services.admin_bot import AdminBotService + +SRC_A = -1009999000020 +SRC_B = -1009999000021 +TRG = -1009999000022 + + +class FakeQueue: + def __init__(self): + self.pushed = [] + + async def push_target_post(self, target_id, payload): + self.pushed.append((target_id, payload)) + + async def get_target_queue_size(self, target_id): + return 0 + + +class FakeAI: + async def rewrite_for_target(self, raw_text, target, has_media=False, *args, **kwargs): + return f"[{target.title}] {raw_text}" + + + +async def _cleanup(repo): + pool = await repo._get_pool() + async with pool.acquire() as conn: + for cid in (SRC_A, SRC_B): + await conn.execute("DELETE FROM posts WHERE source_channel_id = $1;", cid) + await conn.execute("DELETE FROM sources WHERE channel_id = $1;", cid) + await conn.execute("DELETE FROM targets WHERE channel_id = $1;", TRG) + + +async def run_tests(): + await init_db() + repo = Repository() + await _cleanup(repo) + + await repo.add_source(SRC_A, "Source A", None) + await repo.add_source(SRC_B, "Source B", None) + target_id = await repo.add_target(TRG, "Auto Target", None) + + # A brand new target routes nothing. + assert await repo.get_targets_auto_routed_from(SRC_A) == [] + + # Toggle is a switch, not a one-way door. + assert await repo.toggle_target_auto_source(target_id, SRC_A) is True + routed = await repo.get_targets_auto_routed_from(SRC_A) + assert len(routed) == 1 and routed[0].id == target_id + assert routed[0].auto_source_ids == [SRC_A] + assert await repo.toggle_target_auto_source(target_id, SRC_A) is False + assert await repo.get_targets_auto_routed_from(SRC_A) == [] + + # Subscribe to A only; B must stay manual. + await repo.set_target_auto_sources(target_id, [SRC_A]) + + queue = FakeQueue() + bot = AdminBotService(repo=repo, ai_processor=FakeAI(), queue=queue, + review_channel_id=0, admin_user_ids=[1]) + + post_a = await repo.create_raw_post(SRC_A, 1, "post from A", content_hash="auto_a") + post_b = await repo.create_raw_post(SRC_B, 2, "post from B", content_hash="auto_b") + + assert await bot.auto_route_post(post_a) == 1, "subscribed source must route" + assert await bot.auto_route_post(post_b) == 0, "unsubscribed source must not route" + + assert len(queue.pushed) == 1, f"expected 1 enqueue, got {queue.pushed}" + tid, payload = queue.pushed[0] + assert tid == target_id + assert payload["text"] == "[Auto Target] post from A", f"AI rewrite not applied: {payload['text']}" + assert payload["post_id"] == post_a + + # The post is recorded against the target but must NOT be marked published yet. + stored = await repo.get_post_by_id(post_a) + assert stored.status == "pending_review", f"auto-routing must not publish, got {stored.status}" + assert len(stored.published_to) == 1 + assert stored.published_to[0]["target_id"] == target_id + assert stored.published_to[0]["published_at"] is None + + # A soft-deleted post is never auto-routed. + await repo.soft_delete_post(post_b) + await repo.set_target_auto_sources(target_id, [SRC_A, SRC_B]) + assert await bot.auto_route_post(post_b) == 0, "deleted posts must not route" + + await _cleanup(repo) + await close_db_pool() + print("All auto-routing tests passed successfully!") + + +if __name__ == "__main__": + asyncio.run(run_tests()) diff --git a/tests/test_db.py b/tests/test_db.py deleted file mode 100644 index 3b510e7..0000000 --- a/tests/test_db.py +++ /dev/null @@ -1,63 +0,0 @@ -import asyncio -import os -import tempfile -from db.database import init_db -from db.repository import Repository - -async def run_tests(): - with tempfile.NamedTemporaryFile(suffix=".db") as tmp: - db_path = tmp.name - print(f"Testing DB on {db_path}...") - await init_db(db_path) - repo = Repository(db_path) - - # 1. Test Sources - s_id = await repo.add_source(-1001234567890, "Source Tech", "source_tech") - sources = await repo.get_active_sources() - assert len(sources) == 1 - assert sources[0].channel_id == -1001234567890 - - # 2. Test Targets - t_id = await repo.add_target(-1009876543210, "Target Channel", "target_chan", post_interval_min=15) - targets = await repo.get_active_targets() - assert len(targets) == 1 - assert targets[0].post_interval_min == 15 - - # 3. Test Raw Post & Deduplication - post_id = await repo.create_raw_post( - source_channel_id=-1001234567890, - source_message_id=101, - raw_text="Breaking news: AI update released!", - content_hash="hash_12345", - is_duplicate=False - ) - assert post_id is not None - - # Check duplicate lookup - dup_match = await repo.find_duplicate_post("hash_12345") - assert dup_match is not None - assert dup_match.id == post_id - - # 4. Test AI update & Approval flow - await repo.update_ai_result(post_id, subject="AI/Tech", ai_text="Rewritten: AI update is now live!", suggested_target_id=t_id) - pending_review = await repo.get_posts_by_status("pending_review") - assert len(pending_review) == 1 - assert pending_review[0].subject == "AI/Tech" - - await repo.approve_post(post_id, target_channel_id=t_id) - approved_post = await repo.get_next_approved_post_for_target(t_id) - assert approved_post is not None - assert approved_post.id == post_id - - await repo.mark_post_published(post_id) - assert await repo.get_next_approved_post_for_target(t_id) is None - - # 5. Settings - await repo.set_setting("system_prompt", "You are a professional editor.") - val = await repo.get_setting("system_prompt") - assert val == "You are a professional editor." - - print("All database tests passed successfully!") - -if __name__ == "__main__": - asyncio.run(run_tests()) diff --git a/tests/test_dedup.py b/tests/test_dedup.py index ee9c47b..9365290 100644 --- a/tests/test_dedup.py +++ b/tests/test_dedup.py @@ -1,3 +1,7 @@ +import sys + +sys.path.insert(0, "/app") + from core.dedup import normalize_text, compute_content_hash def test_normalize_text(): diff --git a/tests/test_publisher_failures.py b/tests/test_publisher_failures.py new file mode 100644 index 0000000..07cfc8e --- /dev/null +++ b/tests/test_publisher_failures.py @@ -0,0 +1,111 @@ +"""A target the account cannot post to must not spin the same post through the queue forever.""" +import asyncio +import sys + +sys.path.insert(0, "/app") + +from telethon.errors import ChatAdminRequiredError +from db.models import TargetChannel +from services.publisher import PublisherService + + +class FakeQueue: + def __init__(self, payloads): + self.items = list(payloads) + self.pushes = 0 + + async def get_target_queue_size(self, target_id): + return len(self.items) + + async def pop_target_post(self, target_id): + return self.items.pop(0) if self.items else None + + async def push_target_post(self, target_id, payload): + self.pushes += 1 + self.items.append(payload) + + +class FakeRepo: + def __init__(self, targets): + self._targets = targets + self.published = [] + + async def get_active_targets(self): + return self._targets + + async def record_post_published_to_target(self, *a): + self.published.append(a) + + async def update_target_last_post(self, *a): + pass + + +def make_target(): + return TargetChannel(id=2, channel_id=-1003848540849, title="newsjoker", username=None, + post_interval_min=1, last_post_time=None) + + +class FailingClient: + def __init__(self, error): + self.error = error + self.attempts = 0 + + def is_connected(self): + return True + + async def send_message(self, *a, **kw): + self.attempts += 1 + raise self.error + + async def send_file(self, *a, **kw): + self.attempts += 1 + raise self.error + + +async def drive(error, cycles=3): + queue = FakeQueue([{"post_id": 117, "text": "hello", "media_path": None}]) + repo = FakeRepo([make_target()]) + notes = [] + + async def notify(text): + notes.append(text) + + pub = PublisherService(repo=repo, queue=queue, client=FailingClient(error), notify_fn=notify) + for _ in range(cycles): + await pub._process_all_target_queues() + return queue, notes + + +async def _cleanup_error_rows(): + """log_exception writes to the real table even from fakes; remove those rows.""" + from db.database import init_db, get_db_pool + await init_db() + pool = await get_db_pool() + async with pool.acquire() as conn: + await conn.execute( + "DELETE FROM error_logs WHERE service_name = 'publisher.publish' " + "AND context->>'target_id' = '2' AND context->>'post_id' = '117';") + + +async def run_tests(): + # Permanent: drained after the first attempt, admins told exactly once. + queue, notes = await drive(ChatAdminRequiredError(request=None)) + assert queue.pushes == 0, f"permanent failure must not requeue, got {queue.pushes} pushes" + assert len(queue.items) == 0, f"queue should be drained, still holds {queue.items}" + assert len(notes) == 1, f"admins should be warned exactly once, got {len(notes)}" + assert "newsjoker" in notes[0] and "-1003848540849" in notes[0] + + # Transient: the post survives every failed cycle. + queue, notes = await drive(ConnectionError("network blip")) + assert queue.pushes == 3, f"transient failure must requeue each time, got {queue.pushes}" + assert len(queue.items) == 1, f"post must still be queued, got {queue.items}" + assert notes == [], "a transient blip must not raise a permanent-failure alarm" + + await _cleanup_error_rows() + from db.database import close_db_pool + await close_db_pool() + print("All publisher failure-handling tests passed successfully!") + + +if __name__ == "__main__": + asyncio.run(run_tests()) diff --git a/tests/test_scrape_history.py b/tests/test_scrape_history.py new file mode 100644 index 0000000..d9d75b9 --- /dev/null +++ b/tests/test_scrape_history.py @@ -0,0 +1,126 @@ +"""Covers the flow behind the '📥 استخراج ۲۰ پست' button, which used to abort silently.""" +import asyncio +import sys +import types + +sys.path.insert(0, "/app") + +from db.database import init_db, close_db_pool +from db.repository import Repository +from services.collector import CollectorService + +SOURCE_ID = -1009999000010 + + +class FakeMessage: + def __init__(self, msg_id, text): + self.id = msg_id + self.raw_text = text + self.media = None + + async def download_media(self, file=None): + return None + + +class FakeClient: + """Minimal stand-in for the Telethon client used by the collector.""" + + def __init__(self, messages): + self._messages = messages + + def is_connected(self): + return True + + async def is_user_authorized(self): + return True + + async def get_entity(self, ident): + return types.SimpleNamespace(id=abs(SOURCE_ID), title="Scrape Source") + + def iter_messages(self, entity, limit=20): + async def gen(): + for m in self._messages[:limit]: + yield m + return gen() + + +async def _cleanup(repo): + pool = await repo._get_pool() + async with pool.acquire() as conn: + await conn.execute("DELETE FROM posts WHERE source_channel_id = $1;", SOURCE_ID) + await conn.execute("DELETE FROM sources WHERE channel_id = $1;", SOURCE_ID) + # The failure-path assertion logs a real error row; don't leave it behind. + await conn.execute( + "DELETE FROM error_logs WHERE context->>'channel_id' = $1;", str(SOURCE_ID)) + + +async def run_tests(): + await init_db() + repo = Repository() + await _cleanup(repo) + await repo.add_source(SOURCE_ID, "Scrape Source", None) + + reviewed = [] + collector = CollectorService(repo=repo, on_post_received=lambda pid: _record(reviewed, pid)) + # 20 distinct messages plus one exact repeat of the first, to exercise dedup. + messages = [FakeMessage(1000 + i, f"historical post number {i}") for i in range(20)] + messages.append(FakeMessage(1099, "historical post number 0")) + collector.client = FakeClient(messages) + + progress = [] + + async def on_progress(text): + progress.append(text) + + res = await collector.scrape_channel_history(SOURCE_ID, limit=21, progress_callback=on_progress) + + assert res.collected == 21, f"expected 21 posts collected, got {res.collected}" + assert res.scanned == 21, f"expected 21 messages scanned, got {res.scanned}" + assert res.duplicates == 1, f"expected 1 duplicate, got {res.duplicates}" + assert res.already_stored == 0, f"expected 0 already-stored, got {res.already_stored}" + assert len(reviewed) == 21, f"expected 21 review cards dispatched, got {len(reviewed)}" + assert progress and progress[-1].startswith("✅"), f"admin was not told the result: {progress}" + + # Re-running the same window adds nothing new, and the report must say WHY + # rather than looking like a dead button. + progress.clear() + again = await collector.scrape_channel_history(SOURCE_ID, limit=21, progress_callback=on_progress) + assert again.collected == 0, f"repeat scrape should add nothing, got {again.collected}" + assert again.already_stored == 21, f"expected 21 already-stored, got {again.already_stored}" + assert progress[-1].startswith("ℹ️"), f"repeat scrape must be explained: {progress[-1]}" + assert "قبلا ذخیره شده" in progress[-1] + + stored = await repo.get_posts_by_status("pending_review", limit=100) + mine = [p for p in stored if p.source_channel_id == SOURCE_ID] + assert len(mine) == 21, f"expected 21 stored posts, got {len(mine)}" + + # The repeated text must be flagged against the original rather than stored blind. + dupes = [p for p in mine if p.is_duplicate] + assert len(dupes) == 1, f"expected exactly 1 duplicate flagged, got {len(dupes)}" + assert dupes[0].duplicate_of_id is not None + + # A channel that cannot be resolved reports the failure instead of vanishing. + class BrokenClient(FakeClient): + async def get_entity(self, ident): + raise RuntimeError("channel not reachable") + + async def get_dialogs(self, limit=50): + return [] + + collector.client = BrokenClient([]) + progress.clear() + result = await collector.scrape_channel_history(SOURCE_ID, limit=5, progress_callback=on_progress) + assert result.collected == 0 and result.error + assert progress and progress[-1].startswith("❌"), f"failure was not reported: {progress}" + + await _cleanup(repo) + await close_db_pool() + print("All scrape-history tests passed successfully!") + + +async def _record(bucket, post_id): + bucket.append(post_id) + + +if __name__ == "__main__": + asyncio.run(run_tests())