404 lines
18 KiB
Python
404 lines
18 KiB
Python
import os
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from typing import Optional, Callable, Awaitable, Any, List
|
||
from telethon import TelegramClient, events
|
||
from telethon.errors import SessionPasswordNeededError
|
||
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 SOURCE_ACTIVITY_TOTAL, DUPLICATES_DETECTED_TOTAL
|
||
from core.proxy import get_telegram_proxy
|
||
from core.error_logger import log_exception
|
||
|
||
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"❌ خطا در دریافت پستهای کانال <code>{channel_id}</code>: {self.error}"
|
||
header = (
|
||
f"✅ <b>{self.collected}</b> پست جدید از <code>{channel_id}</code> دریافت و به کانال ادمین ارسال شد."
|
||
if self.collected
|
||
else f"ℹ️ هیچ پست <b>جدیدی</b> در <code>{channel_id}</code> پیدا نشد."
|
||
)
|
||
return (
|
||
f"{header}\n\n"
|
||
f"• 🔍 پیام بررسیشده: <b>{self.scanned}</b>\n"
|
||
f"• 🆕 پست جدید: <b>{self.collected}</b>\n"
|
||
f"• 🗂 قبلا ذخیره شده: <b>{self.already_stored}</b>\n"
|
||
f"• ♻️ محتوای تکراری: <b>{self.duplicates}</b>"
|
||
)
|
||
|
||
class CollectorService:
|
||
def __init__(
|
||
self,
|
||
repo: Repository,
|
||
on_post_received: Optional[Callable[[int], Awaitable[None]]] = None,
|
||
queue: Optional[RedisQueue] = None,
|
||
ai_processor: Optional[Any] = None,
|
||
api_id: Optional[int] = None,
|
||
api_hash: Optional[str] = None,
|
||
phone: Optional[str] = None,
|
||
session_name: Optional[str] = None,
|
||
):
|
||
self.repo = repo
|
||
self.on_post_received = on_post_received
|
||
self.queue = queue
|
||
self.ai_processor = ai_processor
|
||
self.api_id = api_id or int(os.getenv("API_ID", "0") or "0")
|
||
self.api_hash = api_hash or os.getenv("API_HASH", "")
|
||
self.phone = phone or os.getenv("PHONE")
|
||
self.session_name = session_name or os.path.join(SESSION_DIR, "collector.session")
|
||
os.makedirs(os.path.dirname(self.session_name), exist_ok=True)
|
||
os.makedirs(MEDIA_DIR, exist_ok=True)
|
||
effective_api_id = self.api_id if self.api_id else 12345
|
||
effective_api_hash = self.api_hash if self.api_hash else "0123456789abcdef0123456789abcdef"
|
||
self.client = TelegramClient(self.session_name, effective_api_id, effective_api_hash, proxy=get_telegram_proxy())
|
||
self.phone_code_hash: Optional[str] = None
|
||
self._handlers_registered = False
|
||
|
||
def set_ai_processor(self, ai_processor: Any) -> None:
|
||
self.ai_processor = ai_processor
|
||
|
||
async def start(self, notify_fn: Optional[Callable[[str], Awaitable[None]]] = None):
|
||
logger.info("Initializing Collector Userbot client...")
|
||
try:
|
||
await self.client.connect()
|
||
|
||
if await self.client.is_user_authorized():
|
||
me = await self.client.get_me()
|
||
logger.info(f"Collector Userbot is authorized as: {me.first_name} (@{me.username})")
|
||
self._register_handlers()
|
||
return True
|
||
|
||
logger.warning("Collector Userbot is not authorized. Requesting login code...")
|
||
if self.phone and notify_fn:
|
||
try:
|
||
sent = await self.client.send_code_request(self.phone)
|
||
self.phone_code_hash = sent.phone_code_hash
|
||
await notify_fn(
|
||
f"🔐 <b>نیاز به ورود ربات جمعآوریکننده</b>\n\n"
|
||
f"کد تایید تلگرام به شماره <code>{self.phone}</code> ارسال شد.\n\n"
|
||
f"لطفا با دستور زیر پاسخ دهید:\n"
|
||
f"<code>/code 12345</code>"
|
||
)
|
||
except Exception as e:
|
||
await log_exception("collector.login", e, {"phone": self.phone})
|
||
await notify_fn(f"❌ خطا در ارسال کد ورود: {e}")
|
||
return False
|
||
except Exception as e:
|
||
await log_exception("collector.start", e)
|
||
return False
|
||
|
||
async def submit_code(self, code: str) -> str:
|
||
if not self.phone or not self.phone_code_hash:
|
||
sent = await self.client.send_code_request(self.phone)
|
||
self.phone_code_hash = sent.phone_code_hash
|
||
|
||
try:
|
||
await self.client.sign_in(phone=self.phone, code=code, phone_code_hash=self.phone_code_hash)
|
||
me = await self.client.get_me()
|
||
self._register_handlers()
|
||
return f"✅ ورود موفقیتآمیز بود! حساب فعال: <b>{me.first_name}</b> (@{me.username or 'ندارد'})."
|
||
except SessionPasswordNeededError:
|
||
return "🔐 <b>رمز دو مرحلهای فعال است.</b> لطفا با این دستور رمز را وارد کنید: <code>/password رمز_عبور</code>"
|
||
except Exception as e:
|
||
await log_exception("collector.submit_code", e)
|
||
return f"❌ خطا در ورود: {e}"
|
||
|
||
async def submit_password(self, password: str) -> str:
|
||
try:
|
||
await self.client.sign_in(password=password)
|
||
me = await self.client.get_me()
|
||
self._register_handlers()
|
||
return f"✅ تایید دو مرحلهای موفق بود! حساب فعال: <b>{me.first_name}</b>."
|
||
except Exception as e:
|
||
await log_exception("collector.submit_password", e)
|
||
return f"❌ خطا در تایید رمز دو مرحلهای: {e}"
|
||
|
||
def _register_handlers(self):
|
||
if self._handlers_registered:
|
||
return
|
||
|
||
@self.client.on(events.NewMessage)
|
||
async def on_new_message(event: events.NewMessage.Event):
|
||
await self._handle_message(event)
|
||
|
||
self._handlers_registered = True
|
||
logger.info("Collector real-time event handlers registered.")
|
||
|
||
async def _resolve_channel_entity(self, channel_id: int, username: Optional[str] = None):
|
||
"""Robustly resolve channel entity even if not yet cached in local Telethon session."""
|
||
if username:
|
||
try:
|
||
clean_user = username.replace("@", "").strip()
|
||
return await self.client.get_entity(clean_user)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
return await self.client.get_entity(channel_id)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
# Refresh dialogs cache
|
||
await self.client.get_dialogs(limit=50)
|
||
return await self.client.get_entity(channel_id)
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
raw_str = str(channel_id).replace("-100", "").replace("-", "")
|
||
return await self.client.get_entity(int(raw_str))
|
||
except Exception as e:
|
||
raise ValueError(f"Could not resolve entity for channel {channel_id} (@{username}): {e}")
|
||
|
||
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:
|
||
return
|
||
|
||
raw_text = event.raw_text or ""
|
||
media_path = None
|
||
media_type = None
|
||
media_hash = None
|
||
|
||
if event.message.media:
|
||
if isinstance(event.message.media, MessageMediaPhoto):
|
||
media_type = "photo"
|
||
elif isinstance(event.message.media, MessageMediaDocument):
|
||
media_type = "document"
|
||
else:
|
||
media_type = "other"
|
||
|
||
filename = f"{chat_id}_{event.message.id}"
|
||
download_target = os.path.join(MEDIA_DIR, filename)
|
||
downloaded_file = await event.message.download_media(file=download_target)
|
||
if downloaded_file:
|
||
media_path = downloaded_file
|
||
media_hash = compute_file_hash(downloaded_file)
|
||
|
||
content_hash = compute_content_hash(raw_text, media_hash)
|
||
|
||
# 1. Extract subject and topic tags via AI
|
||
tags, subject = [], ""
|
||
if self.ai_processor:
|
||
try:
|
||
tags, subject = await self.ai_processor.extract_tags_and_subject(raw_text, media_path)
|
||
except Exception as e:
|
||
logger.debug(f"Tag extraction error: {e}")
|
||
|
||
# 2. Check duplicate via AI semantic similarity on candidate posts sharing tags
|
||
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(
|
||
raw_text, candidates, media_path
|
||
)
|
||
if is_dup:
|
||
is_duplicate = True
|
||
duplicate_of_id = dup_id
|
||
similarity_reason = reason
|
||
DUPLICATES_DETECTED_TOTAL.labels(method="ai_semantic").inc()
|
||
except Exception as e:
|
||
logger.debug(f"Semantic duplicate check error: {e}")
|
||
|
||
# 3. Fallback hash duplicate check if not already caught by AI
|
||
if not is_duplicate and content_hash:
|
||
duplicate_of = await self.repo.find_duplicate_post(content_hash)
|
||
if duplicate_of:
|
||
is_duplicate = True
|
||
duplicate_of_id = duplicate_of.id
|
||
similarity_reason = f"content hash matches post #{duplicate_of.id}"
|
||
DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc()
|
||
|
||
post_id = await self.repo.create_raw_post(
|
||
source_channel_id=chat_id,
|
||
source_message_id=event.message.id,
|
||
raw_text=raw_text,
|
||
media_path=media_path,
|
||
media_type=media_type,
|
||
content_hash=content_hash,
|
||
tags=tags,
|
||
subject=subject,
|
||
is_duplicate=is_duplicate,
|
||
duplicate_of_id=duplicate_of_id,
|
||
similarity_reason=similarity_reason,
|
||
source_created_at=getattr(event.message, "date", None),
|
||
)
|
||
|
||
if post_id:
|
||
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} (subject={subject}, tags={tags}, duplicate={is_duplicate})")
|
||
|
||
if self.on_post_received:
|
||
await self.on_post_received(post_id)
|
||
except Exception as e:
|
||
await log_exception("collector.handle_message", e, {"chat_id": event.chat_id, "msg_id": getattr(event.message, "id", None)})
|
||
|
||
async def scrape_channel_history(
|
||
self,
|
||
channel_id: int,
|
||
limit: int = 20,
|
||
progress_callback: Optional[Callable[[str], Awaitable[None]]] = None
|
||
) -> 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("❌ ربات متصل نیست. لطفا ابتدا لاگین کنید.")
|
||
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):
|
||
messages.append(msg)
|
||
|
||
messages.reverse()
|
||
|
||
result.scanned = len(messages)
|
||
|
||
for message in messages:
|
||
raw_text = message.raw_text or ""
|
||
if not raw_text and not message.media:
|
||
continue
|
||
|
||
media_path = None
|
||
media_type = None
|
||
media_hash = None
|
||
|
||
if message.media:
|
||
if isinstance(message.media, MessageMediaPhoto):
|
||
media_type = "photo"
|
||
elif isinstance(message.media, MessageMediaDocument):
|
||
media_type = "document"
|
||
else:
|
||
media_type = "other"
|
||
|
||
filename = f"{channel_id}_{message.id}"
|
||
download_target = os.path.join(MEDIA_DIR, filename)
|
||
downloaded_file = await message.download_media(file=download_target)
|
||
if downloaded_file:
|
||
media_path = downloaded_file
|
||
media_hash = compute_file_hash(downloaded_file)
|
||
|
||
content_hash = compute_content_hash(raw_text, media_hash)
|
||
|
||
# Extract tags and check semantic duplication
|
||
tags, subject = [], ""
|
||
if self.ai_processor:
|
||
try:
|
||
tags, subject = await self.ai_processor.extract_tags_and_subject(raw_text, media_path)
|
||
except Exception as e:
|
||
logger.debug(f"Tag extraction error: {e}")
|
||
|
||
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(
|
||
raw_text, candidates, media_path
|
||
)
|
||
if is_dup:
|
||
is_duplicate = True
|
||
duplicate_of_id = dup_id
|
||
similarity_reason = reason
|
||
result.duplicates += 1
|
||
DUPLICATES_DETECTED_TOTAL.labels(method="ai_semantic").inc()
|
||
except Exception as e:
|
||
logger.debug(f"Semantic duplicate check error: {e}")
|
||
|
||
if not is_duplicate and content_hash:
|
||
duplicate_of = await self.repo.find_duplicate_post(content_hash)
|
||
if duplicate_of:
|
||
is_duplicate = True
|
||
duplicate_of_id = duplicate_of.id
|
||
similarity_reason = f"content hash matches post #{duplicate_of.id}"
|
||
result.duplicates += 1
|
||
DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc()
|
||
|
||
post_id = await self.repo.create_raw_post(
|
||
source_channel_id=channel_id,
|
||
source_message_id=message.id,
|
||
raw_text=raw_text,
|
||
media_path=media_path,
|
||
media_type=media_type,
|
||
content_hash=content_hash,
|
||
tags=tags,
|
||
subject=subject,
|
||
is_duplicate=is_duplicate,
|
||
duplicate_of_id=duplicate_of_id,
|
||
similarity_reason=similarity_reason,
|
||
source_created_at=getattr(message, "date", None),
|
||
)
|
||
|
||
if post_id:
|
||
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:
|
||
result.already_stored += 1
|
||
|
||
if progress_callback:
|
||
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(result.summary_fa(channel_id))
|
||
return result
|
||
|
||
async def stop(self):
|
||
if self.client.is_connected():
|
||
await self.client.disconnect()
|
||
logger.info("Collector Userbot disconnected.")
|