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

lint: fix bare except and unused variables

Browse Source
This commit is contained in:
mr-forust
2026-06-19 11:52:30 +02:00
parent df644219de
commit ac5beb831b
15 changed files with 1359 additions and 728 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.editorconfig
+24
View File
@@ -0,0 +1,24 @@
root = true
[*]
indent_style = space
indent_size = 2
end_of_line = lf
charset = utf-8
trim_trailing_whitespace = true
insert_final_newline = true
[*.{yml,yaml}]
indent_size = 2
[*.{json,jsonc}]
indent_size = 2
[*.md]
trim_trailing_whitespace = false
[*.py]
indent_size = 4
[{Makefile,makefile}]
indent_style = tab
.hadolint.yaml
+10
View File
@@ -0,0 +1,10 @@
ignored:
- DL3008
- DL3042
- DL3018
- DL3059
trustedRegistries:
- docker.io
- ghcr.io
- quay.io
- gcr.forust.xyz
.markdownlint.json
+8
View File
@@ -0,0 +1,8 @@
{
"default": true,
"MD013": false,
"MD024": false,
"MD033": false,
"MD041": false,
"MD046": false
}
.prettierrc.yaml
+8
View File
@@ -0,0 +1,8 @@
bracketSameLine: true
htmlWhitespaceSensitivity: css
printWidth: 120
tabWidth: 2
singleQuote: true
trailingComma: all
proseWrap: preserve
endOfLine: lf
.vscode/settings.json
Vendored
+34
View File
@@ -0,0 +1,34 @@
{
"[yaml]": {
"editor.tabSize": 2,
"editor.insertSpaces": true,
"editor.formatOnSave": true,
"editor.defaultFormatter": "redhat.vscode-yaml"
},
"[python]": {
"editor.tabSize": 4,
"editor.formatOnSave": true,
"editor.defaultFormatter": "charliermarsh.ruff"
},
"[dockerfile]": {
"editor.formatOnSave": true,
"editor.defaultFormatter": "exiasr.hadolint"
},
"[markdown]": {
"editor.wordWrap": "wordWrapColumn",
"editor.wordWrapColumn": 120
},
"yaml.schemas": {
"kubernetes": ["**/k8s/*.yaml", "**/k8s/*.yml"]
},
"yaml.format.enable": true,
"yaml.validate": true,
"yaml.completion": true,
"prettier.bracketSameLine": true,
"prettier.htmlWhitespaceSensitivity": "css",
"prettier.printWidth": 120,
"editor.formatOnSave": true,
"files.autoSave": "onFocusChange",
"editor.wordWrapColumn": 120,
"editor.wordWrap": "wordWrapColumn"
}
.yamllint
+17
View File
@@ -0,0 +1,17 @@
extends: default
rules:
comments:
min-spaces-from-content: 1
comments-indentation: false
document-start: disable
line-length: disable
braces:
min-spaces-inside: 0
max-spaces-inside: 1
brackets:
min-spaces-inside: 0
max-spaces-inside: 1
indentation:
spaces: 2
indent-sequences: consistent
.zed/settings.json
+39
View File
@@ -0,0 +1,39 @@
{
"tab_size": 2,
"soft_wrap": "prefer_line",
"preferred_line_length": 120,
"format_on_save": "on",
"languages": {
"YAML": {
"tab_size": 2,
"hard_tabs": false,
"format_on_save": "on",
"formatter": {
"language_server": { "name": "yaml-language-server" }
}
},
"Python": {
"tab_size": 4,
"format_on_save": "on",
"formatter": {
"language_server": { "name": "ruff" }
}
}
},
"lsp": {
"yaml-language-server": {
"settings": {
"yaml": {
"schemas": {
"kubernetes": ["**/k8s/*.yaml", "**/k8s/*.yml"]
},
"validate": true,
"completion": true,
"format": {
"enable": true
}
}
}
}
}
}
app_state.py
+6 -2
View File
@@ -57,7 +57,9 @@ class StateStore:
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = self.path.with_suffix(self.path.suffix + ".tmp")
with tmp_path.open("w", encoding="utf-8") as handle:
json.dump(self._merge_defaults(state), handle, ensure_ascii=False, indent=2)
json.dump(
self._merge_defaults(state), handle, ensure_ascii=False, indent=2
)
handle.write("\n")
tmp_path.replace(self.path)
self._cache = None
@@ -71,7 +73,9 @@ class StateStore:
def continuous_config(self) -> Dict[str, Any]:
state = self.load()
return dict(state.get("continuous_scraping") or DEFAULT_STATE["continuous_scraping"])
return dict(
state.get("continuous_scraping") or DEFAULT_STATE["continuous_scraping"]
)
def save_continuous_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
def mutate(state: Dict[str, Any]) -> None:
compose.yaml
+6
View File
@@ -10,6 +10,12 @@ services:
tty: true
ports:
- "7887:8080"
healthcheck:
test: ["CMD", "python", "-c", "import http.client,json;c=http.client.HTTPConnection('localhost',8080);c.request('GET','/health');r=c.getresponse();exit(0)if json.loads(r.read()).get('ok')else exit(1)"]
interval: 30s
timeout: 10s
retries: 3
start_period: 15s
volumes:
- ./data:/app/data
- ./session:/app/session
k8s/telegram-scraper.yaml
+12
View File
@@ -38,6 +38,18 @@ spec:
stdin: true
ports:
- containerPort: 8080
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 15
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 5
periodSeconds: 10
volumeMounts:
- name: telegram-scraper-data
mountPath: /app/data
scraper_jobs.py
+2 -1
View File
@@ -41,7 +41,8 @@ class ScraperJobService:
)
elif job_type == "scrape_selected":
await self._scrape_channels(
scraper, [str(channel_id) for channel_id in payload.get("channels", [])]
scraper,
[str(channel_id) for channel_id in payload.get("channels", [])],
)
elif job_type == "export_all":
await scraper.export_data()
scripts/smoke_test.py
+5 -1
View File
@@ -60,7 +60,11 @@ def main() -> int:
status, body = fetch(endpoint)
if status != 200:
raise RuntimeError(f"{endpoint} returned HTTP {status}")
if endpoint.endswith(".json") or endpoint.startswith("/api") or endpoint.startswith("/health"):
if (
endpoint.endswith(".json")
or endpoint.startswith("/api")
or endpoint.startswith("/health")
):
json.loads(body.decode("utf-8"))
print(f"ok {endpoint}")
return 0
telegram-scraper-OLD.py
+357 -181
View File
@@ -5,18 +5,28 @@ import csv
import asyncio
import time
import sys
import uuid
import warnings
from dataclasses import dataclass
from typing import Dict, List, Optional, Any
from pathlib import Path
from io import StringIO
from telethon import TelegramClient
from telethon.tl.types import MessageMediaPhoto, MessageMediaDocument, MessageMediaWebPage, User, PeerChannel, Channel, Chat
from telethon.tl.types import (
MessageMediaPhoto,
MessageMediaDocument,
MessageMediaWebPage,
User,
PeerChannel,
Channel,
Chat,
)
from telethon.errors import FloodWaitError, SessionPasswordNeededError
import qrcode
warnings.filterwarnings("ignore", message="Using async sessions support is an experimental feature")
warnings.filterwarnings(
"ignore", message="Using async sessions support is an experimental feature"
)
def display_ascii_art():
WHITE = "\033[97m"
@@ -31,6 +41,7 @@ ___________________ _________
"""
print(WHITE + art + RESET)
@dataclass
class MessageData:
message_id: int
@@ -48,9 +59,10 @@ class MessageData:
forwards: Optional[int]
reactions: Optional[str]
class OptimizedTelegramScraper:
def __init__(self):
self.STATE_FILE = 'state.json'
self.STATE_FILE = "state.json"
self.state = self.load_state()
self.client = None
self.continuous_scraping_active = False
@@ -62,21 +74,21 @@ class OptimizedTelegramScraper:
def load_state(self) -> Dict[str, Any]:
if os.path.exists(self.STATE_FILE):
try:
with open(self.STATE_FILE, 'r') as f:
with open(self.STATE_FILE, "r") as f:
return json.load(f)
except:
except Exception:
pass
return {
'api_id': None,
'api_hash': None,
'channels': {},
'channel_names': {},
'scrape_media': True,
"api_id": None,
"api_hash": None,
"channels": {},
"channel_names": {},
"scrape_media": True,
}
def save_state(self):
try:
with open(self.STATE_FILE, 'w') as f:
with open(self.STATE_FILE, "w") as f:
json.dump(self.state, f, indent=2)
except Exception as e:
print(f"Failed to save state: {e}")
@@ -86,17 +98,19 @@ class OptimizedTelegramScraper:
channel_dir = Path(channel)
channel_dir.mkdir(exist_ok=True)
db_file = channel_dir / f'{channel}.db'
db_file = channel_dir / f"{channel}.db"
conn = sqlite3.connect(str(db_file), check_same_thread=False)
conn.execute('''CREATE TABLE IF NOT EXISTS messages
conn.execute("""CREATE TABLE IF NOT EXISTS messages
(id INTEGER PRIMARY KEY, message_id INTEGER UNIQUE, date TEXT,
sender_id INTEGER, first_name TEXT, last_name TEXT, username TEXT,
message TEXT, media_type TEXT, media_path TEXT, reply_to INTEGER,
post_author TEXT, views INTEGER, forwards INTEGER, reactions TEXT)''')
conn.execute('CREATE INDEX IF NOT EXISTS idx_message_id ON messages(message_id)')
conn.execute('CREATE INDEX IF NOT EXISTS idx_date ON messages(date)')
conn.execute('PRAGMA journal_mode=WAL')
conn.execute('PRAGMA synchronous=NORMAL')
post_author TEXT, views INTEGER, forwards INTEGER, reactions TEXT)""")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_message_id ON messages(message_id)"
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_date ON messages(date)")
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.commit()
self.migrate_database(conn)
@@ -111,19 +125,19 @@ class OptimizedTelegramScraper:
columns = {row[1] for row in cursor.fetchall()}
migrations = []
if 'post_author' not in columns:
migrations.append('ALTER TABLE messages ADD COLUMN post_author TEXT')
if 'views' not in columns:
migrations.append('ALTER TABLE messages ADD COLUMN views INTEGER')
if 'forwards' not in columns:
migrations.append('ALTER TABLE messages ADD COLUMN forwards INTEGER')
if 'reactions' not in columns:
migrations.append('ALTER TABLE messages ADD COLUMN reactions TEXT')
if "post_author" not in columns:
migrations.append("ALTER TABLE messages ADD COLUMN post_author TEXT")
if "views" not in columns:
migrations.append("ALTER TABLE messages ADD COLUMN views INTEGER")
if "forwards" not in columns:
migrations.append("ALTER TABLE messages ADD COLUMN forwards INTEGER")
if "reactions" not in columns:
migrations.append("ALTER TABLE messages ADD COLUMN reactions TEXT")
for migration in migrations:
try:
conn.execute(migration)
except:
except Exception:
pass
if migrations:
@@ -139,20 +153,38 @@ class OptimizedTelegramScraper:
return
conn = self.get_db_connection(channel)
data = [(msg.message_id, msg.date, msg.sender_id, msg.first_name,
msg.last_name, msg.username, msg.message, msg.media_type,
msg.media_path, msg.reply_to, msg.post_author, msg.views,
msg.forwards, msg.reactions) for msg in messages]
data = [
(
msg.message_id,
msg.date,
msg.sender_id,
msg.first_name,
msg.last_name,
msg.username,
msg.message,
msg.media_type,
msg.media_path,
msg.reply_to,
msg.post_author,
msg.views,
msg.forwards,
msg.reactions,
)
for msg in messages
]
conn.executemany('''INSERT OR IGNORE INTO messages
conn.executemany(
"""INSERT OR IGNORE INTO messages
(message_id, date, sender_id, first_name, last_name, username,
message, media_type, media_path, reply_to, post_author, views,
forwards, reactions)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', data)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
data,
)
conn.commit()
async def download_media(self, channel: str, message) -> Optional[str]:
if not message.media or not self.state['scrape_media']:
if not message.media or not self.state["scrape_media"]:
return None
if isinstance(message.media, MessageMediaWebPage):
@@ -160,15 +192,15 @@ class OptimizedTelegramScraper:
try:
channel_dir = Path(channel)
media_folder = channel_dir / 'media'
media_folder = channel_dir / "media"
media_folder.mkdir(exist_ok=True)
if isinstance(message.media, MessageMediaPhoto):
original_name = getattr(message.file, 'name', None) or "photo.jpg"
original_name = getattr(message.file, "name", None) or "photo.jpg"
ext = "jpg"
elif isinstance(message.media, MessageMediaDocument):
ext = getattr(message.file, 'ext', 'bin') if message.file else 'bin'
original_name = getattr(message.file, 'name', None) or f"document.{ext}"
ext = getattr(message.file, "ext", "bin") if message.file else "bin"
original_name = getattr(message.file, "name", None) or f"document.{ext}"
else:
return None
@@ -205,14 +237,20 @@ class OptimizedTelegramScraper:
async def update_media_path(self, channel: str, message_id: int, media_path: str):
conn = self.get_db_connection(channel)
conn.execute('UPDATE messages SET media_path = ? WHERE message_id = ?',
(media_path, message_id))
conn.execute(
"UPDATE messages SET media_path = ? WHERE message_id = ?",
(media_path, message_id),
)
conn.commit()
async def scrape_channel(self, channel: str, offset_id: int):
try:
entity = await self.client.get_entity(PeerChannel(int(channel)) if channel.startswith('-') else channel)
result = await self.client.get_messages(entity, offset_id=offset_id, reverse=True, limit=0)
entity = await self.client.get_entity(
PeerChannel(int(channel)) if channel.startswith("-") else channel
)
result = await self.client.get_messages(
entity, offset_id=offset_id, reverse=True, limit=0
)
total_messages = result.total
if total_messages == 0:
@@ -227,7 +265,9 @@ class OptimizedTelegramScraper:
last_message_id = offset_id
semaphore = asyncio.Semaphore(self.max_concurrent_downloads)
async for message in self.client.iter_messages(entity, offset_id=offset_id, reverse=True):
async for message in self.client.iter_messages(
entity, offset_id=offset_id, reverse=True
):
try:
sender = await message.get_sender()
@@ -235,33 +275,45 @@ class OptimizedTelegramScraper:
if message.reactions and message.reactions.results:
reactions_parts = []
for reaction in message.reactions.results:
emoji = getattr(reaction.reaction, 'emoticon', '')
emoji = getattr(reaction.reaction, "emoticon", "")
count = reaction.count
if emoji:
reactions_parts.append(f"{emoji} {count}")
if reactions_parts:
reactions_str = ' '.join(reactions_parts)
reactions_str = " ".join(reactions_parts)
msg_data = MessageData(
message_id=message.id,
date=message.date.strftime('%Y-%m-%d %H:%M:%S'),
date=message.date.strftime("%Y-%m-%d %H:%M:%S"),
sender_id=message.sender_id,
first_name=getattr(sender, 'first_name', None) if isinstance(sender, User) else None,
last_name=getattr(sender, 'last_name', None) if isinstance(sender, User) else None,
username=getattr(sender, 'username', None) if isinstance(sender, User) else None,
message=message.message or '',
media_type=message.media.__class__.__name__ if message.media else None,
first_name=getattr(sender, "first_name", None)
if isinstance(sender, User)
else None,
last_name=getattr(sender, "last_name", None)
if isinstance(sender, User)
else None,
username=getattr(sender, "username", None)
if isinstance(sender, User)
else None,
message=message.message or "",
media_type=message.media.__class__.__name__
if message.media
else None,
media_path=None,
reply_to=message.reply_to_msg_id if message.reply_to else None,
post_author=message.post_author,
views=message.views,
forwards=message.forwards,
reactions=reactions_str
reactions=reactions_str,
)
message_batch.append(msg_data)
if self.state['scrape_media'] and message.media and not isinstance(message.media, MessageMediaWebPage):
if (
self.state["scrape_media"]
and message.media
and not isinstance(message.media, MessageMediaWebPage)
):
media_tasks.append(message)
last_message_id = message.id
@@ -272,15 +324,19 @@ class OptimizedTelegramScraper:
message_batch.clear()
if processed_messages % self.state_save_interval == 0:
self.state['channels'][channel] = last_message_id
self.state["channels"][channel] = last_message_id
self.save_state()
progress = (processed_messages / total_messages) * 100
bar_length = 30
filled_length = int(bar_length * processed_messages // total_messages)
bar = '█' * filled_length + '░' * (bar_length - filled_length)
filled_length = int(
bar_length * processed_messages // total_messages
)
bar = "█" * filled_length + "░" * (bar_length - filled_length)
sys.stdout.write(f"\r📄 Messages: [{bar}] {progress:.1f}% ({processed_messages}/{total_messages})")
sys.stdout.write(
f"\r📄 Messages: [{bar}] {progress:.1f}% ({processed_messages}/{total_messages})"
)
sys.stdout.flush()
except Exception as e:
@@ -304,13 +360,17 @@ class OptimizedTelegramScraper:
batch_size = 10
for i in range(0, len(media_tasks), batch_size):
batch = media_tasks[i : i + batch_size]
tasks = [asyncio.create_task(download_single_media(msg)) for msg in batch]
tasks = [
asyncio.create_task(download_single_media(msg)) for msg in batch
]
for j, task in enumerate(tasks):
try:
media_path = await task
if media_path:
await self.update_media_path(channel, batch[j].id, media_path)
await self.update_media_path(
channel, batch[j].id, media_path
)
successful_downloads += 1
except Exception:
pass
@@ -319,14 +379,18 @@ class OptimizedTelegramScraper:
progress = (completed_media / total_media) * 100
bar_length = 30
filled_length = int(bar_length * completed_media // total_media)
bar = '█' * filled_length + '░' * (bar_length - filled_length)
bar = "█" * filled_length + "░" * (bar_length - filled_length)
sys.stdout.write(f"\r📥 Media: [{bar}] {progress:.1f}% ({completed_media}/{total_media})")
sys.stdout.write(
f"\r📥 Media: [{bar}] {progress:.1f}% ({completed_media}/{total_media})"
)
sys.stdout.flush()
print(f"\n✅ Media download complete! ({successful_downloads}/{total_media} successful)")
print(
f"\n✅ Media download complete! ({successful_downloads}/{total_media} successful)"
)
self.state['channels'][channel] = last_message_id
self.state["channels"][channel] = last_message_id
self.save_state()
print(f"\nCompleted scraping channel {channel}")
@@ -336,19 +400,23 @@ class OptimizedTelegramScraper:
async def rescrape_media(self, channel: str):
conn = self.get_db_connection(channel)
cursor = conn.cursor()
cursor.execute('SELECT message_id FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage" AND media_path IS NULL')
cursor.execute(
'SELECT message_id FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage" AND media_path IS NULL'
)
message_ids = [row[0] for row in cursor.fetchall()]
channel_name = self.state.get('channel_names', {}).get(channel, 'Unknown')
channel_name = self.state.get("channel_names", {}).get(channel, "Unknown")
if not message_ids:
print(f"No media files to reprocess for {channel_name} (ID: {channel})")
return
print(f"📥 Reprocessing {len(message_ids)} media files for {channel_name} (ID: {channel})")
print(
f"📥 Reprocessing {len(message_ids)} media files for {channel_name} (ID: {channel})"
)
try:
if channel.lstrip('-').isdigit():
if channel.lstrip("-").isdigit():
entity = await self.client.get_entity(PeerChannel(int(channel)))
else:
entity = await self.client.get_entity(channel)
@@ -365,14 +433,25 @@ class OptimizedTelegramScraper:
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]
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
]
for j, task in enumerate(tasks):
try:
media_path = await task
if media_path:
await self.update_media_path(channel, valid_messages[j].id, media_path)
await self.update_media_path(
channel, valid_messages[j].id, media_path
)
successful_downloads += 1
except Exception:
pass
@@ -380,13 +459,19 @@ class OptimizedTelegramScraper:
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)
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.write(
f"\r🔄 Rescrape: [{bar}] {progress:.1f}% ({completed_media}/{len(message_ids)})"
)
sys.stdout.flush()
print(f"\n✅ Media reprocessing complete! ({successful_downloads}/{len(message_ids)} successful)")
print(
f"\n✅ Media reprocessing complete! ({successful_downloads}/{len(message_ids)} successful)"
)
except Exception as e:
print(f"Error reprocessing media: {e}")
@@ -395,15 +480,19 @@ class OptimizedTelegramScraper:
conn = self.get_db_connection(channel)
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage"')
cursor.execute(
'SELECT COUNT(*) FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage"'
)
total_with_media = cursor.fetchone()[0]
cursor.execute('SELECT COUNT(*) FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage" AND media_path IS NOT NULL')
cursor.execute(
'SELECT COUNT(*) FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage" AND media_path IS NOT NULL'
)
total_with_files = cursor.fetchone()[0]
missing_count = total_with_media - total_with_files
channel_name = self.state.get('channel_names', {}).get(channel, 'Unknown')
channel_name = self.state.get("channel_names", {}).get(channel, "Unknown")
print(f"\n📊 Media Analysis for {channel_name} (ID: {channel}):")
print(f"Messages with media: {total_with_media}")
print(f"Media files downloaded: {total_with_files}")
@@ -413,17 +502,21 @@ class OptimizedTelegramScraper:
print("✅ All media files are already downloaded!")
return
cursor.execute('SELECT message_id, media_type FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage" AND (media_path IS NULL OR media_path = "")')
cursor.execute(
'SELECT message_id, media_type FROM messages WHERE media_type IS NOT NULL AND media_type != "MessageMediaWebPage" AND (media_path IS NULL OR media_path = "")'
)
missing_media = cursor.fetchall()
if not missing_media:
print("✅ No missing media found!")
return
print(f"\n🔧 Attempting to download {len(missing_media)} missing media files...")
print(
f"\n🔧 Attempting to download {len(missing_media)} missing media files..."
)
try:
if channel.lstrip('-').isdigit():
if channel.lstrip("-").isdigit():
entity = await self.client.get_entity(PeerChannel(int(channel)))
else:
entity = await self.client.get_entity(channel)
@@ -441,15 +534,26 @@ class OptimizedTelegramScraper:
message_ids = [msg[0] for msg in batch]
messages = await self.client.get_messages(entity, ids=message_ids)
valid_messages = [msg for msg in messages if msg and msg.media and not isinstance(msg.media, MessageMediaWebPage)]
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]
tasks = [
asyncio.create_task(download_single_media(msg))
for msg in valid_messages
]
for j, task in enumerate(tasks):
try:
media_path = await task
if media_path:
await self.update_media_path(channel, valid_messages[j].id, media_path)
await self.update_media_path(
channel, valid_messages[j].id, media_path
)
successful_downloads += 1
except Exception:
pass
@@ -457,13 +561,19 @@ class OptimizedTelegramScraper:
completed_media += 1
progress = (completed_media / len(missing_media)) * 100
bar_length = 30
filled_length = int(bar_length * completed_media // len(missing_media))
bar = '█' * filled_length + '░' * (bar_length - filled_length)
filled_length = int(
bar_length * completed_media // len(missing_media)
)
bar = "█" * filled_length + "░" * (bar_length - filled_length)
sys.stdout.write(f"\r🔧 Fix Media: [{bar}] {progress:.1f}% ({completed_media}/{len(missing_media)})")
sys.stdout.write(
f"\r🔧 Fix Media: [{bar}] {progress:.1f}% ({completed_media}/{len(missing_media)})"
)
sys.stdout.flush()
print(f"\n✅ Media fix complete! ({successful_downloads}/{len(missing_media)} successful)")
print(
f"\n✅ Media fix complete! ({successful_downloads}/{len(missing_media)} successful)"
)
except Exception as e:
print(f"Error fixing missing media: {e}")
@@ -475,11 +585,11 @@ class OptimizedTelegramScraper:
while self.continuous_scraping_active:
start_time = time.time()
for channel in self.state['channels']:
for channel in self.state["channels"]:
if not self.continuous_scraping_active:
break
print(f"\nChecking for new messages in channel: {channel}")
await self.scrape_channel(channel, self.state['channels'][channel])
await self.scrape_channel(channel, self.state["channels"][channel])
elapsed = time.time() - start_time
sleep_time = max(0, 60 - elapsed)
@@ -492,19 +602,19 @@ class OptimizedTelegramScraper:
self.continuous_scraping_active = False
def get_export_filename(self, channel: str):
username = self.state.get('channel_names', {}).get(channel, 'no_username')
username = self.state.get("channel_names", {}).get(channel, "no_username")
return f"{channel}_{username}"
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 = Path(channel) / f"{filename}.csv"
cursor = conn.cursor()
cursor.execute('SELECT * FROM messages ORDER BY date')
cursor.execute("SELECT * FROM messages ORDER BY date")
columns = [description[0] for description in cursor.description]
with open(csv_file, 'w', newline='', encoding='utf-8') as f:
with open(csv_file, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(columns)
@@ -517,14 +627,14 @@ 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 = Path(channel) / f"{filename}.json"
cursor = conn.cursor()
cursor.execute('SELECT * FROM messages ORDER BY date')
cursor.execute("SELECT * FROM messages ORDER BY date")
columns = [description[0] for description in cursor.description]
with open(json_file, 'w', encoding='utf-8') as f:
f.write('[\n')
with open(json_file, "w", encoding="utf-8") as f:
f.write("[\n")
first_row = True
while True:
@@ -534,21 +644,21 @@ class OptimizedTelegramScraper:
for row in rows:
if not first_row:
f.write(',\n')
f.write(",\n")
else:
first_row = False
data = dict(zip(columns, row))
json.dump(data, f, ensure_ascii=False, indent=2)
f.write('\n]')
f.write("\n]")
async def export_data(self):
if not self.state['channels']:
if not self.state["channels"]:
print("No channels to export")
return
for channel in self.state['channels']:
for channel in self.state["channels"]:
print(f"Exporting data for channel {channel}...")
try:
self.export_to_csv(channel)
@@ -558,22 +668,30 @@ class OptimizedTelegramScraper:
print(f"❌ Export failed for channel {channel}: {e}")
async def view_channels(self):
if not self.state['channels']:
if not self.state["channels"]:
print("No channels saved")
return
print("\nCurrent channels:")
for i, (channel, last_id) in enumerate(self.state['channels'].items(), 1):
for i, (channel, last_id) in enumerate(self.state["channels"].items(), 1):
try:
conn = self.get_db_connection(channel)
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM messages')
cursor.execute("SELECT COUNT(*) FROM messages")
count = cursor.fetchone()[0]
channel_name = self.state.get('channel_names', {}).get(channel, 'Unknown')
print(f"[{i}] {channel_name} (ID: {channel}), Last Message ID: {last_id}, Messages: {count}")
except:
channel_name = self.state.get('channel_names', {}).get(channel, 'Unknown')
print(f"[{i}] {channel_name} (ID: {channel}), Last Message ID: {last_id}")
channel_name = self.state.get("channel_names", {}).get(
channel, "Unknown"
)
print(
f"[{i}] {channel_name} (ID: {channel}), Last Message ID: {last_id}, Messages: {count}"
)
except Exception:
channel_name = self.state.get("channel_names", {}).get(
channel, "Unknown"
)
print(
f"[{i}] {channel_name} (ID: {channel}), Last Message ID: {last_id}"
)
async def list_channels(self):
try:
@@ -582,23 +700,42 @@ class OptimizedTelegramScraper:
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"
username = getattr(entity, 'username', None) or 'no_username'
print(f"[{count}] {dialog.title} (ID: {dialog.id}, Type: {channel_type}, Username: @{username})")
channels_data.append({
'number': count,
'channel_name': dialog.title,
'channel_id': str(dialog.id),
'username': username,
'type': channel_type
})
if dialog.id != 777000 and (
isinstance(entity, Channel) or isinstance(entity, Chat)
):
channel_type = (
"Channel"
if isinstance(entity, Channel) and entity.broadcast
else "Group"
)
username = getattr(entity, "username", None) or "no_username"
print(
f"[{count}] {dialog.title} (ID: {dialog.id}, Type: {channel_type}, Username: @{username})"
)
channels_data.append(
{
"number": count,
"channel_name": dialog.title,
"channel_id": str(dialog.id),
"username": username,
"type": channel_type,
}
)
count += 1
if channels_data:
csv_file = Path('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'])
csv_file = Path("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()
writer.writerows(channels_data)
print(f"\n✅ Saved channels list to {csv_file}")
@@ -661,18 +798,20 @@ class OptimizedTelegramScraper:
return False
async def initialize_client(self):
if not all([self.state.get('api_id'), self.state.get('api_hash')]):
if not all([self.state.get("api_id"), self.state.get("api_hash")]):
print("\n=== API Configuration Required ===")
print("You need to provide API credentials from https://my.telegram.org")
try:
self.state['api_id'] = int(input("Enter your API ID: "))
self.state['api_hash'] = input("Enter your API Hash: ")
self.state["api_id"] = int(input("Enter your API ID: "))
self.state["api_hash"] = input("Enter your API Hash: ")
self.save_state()
except ValueError:
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(
"session", self.state["api_id"], self.state["api_hash"]
)
try:
await self.client.connect()
@@ -687,11 +826,13 @@ class OptimizedTelegramScraper:
while True:
choice = input("Enter your choice (1 or 2): ").strip()
if choice in ['1', '2']:
if choice in ["1", "2"]:
break
print("Please enter 1 or 2")
success = await self.qr_code_auth() if choice == '1' else await self.phone_auth()
success = (
await self.qr_code_auth() if choice == "1" else await self.phone_auth()
)
if not success:
print("Authentication failed. Please try again.")
@@ -703,16 +844,16 @@ class OptimizedTelegramScraper:
return True
def parse_channel_selection(self, choice):
channels_list = list(self.state['channels'].keys())
channels_list = list(self.state["channels"].keys())
selected_channels = []
if choice.lower() == 'all':
if choice.lower() == "all":
return channels_list
for selection in [x.strip() for x in choice.split(',')]:
for selection in [x.strip() for x in choice.split(",")]:
try:
if selection.startswith('-'):
if selection in self.state['channels']:
if selection.startswith("-"):
if selection in self.state["channels"]:
selected_channels.append(selection)
else:
print(f"Channel ID {selection} not found in your channels")
@@ -721,14 +862,18 @@ class OptimizedTelegramScraper:
if 1 <= num <= len(channels_list):
selected_channels.append(channels_list[num - 1])
else:
print(f"Invalid channel number: {num}. Valid range: 1-{len(channels_list)}")
print(
f"Invalid channel number: {num}. Valid range: 1-{len(channels_list)}"
)
except ValueError:
print(f"Invalid input: {selection}. Use numbers (1,2,3) or full IDs (-100123...)")
print(
f"Invalid input: {selection}. Use numbers (1,2,3) or full IDs (-100123...)"
)
return selected_channels
async def scrape_specific_channels(self):
if not self.state['channels']:
if not self.state["channels"]:
print("No channels available. Use [L] to add channels first")
return
@@ -745,7 +890,7 @@ class OptimizedTelegramScraper:
print(f"\n🚀 Starting scrape of {len(selected_channels)} channel(s)...")
for i, channel in enumerate(selected_channels, 1):
print(f"\n[{i}/{len(selected_channels)}] Scraping: {channel}")
await self.scrape_channel(channel, self.state['channels'][channel])
await self.scrape_channel(channel, self.state["channels"][channel])
print(f"\n✅ Completed scraping {len(selected_channels)} channel(s)!")
else:
print("❌ No valid channels selected")
@@ -757,7 +902,9 @@ class OptimizedTelegramScraper:
print("=" * 40)
print("[S] Scrape channels")
print("[C] Continuous scraping")
print(f"[M] Media scraping: {'ON' if self.state['scrape_media'] else 'OFF'}")
print(
f"[M] Media scraping: {'ON' if self.state['scrape_media'] else 'OFF'}"
)
print("[L] List & add channels")
print("[R] Remove channels")
print("[E] Export data")
@@ -769,8 +916,8 @@ class OptimizedTelegramScraper:
choice = input("Enter your choice: ").lower().strip()
try:
if choice == 'r':
if not self.state['channels']:
if choice == "r":
if not self.state["channels"]:
print("No channels to remove")
continue
@@ -784,8 +931,8 @@ class OptimizedTelegramScraper:
if selected_channels:
removed_count = 0
for channel in selected_channels:
if channel in self.state['channels']:
del self.state['channels'][channel]
if channel in self.state["channels"]:
del self.state["channels"][channel]
print(f"✅ Removed channel {channel}")
removed_count += 1
else:
@@ -800,19 +947,21 @@ class OptimizedTelegramScraper:
else:
print("No valid channels selected")
elif choice == 's':
elif choice == "s":
await self.scrape_specific_channels()
elif choice == 'm':
self.state['scrape_media'] = not self.state['scrape_media']
elif choice == "m":
self.state["scrape_media"] = not self.state["scrape_media"]
self.save_state()
print(f"\n✅ Media scraping {'enabled' if self.state['scrape_media'] else 'disabled'}")
print(
f"\n✅ Media scraping {'enabled' if self.state['scrape_media'] else 'disabled'}"
)
elif choice == 'c':
elif choice == "c":
task = asyncio.create_task(self.continuous_scraping())
print("Continuous scraping started. Press Ctrl+C to stop.")
try:
await asyncio.sleep(float('inf'))
await asyncio.sleep(float("inf"))
except KeyboardInterrupt:
self.continuous_scraping_active = False
task.cancel()
@@ -822,10 +971,10 @@ class OptimizedTelegramScraper:
except asyncio.CancelledError:
pass
elif choice == 'e':
elif choice == "e":
await self.export_data()
elif choice == 'l':
elif choice == "l":
channels_data = await self.list_channels()
if not channels_data:
@@ -841,24 +990,37 @@ class OptimizedTelegramScraper:
if selection:
added_count = 0
if selection.lower() == 'all':
if selection.lower() == "all":
for channel_info in channels_data:
channel_id = channel_info['channel_id']
if channel_id not in self.state['channels']:
self.state['channels'][channel_id] = 0
if 'channel_names' not in self.state:
self.state['channel_names'] = {}
self.state['channel_names'][channel_id] = channel_info['username']
print(f"✅ Added channel {channel_info['channel_name']} (ID: {channel_id})")
channel_id = channel_info["channel_id"]
if channel_id not in self.state["channels"]:
self.state["channels"][channel_id] = 0
if "channel_names" not in self.state:
self.state["channel_names"] = {}
self.state["channel_names"][channel_id] = (
channel_info["username"]
)
print(
f"✅ Added channel {channel_info['channel_name']} (ID: {channel_id})"
)
added_count += 1
else:
print(f"Channel {channel_info['channel_name']} already added")
print(
f"Channel {channel_info['channel_name']} already added"
)
else:
for sel in [x.strip() for x in selection.split(',')]:
for sel in [x.strip() for x in selection.split(",")]:
try:
if sel.startswith('-'):
if sel.startswith("-"):
channel_id = sel
channel_info = next((c for c in channels_data if c['channel_id'] == channel_id), None)
channel_info = next(
(
c
for c in channels_data
if c["channel_id"] == channel_id
),
None,
)
if not channel_info:
print(f"Channel ID {channel_id} not found")
continue
@@ -866,19 +1028,27 @@ class OptimizedTelegramScraper:
num = int(sel)
if 1 <= num <= len(channels_data):
channel_info = channels_data[num - 1]
channel_id = channel_info['channel_id']
channel_id = channel_info["channel_id"]
else:
print(f"Invalid number: {num}. Choose 1-{len(channels_data)}")
print(
f"Invalid number: {num}. Choose 1-{len(channels_data)}"
)
continue
if channel_id in self.state['channels']:
print(f"Channel {channel_info['channel_name']} already added")
if channel_id in self.state["channels"]:
print(
f"Channel {channel_info['channel_name']} already added"
)
else:
self.state['channels'][channel_id] = 0
if 'channel_names' not in self.state:
self.state['channel_names'] = {}
self.state['channel_names'][channel_id] = channel_info['username']
print(f"✅ Added channel {channel_info['channel_name']} (ID: {channel_id})")
self.state["channels"][channel_id] = 0
if "channel_names" not in self.state:
self.state["channel_names"] = {}
self.state["channel_names"][channel_id] = (
channel_info["username"]
)
print(
f"✅ Added channel {channel_info['channel_name']} (ID: {channel_id})"
)
added_count += 1
except ValueError:
@@ -891,13 +1061,15 @@ class OptimizedTelegramScraper:
else:
print("No new channels were added")
elif choice == 't':
if not self.state['channels']:
elif choice == "t":
if not self.state["channels"]:
print("No channels available. Add channels first")
continue
await self.view_channels()
print("\nEnter channel NUMBER (1,2,3...) or full channel ID (-100123...)")
print(
"\nEnter channel NUMBER (1,2,3...) or full channel ID (-100123...)"
)
selection = input("Enter your selection: ").strip()
selected_channels = self.parse_channel_selection(selection)
@@ -910,13 +1082,15 @@ class OptimizedTelegramScraper:
else:
print("No valid channel selected")
elif choice == 'f':
if not self.state['channels']:
elif choice == "f":
if not self.state["channels"]:
print("No channels available. Add channels first")
continue
await self.view_channels()
print("\nEnter channel NUMBER (1,2,3...) or full channel ID (-100123...)")
print(
"\nEnter channel NUMBER (1,2,3...) or full channel ID (-100123...)"
)
selection = input("Enter your selection: ").strip()
selected_channels = self.parse_channel_selection(selection)
@@ -928,7 +1102,7 @@ class OptimizedTelegramScraper:
else:
print("No valid channel selected")
elif choice == 'q':
elif choice == "q":
print("\n👋 Goodbye!")
self.close_db_connections()
if self.client:
@@ -953,11 +1127,13 @@ class OptimizedTelegramScraper:
else:
print("Failed to initialize client. Exiting.")
async def main():
scraper = OptimizedTelegramScraper()
await scraper.run()
if __name__ == '__main__':
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
telegram_scraper_with_forwarding.py
+577 -311
View File
File diff suppressed because it is too large Load Diff
webui_server.py
+35 -13
View File
@@ -306,7 +306,9 @@ class JobRunner:
if job.status == "running":
running_jobs.append(job)
if running_jobs:
logger.warning("Waiting for %d running job(s) to finish...", len(running_jobs))
logger.warning(
"Waiting for %d running job(s) to finish...", len(running_jobs)
)
deadline = time.time() + timeout
while time.time() < deadline:
with self.lock:
@@ -561,6 +563,7 @@ class TelegramAuthManager:
async def _disconnect():
if self.client:
await self.client.disconnect()
if self.client:
try:
future = asyncio.run_coroutine_threadsafe(_disconnect(), self.loop)
@@ -831,7 +834,10 @@ def openapi_payload() -> Dict[str, Any]:
"requestBody": json_body(
{
"api_id": {"type": "integer", "example": 123456},
"api_hash": {"type": "string", "example": "0123456789abcdef"},
"api_hash": {
"type": "string",
"example": "0123456789abcdef",
},
},
["api_id", "api_hash"],
),
@@ -929,7 +935,11 @@ def openapi_payload() -> Dict[str, Any]:
{
"name": "limit",
"in": "query",
"schema": {"type": "integer", "default": 120, "maximum": 300},
"schema": {
"type": "integer",
"default": 120,
"maximum": 300,
},
},
{
"name": "before",
@@ -945,7 +955,10 @@ def openapi_payload() -> Dict[str, Any]:
"summary": "Track a channel",
"requestBody": json_body(
{
"channel_id": {"type": "string", "example": "-1001234567890"},
"channel_id": {
"type": "string",
"example": "-1001234567890",
},
"name": {"type": "string", "example": "Research feed"},
},
["channel_id"],
@@ -990,7 +1003,10 @@ def openapi_payload() -> Dict[str, Any]:
"schema": {"type": "string"},
}
],
"responses": {**json_response, "404": {"description": "Job not found"}},
"responses": {
**json_response,
"404": {"description": "Job not found"},
},
}
},
"/api/jobs/scrape": {
@@ -1416,15 +1432,18 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
try:
start, end = self._parse_range(range_header, file_size)
except ValueError:
self.send_error_json(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "Invalid range")
self.send_error_json(
HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "Invalid range"
)
return
content_length = end - start + 1
self.send_response(HTTPStatus.PARTIAL_CONTENT)
self.send_header("Content-Type", content_type)
self.send_header("Accept-Ranges", "bytes")
self.send_header("Last-Modified", time.strftime(
"%a, %d %b %Y %H:%M:%S GMT", time.gmtime(last_modified)
))
self.send_header(
"Last-Modified",
time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(last_modified)),
)
self.send_header("Content-Range", f"bytes {start}-{end}/{file_size}")
self.send_header("Content-Length", str(content_length))
self.end_headers()
@@ -1434,9 +1453,10 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", content_type)
self.send_header("Accept-Ranges", "bytes")
self.send_header("Last-Modified", time.strftime(
"%a, %d %b %Y %H:%M:%S GMT", time.gmtime(last_modified)
))
self.send_header(
"Last-Modified",
time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime(last_modified)),
)
self.send_header("Content-Length", str(file_size))
self.end_headers()
if not head_only:
@@ -1495,7 +1515,9 @@ class TelegramScraperWebServer(ThreadingHTTPServer):
self.job_runner = JobRunner()
self.auth_manager = TelegramAuthManager()
self.continuous_manager = ContinuousScrapeManager()
if START_CONTINUOUS and self.continuous_manager.snapshot()["config"].get("enabled", True):
if START_CONTINUOUS and self.continuous_manager.snapshot()["config"].get(
"enabled", True
):
self.continuous_manager.start()