forust/telegram-scraper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Activity

Prepare scraper for shared data and server mode

Browse Source
This commit is contained in:
mr-forust
2026-04-25 02:19:43 +02:00
parent ec14ba8eb3
commit 535b8195ef
1 changed files with 60 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_scraper_with_forwarding.py
+60 -38
View File
@@ -61,7 +61,11 @@ class ForwardingRule:
class OptimizedTelegramScraper:
def __init__(self):
self.STATE_FILE = 'state.json'
self.DATA_DIR = Path('data')
self.SESSION_DIR = Path('session')
self.DATA_DIR.mkdir(exist_ok=True)
self.SESSION_DIR.mkdir(exist_ok=True)
self.STATE_FILE = str(self.DATA_DIR / 'state.json')
self.state = self.load_state()
self.client = None
self.continuous_scraping_active = False
@@ -147,8 +151,8 @@ class OptimizedTelegramScraper:
def get_db_connection(self, channel: str) -> sqlite3.Connection:
if channel not in self.db_connections:
channel_dir = Path(channel)
channel_dir.mkdir(exist_ok=True)
channel_dir = self.DATA_DIR / channel
channel_dir.mkdir(parents=True, exist_ok=True)
db_file = channel_dir / f'{channel}.db'
conn = sqlite3.connect(str(db_file), check_same_thread=False, timeout=30)
@@ -223,7 +227,7 @@ class OptimizedTelegramScraper:
return None
try:
channel_dir = Path(channel)
channel_dir = self.DATA_DIR / channel
media_folder = channel_dir / 'media'
media_folder.mkdir(exist_ok=True)
@@ -299,10 +303,7 @@ class OptimizedTelegramScraper:
async def forward_message(self, message, rule: ForwardingRule, source_channel_id: int = None):
try:
if rule.destination_channel.lstrip('-').isdigit():
dest_entity = await self.client.get_entity(PeerChannel(int(rule.destination_channel)))
else:
dest_entity = await self.client.get_entity(rule.destination_channel)
dest_entity = await self._resolve_entity(rule.destination_channel)
if rule.forward_mode == "forward":
await self.client.forward_messages(dest_entity, message)
@@ -527,9 +528,10 @@ class OptimizedTelegramScraper:
test_id = int(dest_input)
# Try to resolve the channel via Telethon to verify it exists
try:
entity = await self.client.get_entity(PeerChannel(test_id))
entity = await self._resolve_entity(dest_input)
dest_channel = dest_input
print(f"✅ Found channel: {getattr(entity, 'title', dest_input)}")
name = getattr(entity, 'title', None) or ' '.join(filter(None, [getattr(entity, 'first_name', None), getattr(entity, 'last_name', None)])) or dest_input
print(f"✅ Found: {name}")
except Exception:
print(f"❌ Could not access channel ID {dest_input}. Make sure this account is a member.")
return
@@ -692,12 +694,25 @@ class OptimizedTelegramScraper:
else:
print("❌ Failed to delete rule")
async def _resolve_entity(self, channel: str):
"""Resolve a channel/chat/user entity from an ID string or username."""
if channel.lstrip('-').isdigit():
cid = int(channel)
# Channel/supergroup IDs are negative and large; user IDs are positive or small negative
try:
return await self.client.get_entity(PeerChannel(cid))
except Exception:
# Fallback: let Telethon figure it out (works for users and basic groups)
return await self.client.get_entity(cid)
else:
return await self.client.get_entity(channel)
async def scrape_channel(self, channel: str, offset_id: int):
try:
if not self.client.is_connected():
await self.client.connect()
entity = await self.client.get_entity(PeerChannel(int(channel)) if channel.startswith('-') else channel)
entity = await self._resolve_entity(channel)
result = await self.client.get_messages(entity, offset_id=offset_id, reverse=True, limit=0)
total_messages = result.total
@@ -834,14 +849,11 @@ class OptimizedTelegramScraper:
print(f"📥 Reprocessing {len(message_ids)} media files for {channel_name} (ID: {channel})")
try:
if channel.lstrip('-').isdigit():
entity = await self.client.get_entity(PeerChannel(int(channel)))
else:
entity = await self.client.get_entity(channel)
entity = await self._resolve_entity(channel)
semaphore = asyncio.Semaphore(self.max_concurrent_downloads)
completed_media = 0
successful_downloads = 0
async def download_single_media(message):
async with semaphore:
return await self.download_media(channel, message)
@@ -850,7 +862,7 @@ class OptimizedTelegramScraper:
for i in range(0, len(message_ids), batch_size):
batch_ids = message_ids[i:i + batch_size]
messages = await self.client.get_messages(entity, ids=batch_ids)
valid_messages = [msg for msg in messages if msg and msg.media and not isinstance(msg.media, MessageMediaWebPage)]
tasks = [asyncio.create_task(download_single_media(msg)) for msg in valid_messages]
@@ -862,13 +874,13 @@ class OptimizedTelegramScraper:
successful_downloads += 1
except Exception:
pass
completed_media += 1
progress = (completed_media / len(message_ids)) * 100
bar_length = 30
filled_length = int(bar_length * completed_media // len(message_ids))
bar = '█' * filled_length + '░' * (bar_length - filled_length)
sys.stdout.write(f"\r🔄 Rescrape: [{bar}] {progress:.1f}% ({completed_media}/{len(message_ids)})")
sys.stdout.flush()
@@ -909,14 +921,11 @@ class OptimizedTelegramScraper:
print(f"\n🔧 Attempting to download {len(missing_media)} missing media files...")
try:
if channel.lstrip('-').isdigit():
entity = await self.client.get_entity(PeerChannel(int(channel)))
else:
entity = await self.client.get_entity(channel)
entity = await self._resolve_entity(channel)
semaphore = asyncio.Semaphore(self.max_concurrent_downloads)
completed_media = 0
successful_downloads = 0
async def download_single_media(message):
async with semaphore:
return await self.download_media(channel, message)
@@ -1063,7 +1072,7 @@ class OptimizedTelegramScraper:
def export_to_csv(self, channel: str):
conn = self.get_db_connection(channel)
filename = self.get_export_filename(channel)
csv_file = Path(channel) / f'{filename}.csv'
csv_file = self.DATA_DIR / channel / f'{filename}.csv'
cursor = conn.cursor()
cursor.execute('SELECT * FROM messages ORDER BY date')
@@ -1082,7 +1091,7 @@ class OptimizedTelegramScraper:
def export_to_json(self, channel: str):
conn = self.get_db_connection(channel)
filename = self.get_export_filename(channel)
json_file = Path(channel) / f'{filename}.json'
json_file = self.DATA_DIR / channel / f'{filename}.json'
cursor = conn.cursor()
cursor.execute('SELECT * FROM messages ORDER BY date')
@@ -1137,11 +1146,11 @@ class OptimizedTelegramScraper:
if not channel_name or channel_name in ['Unknown', 'no_username']:
try:
if channel.lstrip('-').isdigit():
entity = await self.client.get_entity(PeerChannel(int(channel)))
else:
entity = await self.client.get_entity(channel)
channel_name = getattr(entity, 'title', None) or getattr(entity, 'username', None) or 'Unknown'
entity = await self._resolve_entity(channel)
channel_name = getattr(entity, 'title', None) or getattr(entity, 'first_name', None) or getattr(entity, 'username', None) or 'Unknown'
# For users, combine first+last name
if isinstance(entity, User) and entity.first_name:
channel_name = ' '.join(filter(None, [entity.first_name, entity.last_name]))
if 'channel_names' not in self.state:
self.state['channel_names'] = {}
self.state['channel_names'][channel] = channel_name
@@ -1163,15 +1172,21 @@ class OptimizedTelegramScraper:
if not self.client.is_connected():
await self.client.connect()
print("\nList of channels and groups joined by account:")
print("\nList of channels, groups and private chats joined by account:")
count = 1
channels_data = []
async for dialog in self.client.iter_dialogs():
entity = dialog.entity
if dialog.id != 777000 and (isinstance(entity, Channel) or isinstance(entity, Chat)):
channel_type = "Channel" if isinstance(entity, Channel) and entity.broadcast else "Group"
if dialog.id != 777000 and (isinstance(entity, (Channel, Chat, User))):
if isinstance(entity, Channel):
channel_type = "Channel" if entity.broadcast else "Group"
elif isinstance(entity, Chat):
channel_type = "Group"
else:
channel_type = "Private"
username = getattr(entity, 'username', None) or 'no_username'
print(f"[{count}] {dialog.title} (ID: {dialog.id}, Type: {channel_type}, Username: @{username})")
display_username = f"@{username}" if username != 'no_username' else '(no username)'
print(f"[{count}] {dialog.title} (ID: {dialog.id}, Type: {channel_type}, Username: {display_username})")
channels_data.append({
'number': count,
'channel_name': dialog.title,
@@ -1182,7 +1197,7 @@ class OptimizedTelegramScraper:
count += 1
if channels_data:
csv_file = Path('channels_list.csv')
csv_file = self.DATA_DIR / 'channels_list.csv'
with open(csv_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=['number', 'channel_name', 'channel_id', 'username', 'type'])
writer.writeheader()
@@ -1246,8 +1261,11 @@ class OptimizedTelegramScraper:
print(f"\n❌ Phone authentication failed: {e}")
return False
async def initialize_client(self):
async def initialize_client(self, interactive: bool = True):
if not all([self.state.get('api_id'), self.state.get('api_hash')]):
if not interactive:
print("API credentials are missing in state.json")
return False
print("\n=== API Configuration Required ===")
print("You need to provide API credentials from https://my.telegram.org")
try:
@@ -1258,7 +1276,7 @@ class OptimizedTelegramScraper:
print("Invalid API ID. Must be a number.")
return False
self.client = TelegramClient('session', self.state['api_id'], self.state['api_hash'])
self.client = TelegramClient(str(self.SESSION_DIR / 'session'), self.state['api_id'], self.state['api_hash'])
try:
await self.client.connect()
@@ -1267,6 +1285,10 @@ class OptimizedTelegramScraper:
return False
if not await self.client.is_user_authorized():
if not interactive:
print("Telegram session is not authorized. Run the CLI once to sign in.")
await self.client.disconnect()
return False
print("\n=== Choose Authentication Method ===")
print("[1] QR Code (Recommended - No phone number needed)")
print("[2] Phone Number (Traditional method)")