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

fix(server): harden deployment, media, state, jobs

Browse Source 
- Lock down /media/: deny state.json, DBs, sessions; allowlist extensions incl. archives/docs (M-1)
- parse_bool() fixes; HEAD 404; shutdown drains queue; range edge cases (M-3, M-4, M-7, M-8)
- int() coercion -> 400; no filesystem paths in errors; path-only access log (M-19, L-1)
- Security headers, QR TTL 60s, trusted-host allowlist, legacy add/remove via update() (L-4, L-5, L-6, L-8)
- Clean continuous channels on import and migration; restart-during-drain; tombstone managers (F-1, F-3, F-4)
- Durability: fsync + unique tmp + stale sweep + 0600/0700 perms (M-10, M-18)
- Jobs run on dedicated loop thread; set_scrape_media passthrough; media chunked; state throttled;
  exact media file reuse; honest scrape failure status (M-11, M-12, M-13, M-14)
- Health aggregates per-account; legacy GETs delegate post-migration (M-15, M-9)
- k8s: runAsNonRoot 1000 + resource limits, no readOnlyRootFilesystem (M-16)
- UI: dropped-invalid and credentials-reentry toasts; swagger XSS-safe (F-2, L-9, L-2)
- CI: non-blocking pip-audit job in both workflows (L-3)
- 50 tests passing; REVIEW.md updated (C-1/M-20 won't fix: local-only by design)
This commit is contained in:
mr-forust
2026-09-07 12:36:33 +02:00
parent a2468a2a2c
commit e93e68db7e
13 changed files with 1182 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram_scraper_with_forwarding.py
+131 -53
View File
@@ -99,7 +99,7 @@ class OptimizedTelegramScraper:
base_dir = base_dir or BASE_DIR
self.BASE_DIR = base_dir
self.SESSION_DIR = base_dir / "session"
self.SESSION_DIR.mkdir(exist_ok=True)
self.SESSION_DIR.mkdir(exist_ok=True, mode=0o700)
if account_id:
self.DATA_DIR = base_dir / "data" / "accounts" / account_id
@@ -108,7 +108,7 @@ class OptimizedTelegramScraper:
self.DATA_DIR = base_dir / "data"
self.state_store = StateStore(self.DATA_DIR / "state.json")
self.DATA_DIR.mkdir(parents=True, exist_ok=True)
self.DATA_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
self.STATE_FILE = str(self.DATA_DIR / "state.json")
self.state = self.load_state()
@@ -118,6 +118,8 @@ class OptimizedTelegramScraper:
self.max_concurrent_downloads = 5
self.batch_size = 100
self.state_save_interval = 50
self.state_save_throttle_seconds = 5.0
self.last_state_save = None
self.db_connections = {}
self.forwarding_handler = None
@@ -127,9 +129,25 @@ class OptimizedTelegramScraper:
def save_state(self):
try:
self.state_store.save(self.state)
self.last_state_save = time.time()
except Exception as e:
print(f"Failed to save state: {e}")
def _save_state_throttled(self):
"""Throttled intermediate state persistence.
Avoids rewriting the whole per-account JSON on every
``state_save_interval`` messages (which is far too chatty for long
channels). Skips saves that fall within the throttle window; the
final save at end-of-scrape always runs regardless.
"""
now = time.time()
if (
self.last_state_save is None
or (now - self.last_state_save) >= self.state_save_throttle_seconds
):
self.save_state()
def get_forwarding_rules(self) -> List[ForwardingRule]:
rules = []
for rule_dict in self.state.get("forwarding_rules", []):
@@ -187,7 +205,7 @@ class OptimizedTelegramScraper:
def get_db_connection(self, channel: str) -> sqlite3.Connection:
if channel not in self.db_connections:
channel_dir = self.DATA_DIR / channel
channel_dir.mkdir(parents=True, exist_ok=True)
channel_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
db_file = channel_dir / f"{channel}.db"
conn = sqlite3.connect(str(db_file), check_same_thread=False, timeout=30)
@@ -300,7 +318,7 @@ class OptimizedTelegramScraper:
try:
channel_dir = self.DATA_DIR / channel
media_folder = channel_dir / "media"
media_folder.mkdir(exist_ok=True)
media_folder.mkdir(exist_ok=True, mode=0o700)
if isinstance(message.media, MessageMediaPhoto):
original_name = getattr(message.file, "name", None) or "photo.jpg"
@@ -316,9 +334,18 @@ class OptimizedTelegramScraper:
unique_filename = f"{message.id}-{base_name}{extension}"
media_path = media_folder / unique_filename
existing_files = list(media_folder.glob(f"{message.id}-*"))
if existing_files:
return str(existing_files[0])
# Prefer the exact expected filename. Fall back to a matching
# "{id}-*" file only if it is non-empty (and pick the newest one,
# in case a stale/partial/other-extension file is present).
if media_path.exists() and media_path.stat().st_size > 0:
return str(media_path)
candidates = [
p for p in media_folder.glob(f"{message.id}-*")
if p != media_path and p.is_file() and p.stat().st_size > 0
]
if candidates:
return str(max(candidates, key=lambda p: p.stat().st_mtime))
for attempt in range(3):
try:
@@ -859,7 +886,14 @@ class OptimizedTelegramScraper:
else:
return await self.client.get_entity(channel)
async def scrape_channel(self, channel: str, offset_id: int):
async def scrape_channel(self, channel: str, offset_id: int) -> bool:
"""Scrape a single channel. Returns True on success, False on failure.
Offset progress is persisted in a ``finally`` block so partial
progress is never lost even when an error occurs mid-scrape.
"""
last_message_id = offset_id
success = False
try:
if not self.client.is_connected():
await self.client.connect()
@@ -872,15 +906,70 @@ class OptimizedTelegramScraper:
if total_messages == 0:
print(f"No messages found in channel {channel}")
return
return True
print(f"Found {total_messages} messages in channel {channel}")
message_batch = []
media_tasks = []
processed_messages = 0
last_message_id = offset_id
semaphore = asyncio.Semaphore(self.max_concurrent_downloads)
media_flush_chunk = 50
# Media progress counters tracked across chunked flushes so the
# progress bar stays coherent even though downloads happen in
# bounded chunks during the pass instead of all at the end.
total_media = 0
completed_media = 0
successful_downloads = 0
async def flush_media_batch():
"""Download the accumulated media messages in small batches.
Keeps memory bounded (we never hold references to every
media-capable message for the whole channel), and updates the
shared media progress counters via ``nonlocal``.
"""
nonlocal total_media, completed_media, successful_downloads
if not media_tasks:
return
batch = list(media_tasks)
media_tasks.clear()
total_media += len(batch)
async def download_single_media(message):
async with semaphore:
return await self.download_media(channel, message)
sub_batch = 10
for i in range(0, len(batch), sub_batch):
sub = batch[i : i + sub_batch]
tasks = [
asyncio.create_task(download_single_media(msg)) for msg in sub
]
for j, task in enumerate(tasks):
try:
media_path = await task
if media_path:
await self.update_media_path(
channel, sub[j].id, media_path
)
successful_downloads += 1
except Exception:
pass
completed_media += 1
if total_media:
mprogress = (completed_media / total_media) * 100
bar_length = 30
mfilled = int(
bar_length * completed_media // total_media
)
mbar = "█" * mfilled + "░" * (bar_length - mfilled)
sys.stdout.write(
f"\r📥 Media: [{mbar}] {mprogress:.1f}% "
f"({completed_media}/{total_media})"
)
sys.stdout.flush()
async for message in self.client.iter_messages(
entity, offset_id=offset_id, reverse=True
@@ -932,6 +1021,11 @@ class OptimizedTelegramScraper:
and not isinstance(message.media, MessageMediaWebPage)
):
media_tasks.append(message)
# Flush the pending media list as soon as it reaches the
# bounded chunk so we never hold thousands of message
# references in memory for the whole channel.
if len(media_tasks) >= media_flush_chunk:
await flush_media_batch()
last_message_id = message.id
processed_messages += 1
@@ -939,10 +1033,15 @@ class OptimizedTelegramScraper:
if len(message_batch) >= self.batch_size:
self.batch_insert_messages(channel, message_batch)
message_batch.clear()
# After each insert batch, also flush any accumulated
# media (bounded) rather than deferring everything to
# the end of the full pass.
if media_tasks:
await flush_media_batch()
if processed_messages % self.state_save_interval == 0:
self.state["channels"][channel] = last_message_id
self.save_state()
self._save_state_throttled()
progress = (processed_messages / total_messages) * 100
bar_length = 30
@@ -963,56 +1062,35 @@ class OptimizedTelegramScraper:
self.batch_insert_messages(channel, message_batch)
if media_tasks:
total_media = len(media_tasks)
completed_media = 0
successful_downloads = 0
print(f"\n📥 Downloading {total_media} media files...")
semaphore = asyncio.Semaphore(self.max_concurrent_downloads)
async def download_single_media(message):
async with semaphore:
return await self.download_media(channel, message)
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
]
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
)
successful_downloads += 1
except Exception:
pass
completed_media += 1
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)
sys.stdout.write(
f"\r📥 Media: [{bar}] {progress:.1f}% ({completed_media}/{total_media})"
)
sys.stdout.flush()
await flush_media_batch()
if total_media:
print(
f"\n✅ Media download complete! ({successful_downloads}/{total_media} successful)"
)
self.state["channels"][channel] = last_message_id
self.save_state()
print(f"Completed scraping channel {channel}")
except Exception as e:
print(f"Error with channel {channel}: {e}")
# Final state save moved to ``finally`` below so the offset
# persists even when an exception aborts the scrape mid-way.
success = True
except Exception:
logger.exception("Error with channel %s", channel)
finally:
# Persist the last-known offset even on partial failure so a
# re-scrape resumes from the furthest point reached, not from the
# start. save_state() is itself best-effort (logs internally),
# so a save failure here must not mask the scrape's own result.
try:
self.state["channels"][channel] = last_message_id
self.save_state()
except Exception:
logger.exception("Failed to save state for channel %s", channel)
return success
async def rescrape_media(self, channel: str):
conn = self.get_db_connection(channel)