hoelee/telegram-backup-downloader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Telegram Backup Downloader v5

Browse Source
This commit is contained in:
hoelee
2026-08-07 07:12:24 +08:00
commit 3efa56759f
9 changed files with 1573 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
.dockerignore Normal file
View File

@@ -0,0 +1,11 @@
__pycache__/
*.pyc
*.pyo
*.session
*.db
channels/
logs/
data/
.env
config.json
.hermes/

11
.gitignore vendored Normal file
View File

@@ -0,0 +1,11 @@
__pycache__/
*.pyc
*.pyo
*.session
*.db
channels/
logs/
data/
.env
config.json
.hermes/

13
Dockerfile Normal file
View File

@@ -0,0 +1,13 @@
FROM python:3.12-slim
WORKDIR /app
RUN addgroup --system appgroup && adduser --system --ingroup appgroup appuser
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY downloadv5.py .
RUN mkdir -p channels logs data && chown -R appuser:appgroup /app
USER appuser
VOLUME ["/app/channels", "/app/logs", "/app/data"]
EXPOSE 8080
HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health', timeout=3)" || exit 1
CMD ["python", "downloadv5.py"]

74
README.md Normal file
View File

@@ -0,0 +1,74 @@
# Telegram Backup Downloader v5
Downloads text, photos, videos, and documents from configured Telegram channels. Message text is stored in `messages.txt` and JSONL metadata; media is sorted into per-channel folders. SQLite tracks completed work, failed downloads, and sync progress.
## Prerequisites
- Python 3.12 or Docker
- A Telegram API ID and API hash from [my.telegram.org](https://my.telegram.org)
- A Telegram account that can access every configured channel
## Local Quick Start
1. Create a virtual environment and install dependencies: `python -m venv .venv` then `.venv\Scripts\pip install -r requirements.txt` on Windows, or `.venv/bin/pip install -r requirements.txt` on Linux/macOS.
2. Copy `config.example.json` to `config.json` and enter your Telegram values.
3. Run `python downloadv5.py`. The first run requests Telegram login verification if no session file exists.
Backups are written to `channels/`; logs are written to `logs/app.log`. Stop gracefully with Ctrl+C.
## Docker
1. Create and configure `config.json` from the example.
2. Run `docker compose up -d --build`.
3. Follow logs with `docker compose logs -f telegram-backup`.
The compose configuration persists channels, logs, data, and the Telegram session. Set `status_port` to `8080` to use the supplied health check and publish the status API.
## Configuration
| Key | Required | Default | Description |
|---|---:|---:|---|
| `api_id` | Yes | - | Numeric Telegram API ID. |
| `api_hash` | Yes | - | Telegram API hash. |
| `phone_number` | Yes | - | Account phone number in international format. |
| `session_name` | Yes | - | Telethon session basename. |
| `channels` | Yes | - | Channel usernames or numeric peer IDs. |
| `parallel_downloads` | No | `3` | Number of media workers. |
| `download_timeout_seconds` | No | `600` | Per-download timeout. |
| `download_retry_count` | No | `3` | Attempts made in one processing pass. |
| `max_lifetime_retries` | No | `20` | Total failed passes before an item is dropped; `0` disables the cap. |
| `queue_max_size` | No | `5000` | Normal download queue capacity. |
| `min_disk_space_gb` | No | `6` | Pause media downloads below this free space. |
| `channel_auto_disable_after` | No | `5` | Consecutive resolution failures before disabling a channel; `0` disables this. |
| `media_record_ttl_days` | No | `90` | Retention for completed media DB records; `0` disables pruning. |
| `retry_drop_log` | No | `true` | Write discarded downloads to `logs/dropped_downloads.jsonl`. |
| `resync_interval_minutes` | No | `60` | Periodic backfill interval; `0` disables it. |
| `status_port` | No | `0` | HTTP port; `0` disables the server. |
| `db_path` | No | `telegram_state.db` | SQLite state database path. |
| `channel_overrides` | No | `{}` | Per-channel `turnon` controls, for example `{"-1001":{"turnon":false}}`. |
| `manual_downloads` | No | `{}` | Message IDs to prioritize, keyed by channel ID. |
`config.json` is watched every 10 seconds. Changes to channels, overrides, and manual downloads are applied without restarting. Do not set `status_port`, `db_path`, worker count, or API credentials expecting a live process to rebind/recreate those resources; restart after changing them.
## HTTP API
The API listens on `0.0.0.0:<status_port>` with no authentication. Only expose it on a trusted network.
| Method | Endpoint | Description |
|---|---|---|
| GET | `/health` | Returns `{"status":"ok"}`. |
| GET | `/status` | Connection, queue, workers, disk, database, failure, and channel status. |
| GET | `/logs?lines=100` | Last 1-5000 lines of `logs/app.log`. |
| POST | `/reload` | Reloads `config.json`. |
| POST | `/db/cleanup` | Prunes expired records and vacuums SQLite. |
| POST | `/channel/{id}/enable` | Enables a channel override and resolves channels. |
| POST | `/channel/{id}/disable` | Disables a channel override. |
## Troubleshooting
- **Login problems:** delete only the session file if you intentionally need to authenticate again, then restart.
- **No downloads:** verify the account belongs to or can view the channel, and confirm the channel override is enabled.
- **Downloads paused:** inspect `/status` or the log for disk-space warnings. Downloads resume automatically after space is recovered.
- **Repeated failed media:** inspect `logs/app.log`; items reaching the lifetime cap are recorded in `logs/dropped_downloads.jsonl` when enabled.
- **Port unavailable:** set `status_port` to another free port, update compose port mapping if needed, and restart.
- **Database location in Docker:** set `db_path` to `data/telegram_state.db` if you want the database in the mounted `data` directory.

22
config.example.json Normal file
View File

@@ -0,0 +1,22 @@
{
"_comment": "Copy this to config.json and fill in your values",
"api_id": 12345678,
"api_hash": "your_api_hash_here",
"phone_number": "+1234567890",
"session_name": "telegram_session",
"channels": ["@channel_username", "-1001234567890"],
"parallel_downloads": 3,
"download_timeout_seconds": 600,
"download_retry_count": 3,
"max_lifetime_retries": 20,
"queue_max_size": 5000,
"min_disk_space_gb": 6,
"channel_auto_disable_after": 5,
"media_record_ttl_days": 90,
"retry_drop_log": true,
"resync_interval_minutes": 60,
"status_port": 8080,
"db_path": "telegram_state.db",
"channel_overrides": {},
"manual_downloads": {}
}

22
docker-compose.yml Normal file
View File

@@ -0,0 +1,22 @@
version: "3.9"
services:
telegram-backup:
build: .
container_name: telegram-backup
restart: unless-stopped
volumes:
- ./config.json:/app/config.json:ro
- ./telegram_session.session:/app/telegram_session.session
- ./data:/app/data
- ./channels:/app/channels
- ./logs:/app/logs
ports:
- "8080:8080"
environment:
- TZ=Asia/Kuala_Lumpur
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/health', timeout=3)"]
interval: 30s
timeout: 5s
retries: 3
start_period: 15s

633
downloadv4.py Normal file
View File

@@ -0,0 +1,633 @@
import asyncio
import json
import logging
import os
import re
import signal
import sys
import shutil
from datetime import datetime, timezone, timedelta
from logging.handlers import RotatingFileHandler
from pathlib import Path
import aiofiles
import aiosqlite
from telethon import TelegramClient, events, utils
from telethon.errors import (
FloodWaitError,
FileReferenceExpiredError,
RPCError,
)
# =========================
# LOAD CONFIG & CONSTANTS
# =========================
with open("config.json", "r", encoding="utf-8") as f:
CONFIG = json.load(f)
API_ID = CONFIG["api_id"]
API_HASH = CONFIG["api_hash"]
PHONE = CONFIG["phone_number"]
SESSION_NAME = CONFIG["session_name"]
CHANNELS = CONFIG["channels"]
PARALLEL_DOWNLOADS = CONFIG.get("parallel_downloads", 3)
DOWNLOAD_TIMEOUT = CONFIG.get("download_timeout_seconds", 600)
DOWNLOAD_RETRY = CONFIG.get("download_retry_count", 3)
QUEUE_MAX_SIZE = CONFIG.get("queue_max_size", 5000)
MIN_DISK_SPACE = 6 * 1024 * 1024 * 1024 # 6GB in bytes
BASE_DIR = Path("channels")
LOG_DIR = Path("logs")
BASE_DIR.mkdir(exist_ok=True)
LOG_DIR.mkdir(exist_ok=True)
# =========================
# LOGGING
# =========================
logger = logging.getLogger("telegram_archiver")
logger.setLevel(logging.INFO)
formatter = logging.Formatter(
"%(asctime)s [%(levelname)s] %(message)s"
)
file_handler = RotatingFileHandler(
LOG_DIR / "app.log",
maxBytes=10 * 1024 * 1024,
backupCount=10,
encoding="utf-8"
)
file_handler.setFormatter(formatter)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# =========================
# SQLITE
# =========================
DB_FILE = "telegram_state.db"
CREATE_TABLES_SQL = """
CREATE TABLE IF NOT EXISTS channel_state (
channel_id TEXT PRIMARY KEY,
last_message_id INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS downloaded_media (
channel_id TEXT,
message_id INTEGER,
file_name TEXT,
file_size INTEGER,
PRIMARY KEY(channel_id, message_id)
);
CREATE TABLE IF NOT EXISTS logged_messages (
channel_id TEXT,
message_id INTEGER,
PRIMARY KEY(channel_id, message_id)
);
CREATE TABLE IF NOT EXISTS failed_downloads (
channel_id TEXT,
message_id INTEGER,
last_error TEXT,
retry_count INTEGER,
PRIMARY KEY(channel_id,message_id)
);
"""
db = None
# =========================
# GLOBALS & SAFEGUARDS
# =========================
INVALID_CHARS = r'[<>:"/\\\\|?*]'
client = TelegramClient(
SESSION_NAME,
API_ID,
API_HASH,
auto_reconnect=True,
connection_retries=None,
retry_delay=5,
)
download_queue = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
priority_download_queue = asyncio.Queue()
shutdown_event = asyncio.Event()
shutdown_lock = asyncio.Lock()
CONFIG_LOCK = asyncio.Lock()
# Will be initialized inside main() to tie it to the correct event loop
download_allowed = None
CONFIG_FILE = "config.json"
MANUAL_IN_PROGRESS = set()
active_tasks = set()
worker_tasks = []
config_task = None
sync_task = None
disk_task = None
MONITORED_CHANNEL_IDS = set()
CHANNEL_ENTITY_MAP = {}
CHANNEL_OBJECT_MAP = {}
MY_TZ = timezone(timedelta(hours=8))
# =========================
# HELPERS
# =========================
def sanitize_filename(name):
name = re.sub(INVALID_CHARS, "_", name)
name = name.strip().rstrip(".")
return name[:200]
def parse_channel_id(raw_id):
if isinstance(raw_id, str) and raw_id.lstrip('-').isdigit():
return int(raw_id)
return raw_id
# =========================
# DATABASE OPERATIONS
# =========================
async def init_db():
global db
db = await aiosqlite.connect(DB_FILE)
await db.execute("PRAGMA journal_mode=WAL;")
await db.execute("PRAGMA synchronous=NORMAL;")
await db.executescript(CREATE_TABLES_SQL)
await db.commit()
async def add_logged_message(channel_id, message_id):
await db.execute("INSERT OR IGNORE INTO logged_messages VALUES (?,?)", (channel_id, message_id))
await db.commit()
async def resolve_channels():
MONITORED_CHANNEL_IDS.clear()
CHANNEL_ENTITY_MAP.clear()
CHANNEL_OBJECT_MAP.clear()
for channel in CHANNELS:
try:
parsed_channel = parse_channel_id(channel)
entity = await client.get_entity(parsed_channel)
entity_id = str(utils.get_peer_id(entity))
entity_title = getattr(entity, "title", entity_id)
MONITORED_CHANNEL_IDS.add(entity_id)
CHANNEL_ENTITY_MAP[entity_id] = entity_title
CHANNEL_OBJECT_MAP[entity_id] = entity
except Exception as e:
logger.error(f"[RESOLVE ERROR] Failed to resolve channel '{channel}': {e}")
async def get_last_message_id(channel_id):
async with db.execute("SELECT last_message_id FROM channel_state WHERE channel_id=?", (channel_id,)) as cursor:
row = await cursor.fetchone()
return row[0] if row else 0
async def update_last_message_id(channel_id, message_id):
await db.execute(
"""
INSERT INTO channel_state(channel_id, last_message_id)
VALUES (?, ?) ON CONFLICT(channel_id)
DO UPDATE SET last_message_id = MAX(last_message_id, excluded.last_message_id)
""", (channel_id, message_id)
)
await db.commit()
async def media_exists(channel_id, message_id):
async with db.execute("SELECT 1 FROM downloaded_media WHERE channel_id=? AND message_id=?", (channel_id, message_id)) as cursor:
return await cursor.fetchone() is not None
async def add_media_record(channel_id, message_id, file_name, file_size):
await db.execute(
"INSERT OR IGNORE INTO downloaded_media (channel_id, message_id, file_name, file_size) VALUES (?, ?, ?, ?)",
(channel_id, message_id, file_name, file_size)
)
await db.commit()
async def add_failed_download(channel_id, message_id, error):
await db.execute(
"""
INSERT INTO failed_downloads (channel_id, message_id, last_error, retry_count)
VALUES (?, ?, ?, 1) ON CONFLICT(channel_id,message_id)
DO UPDATE SET retry_count = retry_count + 1, last_error = excluded.last_error
""", (channel_id, message_id, error[:1000])
)
await db.commit()
async def text_logged(channel_id, message_id):
async with db.execute("SELECT 1 FROM logged_messages WHERE channel_id=? AND message_id=?", (channel_id, message_id)) as cursor:
return await cursor.fetchone() is not None
async def save_config():
async with CONFIG_LOCK:
async with aiofiles.open(CONFIG_FILE, "w", encoding="utf-8") as f:
await f.write(json.dumps(CONFIG, indent=2, ensure_ascii=False))
def get_channel_paths(channel_name):
safe_name = sanitize_filename(channel_name)
root = BASE_DIR / safe_name
folders = {
"root": root,
"photo": root / "photo",
"video": root / "video",
"document": root / "document",
"messages": root / "messages.txt",
"metadata": root / "metadata.jsonl",
}
for p in folders.values():
if isinstance(p, Path) and p.suffix == "":
p.mkdir(parents=True, exist_ok=True)
return folders
async def append_message_log(file_path, text):
async with aiofiles.open(file_path, "a", encoding="utf-8") as f:
await f.write(text)
async def append_metadata(file_path, data):
async with aiofiles.open(file_path, "a", encoding="utf-8") as f:
await f.write(json.dumps(data, ensure_ascii=False) + "\n")
def detect_media_type(message):
if message.photo: return "photo"
if message.video: return "video"
if message.document: return "document"
return None
# =========================
# CORE PROCESSING
# =========================
async def download_media(message, paths, channel_id):
if message.out: return
media_type = detect_media_type(message)
if not media_type: return
if await media_exists(channel_id, message.id): return
# === DISK SPACE SAFEGUARD ===
if not download_allowed.is_set():
logger.info(f"⏳ [PAUSED] Waiting for disk space to process media for message ID: {message.id}")
await download_allowed.wait()
if shutdown_event.is_set(): return
# ============================
entity = CHANNEL_OBJECT_MAP.get(channel_id)
if not entity: return
file_name = message.file.name if message.file else None
if not file_name:
ext = message.file.ext if message.file else ""
file_name = f"{message.id}{ext}"
file_name = sanitize_filename(file_name)
final_name = f"{message.id}_{file_name}"
target_dir = paths[media_type]
temp_path = target_dir / f"{final_name}.part"
target_path = target_dir / final_name
if target_path.exists():
size = target_path.stat().st_size
await add_media_record(channel_id, message.id, final_name, size)
return
logger.info(f"[DOWNLOAD] {final_name}")
success = False
last_error = "Unknown"
for attempt in range(1, DOWNLOAD_RETRY + 1):
if shutdown_event.is_set() or not client.is_connected():
break
try:
temp_path.unlink(missing_ok=True)
fresh_message = await client.get_messages(entity=entity, ids=message.id)
if not fresh_message:
raise Exception(f"Message {message.id} not found")
await asyncio.wait_for(fresh_message.download_media(file=temp_path), timeout=DOWNLOAD_TIMEOUT)
if not temp_path.exists():
raise Exception("Temp file missing")
size = temp_path.stat().st_size
if size <= 0:
raise Exception("Downloaded file is 0 bytes")
os.replace(temp_path, target_path)
await add_media_record(channel_id, message.id, final_name, size)
logger.info(f"[DOWNLOAD OK] {final_name} ({size} bytes)")
success = True
break
except Exception as e:
last_error = str(e)
logger.exception(f"[DOWNLOAD ERROR] {final_name}: {e}")
finally:
if temp_path.exists():
try:
if temp_path.stat().st_size == 0:
temp_path.unlink(missing_ok=True)
except Exception:
pass
await asyncio.sleep(min(5 * attempt, 5))
if not success and not shutdown_event.is_set():
await add_failed_download(channel_id, message.id, last_error)
logger.error(f"[FAILED] {final_name} after {DOWNLOAD_RETRY} attempts")
elif success:
await db.execute("DELETE FROM failed_downloads WHERE channel_id=? AND message_id=?", (channel_id, message.id))
await db.commit()
async def process_message(message, entity_title, channel_id):
paths = get_channel_paths(entity_title)
sender = await message.get_sender()
sender_name = "Unknown"
username = ""
if sender:
first = getattr(sender, "first_name", "") or ""
last = getattr(sender, "last_name", "") or ""
sender_name = f"{first} {last}".strip() or "Unknown"
username = getattr(sender, "username", "") or ""
timestamp = message.date.astimezone(MY_TZ).strftime("%Y-%m-%d %I:%M:%S %p")
if message.message:
if not await text_logged(channel_id, message.id):
media_info = ""
media_name = None
media_size = None
if message.media and message.file:
media_name = sanitize_filename(message.file.name or f"{message.id}{message.file.ext or ''}")
media_size = getattr(message.file, "size", 0) or 0
media_size_mb = media_size / (1024 * 1024)
media_info = f"\n\n[Media]\nFilename: {media_name}\nSize: {media_size_mb:.3f} MB"
text_log = (
f"[{timestamp}]\nSender Name: {sender_name}\nUsername: {username}\n"
f"Message ID: {message.id}\n{media_info}\n\n{message.message}\n{'-'*50}\n"
)
await append_message_log(paths["messages"], text_log)
await append_metadata(paths["metadata"], {
"timestamp": timestamp, "message_id": message.id, "sender_name": sender_name,
"username": username, "text": message.message, "media_filename": media_name, "media_size_bytes": media_size
})
await add_logged_message(channel_id, message.id)
if message.media:
await download_media(message, paths, channel_id)
await update_last_message_id(channel_id, message.id)
# =========================
# BACKGROUND MONITORS
# =========================
async def disk_monitor():
"""Continuously monitors disk space and pauses downloads if below 6GB."""
loop = asyncio.get_running_loop()
while not shutdown_event.is_set():
try:
free_space = shutil.disk_usage(BASE_DIR).free
if free_space < MIN_DISK_SPACE:
download_allowed.clear()
logger.warning(f"🚨 DISK SPACE CRITICAL: Only {free_space / 1024**3:.2f} GB left! (< 6GB)")
print("\n" + "="*60)
print("🛑 DOWNLOADS PAUSED DUE TO LOW DISK SPACE.")
print(f"Current Free Space: {free_space / 1024**3:.2f} GB")
print("Please free up storage space on the drive.")
print("="*60 + "\n")
# Asynchronously wait for terminal input without blocking the main event loop
await loop.run_in_executor(None, input, "👉 Press [ENTER] here once space is cleared to re-check...\n")
logger.info("Re-checking disk space...")
else:
if not download_allowed.is_set():
logger.info("✅ Disk space sufficient. Resuming operations.")
download_allowed.set()
await asyncio.sleep(5)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"[DISK MONITOR] Error: {e}")
await asyncio.sleep(5)
# =========================
# WORKERS & QUEUES
# =========================
async def worker(worker_id):
logger.info(f"[WORKER {worker_id}] Started")
while True:
if shutdown_event.is_set() and download_queue.empty() and priority_download_queue.empty():
break
task = None
queue_used = None
try:
try:
item = priority_download_queue.get_nowait()
queue_used = priority_download_queue
except asyncio.QueueEmpty:
item = await asyncio.wait_for(download_queue.get(), timeout=1)
queue_used = download_queue
message, entity_title, channel_id, is_manual, raw_channel_id = item
task = asyncio.create_task(process_message(message, entity_title, channel_id))
active_tasks.add(task)
await task
if is_manual:
await remove_manual_download(raw_channel_id, message.id)
MANUAL_IN_PROGRESS.discard(f"{channel_id}:{message.id}")
except asyncio.TimeoutError:
continue
except asyncio.CancelledError:
break
except Exception:
logger.exception(f"[WORKER {worker_id}]")
finally:
if task:
active_tasks.discard(task)
if queue_used:
queue_used.task_done()
logger.info(f"[WORKER {worker_id}] Exited")
# (Omitted config watchers and specific round-robin code below for brevity, use your existing functions)
# ... [Insert reload_config, config_watcher, apply_channel_overrides, remove_manual_download, process_manual_downloads, initial_sync, new_message_handler as written previously] ...
# NOTE: Make sure to include all of them when running. I am keeping the response focused on the modifications and the main block below!
# =========================
# INITIAL SYNC (ROUND-ROBIN)
# =========================
async def initial_sync():
try:
active_syncs = []
for entity_id, entity in CHANNEL_OBJECT_MAP.items():
if shutdown_event.is_set() or not client.is_connected():
return
try:
entity_title = CHANNEL_ENTITY_MAP[entity_id]
last_message_id = await get_last_message_id(entity_id)
min_id = max(0, last_message_id - 5)
logger.info(f"[SYNC INIT] {entity_title} (last_id={last_message_id}, sync min_id={min_id})")
iterator = client.iter_messages(entity, min_id=min_id, reverse=True).__aiter__()
active_syncs.append({"entity_id": entity_id, "entity_title": entity_title, "iterator": iterator, "count": 0})
except Exception:
logger.exception(f"[SYNC INIT ERROR] Failed to initialize iterator for {entity_id}")
chunk_size = 15
while active_syncs and not shutdown_event.is_set():
if not client.is_connected():
break
for sync in list(active_syncs):
if shutdown_event.is_set() or not client.is_connected(): break
exhausted = False
for _ in range(chunk_size):
try:
message = await sync["iterator"].__anext__()
await download_queue.put((message, sync["entity_title"], sync["entity_id"], False, None))
sync["count"] += 1
except StopAsyncIteration:
exhausted = True
break
except Exception as e:
logger.error(f"[SYNC ITER ERROR] Channel {sync['entity_title']}: {e}")
exhausted = True
break
if exhausted:
logger.info(f"[SYNC DONE] {sync['entity_title']} complete. Queued: {sync['count']}")
active_syncs.remove(sync)
await asyncio.sleep(0.01)
except asyncio.CancelledError:
pass
except Exception:
logger.exception("[SYNC CRITICAL ERROR]")
# =========================
# SHUTDOWN
# =========================
async def shutdown():
async with shutdown_lock:
if shutdown_event.is_set(): return
shutdown_event.set()
logger.info("[SHUTDOWN] Graceful shutdown started")
global config_task, sync_task, disk_task
if config_task and not config_task.done(): config_task.cancel()
if sync_task and not sync_task.done(): sync_task.cancel()
if disk_task and not disk_task.done(): disk_task.cancel()
try:
if client.is_connected(): await asyncio.wait_for(client.disconnect(), timeout=10)
except Exception:
logger.exception("[DISCONNECT ERROR]")
while not download_queue.empty():
try: download_queue.get_nowait(); download_queue.task_done()
except asyncio.QueueEmpty: break
while not priority_download_queue.empty():
try: priority_download_queue.get_nowait(); priority_download_queue.task_done()
except asyncio.QueueEmpty: break
if active_tasks: await asyncio.gather(*active_tasks, return_exceptions=True)
for task in worker_tasks: task.cancel()
await asyncio.gather(*worker_tasks, return_exceptions=True)
if db:
await db.commit()
await db.close()
logger.info("[SHUTDOWN COMPLETE]")
# =========================
# MAIN
# =========================
async def main():
global config_task, sync_task, disk_task, download_allowed
# Initialize the global event tracker inside the running loop
download_allowed = asyncio.Event()
download_allowed.set()
await init_db()
await client.start(phone=PHONE)
me = await client.get_me()
logger.info(f"[CONNECTED] {me.first_name}")
await resolve_channels()
for i in range(PARALLEL_DOWNLOADS):
worker_tasks.append(asyncio.create_task(worker(i + 1)))
logger.info("[LISTENER ACTIVE]")
# Track background tasks
disk_task = asyncio.create_task(disk_monitor())
sync_task = asyncio.create_task(initial_sync())
listener_task = asyncio.create_task(client.run_until_disconnected())
await listener_task
while not shutdown_event.is_set():
await asyncio.sleep(5)
# =========================
# SIGNAL & ENTRY
# =========================
def handle_exit(*args):
loop.call_soon_threadsafe(lambda: asyncio.create_task(shutdown()))
if __name__ == "__main__":
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
signal.signal(signal.SIGINT, handle_exit)
if hasattr(signal, "SIGTERM"):
signal.signal(signal.SIGTERM, handle_exit)
try:
loop.run_until_complete(main())
finally:
loop.run_until_complete(shutdown())

783
downloadv5.py Normal file
View File

@@ -0,0 +1,783 @@
"""Reliable Telegram channel backup downloader.
Configuration is read from config.json in the current working directory.
"""
import asyncio
import json
import logging
import os
import re
import shutil
import signal
import sys
from datetime import datetime, timedelta, timezone
from logging.handlers import RotatingFileHandler
from pathlib import Path
import aiofiles
import aiosqlite
from telethon import TelegramClient, events, utils
CONFIG_FILE = Path("config.json")
BASE_DIR = Path("channels")
LOG_DIR = Path("logs")
BASE_DIR.mkdir(exist_ok=True)
LOG_DIR.mkdir(exist_ok=True)
with CONFIG_FILE.open("r", encoding="utf-8") as config_file:
CONFIG = json.load(config_file)
API_ID = CONFIG.get("api_id")
API_HASH = CONFIG.get("api_hash")
PHONE = CONFIG.get("phone_number")
SESSION_NAME = CONFIG.get("session_name")
CHANNELS = CONFIG.get("channels", [])
PARALLEL_DOWNLOADS = CONFIG.get("parallel_downloads", 3)
DOWNLOAD_TIMEOUT = CONFIG.get("download_timeout_seconds", 600)
DOWNLOAD_RETRY = CONFIG.get("download_retry_count", 3)
QUEUE_MAX_SIZE = CONFIG.get("queue_max_size", 5000)
MIN_DISK_SPACE_GB = CONFIG.get("min_disk_space_gb", 6)
MIN_DISK_SPACE = MIN_DISK_SPACE_GB * 1024**3
STATUS_PORT = CONFIG.get("status_port", 0)
RESYNC_INTERVAL = CONFIG.get("resync_interval_minutes", 60)
MAX_LIFETIME_RETRIES = CONFIG.get("max_lifetime_retries", 20)
CHANNEL_AUTO_DISABLE_AFTER = CONFIG.get("channel_auto_disable_after", 5)
MEDIA_RECORD_TTL_DAYS = CONFIG.get("media_record_ttl_days", 90)
RETRY_DROP_LOG = CONFIG.get("retry_drop_log", True)
DB_PATH = CONFIG.get("db_path", "telegram_state.db")
START_TIME = datetime.now(timezone.utc)
logger = logging.getLogger("telegram_archiver")
logger.setLevel(logging.INFO)
logger.propagate = False
if not logger.handlers:
formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
file_handler = RotatingFileHandler(LOG_DIR / "app.log", maxBytes=10 * 1024 * 1024,
backupCount=10, encoding="utf-8")
console_handler = logging.StreamHandler(sys.stdout)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
CREATE_TABLES_SQL = """
CREATE TABLE IF NOT EXISTS channel_state (
channel_id TEXT PRIMARY KEY, last_message_id INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS downloaded_media (
channel_id TEXT, message_id INTEGER, file_name TEXT, file_size INTEGER,
downloaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(channel_id, message_id)
);
CREATE TABLE IF NOT EXISTS logged_messages (
channel_id TEXT, message_id INTEGER, PRIMARY KEY(channel_id, message_id)
);
CREATE TABLE IF NOT EXISTS failed_downloads (
channel_id TEXT, message_id INTEGER, last_error TEXT, retry_count INTEGER,
PRIMARY KEY(channel_id, message_id)
);
CREATE TABLE IF NOT EXISTS channel_fail_count (
channel_id TEXT PRIMARY KEY, consecutive_fails INTEGER DEFAULT 0, last_fail TEXT
);
"""
INVALID_CHARS = r'[<>:"/\\|?*]'
MY_TZ = timezone(timedelta(hours=8))
db = None
download_allowed = None
download_queue = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
priority_download_queue = asyncio.Queue()
shutdown_event = asyncio.Event()
shutdown_lock = asyncio.Lock()
CONFIG_LOCK = asyncio.Lock()
MANUAL_IN_PROGRESS = set()
MONITORED_CHANNEL_IDS = set()
CHANNEL_ENTITY_MAP = {}
CHANNEL_OBJECT_MAP = {}
active_tasks = set()
worker_tasks = []
worker_busy_count = 0
config_task = None
sync_task = None
disk_task = None
resync_task = None
status_task = None
supervisor_task = None
client = TelegramClient(SESSION_NAME, API_ID, API_HASH, auto_reconnect=True,
connection_retries=None, retry_delay=5)
def validate_config(cfg):
required = {
"api_id": int,
"api_hash": str,
"phone_number": str,
"session_name": str,
"channels": list,
}
for key, expected_type in required.items():
if key not in cfg:
raise ValueError(f"Missing required config key: {key}")
if not isinstance(cfg[key], expected_type) or (expected_type is str and not cfg[key].strip()):
raise ValueError(f"Config key '{key}' must be a non-empty {expected_type.__name__}")
if cfg["api_id"] <= 0:
raise ValueError("Config key 'api_id' must be a positive integer")
if not all(isinstance(channel, (str, int)) for channel in cfg["channels"]):
raise ValueError("Config key 'channels' must contain strings or integers")
def sanitize_filename(name):
return re.sub(INVALID_CHARS, "_", name).strip().rstrip(".")[:200] or "unnamed"
def parse_channel_id(raw_id):
if isinstance(raw_id, str) and raw_id.lstrip("-").isdigit():
return int(raw_id)
return raw_id
def get_channel_paths(channel_name):
root = BASE_DIR / sanitize_filename(channel_name)
paths = {"root": root, "photo": root / "photo", "video": root / "video",
"document": root / "document", "messages": root / "messages.txt",
"metadata": root / "metadata.jsonl"}
for key in ("root", "photo", "video", "document"):
paths[key].mkdir(parents=True, exist_ok=True)
return paths
def detect_media_type(message):
if message.photo:
return "photo"
if message.video:
return "video"
if message.document:
return "document"
return None
async def init_db():
global db
db = await aiosqlite.connect(DB_PATH)
await db.execute("PRAGMA journal_mode=WAL")
await db.execute("PRAGMA synchronous=NORMAL")
await db.executescript(CREATE_TABLES_SQL)
await db.commit()
try:
await db.execute("ALTER TABLE downloaded_media ADD COLUMN downloaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP")
await db.commit()
except Exception:
# SQLite raises if the column exists; older SQLite builds also reject
# CURRENT_TIMESTAMP as an ALTER TABLE default, handled below.
pass
async with db.execute("PRAGMA table_info(downloaded_media)") as cur:
columns = {row[1] for row in await cur.fetchall()}
if "downloaded_at" not in columns:
await db.execute("ALTER TABLE downloaded_media ADD COLUMN downloaded_at TEXT")
await db.execute("UPDATE downloaded_media SET downloaded_at=CURRENT_TIMESTAMP WHERE downloaded_at IS NULL")
await db.commit()
async def get_last_message_id(channel_id):
async with db.execute("SELECT last_message_id FROM channel_state WHERE channel_id=?", (channel_id,)) as cur:
row = await cur.fetchone()
return row[0] if row else 0
async def update_last_message_id(channel_id, message_id):
await db.execute("""INSERT INTO channel_state(channel_id,last_message_id) VALUES (?,?)
ON CONFLICT(channel_id) DO UPDATE SET last_message_id=MAX(last_message_id,excluded.last_message_id)""",
(channel_id, message_id))
await db.commit()
async def media_exists(channel_id, message_id):
async with db.execute("SELECT 1 FROM downloaded_media WHERE channel_id=? AND message_id=?", (channel_id, message_id)) as cur:
return await cur.fetchone() is not None
async def add_media_record(channel_id, message_id, file_name, file_size):
await db.execute("""INSERT OR IGNORE INTO downloaded_media
(channel_id,message_id,file_name,file_size) VALUES (?,?,?,?)""",
(channel_id, message_id, file_name, file_size))
await db.commit()
async def text_logged(channel_id, message_id):
async with db.execute("SELECT 1 FROM logged_messages WHERE channel_id=? AND message_id=?", (channel_id, message_id)) as cur:
return await cur.fetchone() is not None
async def add_logged_message(channel_id, message_id):
await db.execute("INSERT OR IGNORE INTO logged_messages VALUES (?,?)", (channel_id, message_id))
await db.commit()
async def add_failed_download(channel_id, message_id, error):
await db.execute("""INSERT INTO failed_downloads(channel_id,message_id,last_error,retry_count)
VALUES (?,?,?,1) ON CONFLICT(channel_id,message_id) DO UPDATE SET
retry_count=retry_count+1,last_error=excluded.last_error""", (channel_id, message_id, error[:1000]))
await db.commit()
async with db.execute("SELECT retry_count FROM failed_downloads WHERE channel_id=? AND message_id=?", (channel_id, message_id)) as cur:
row = await cur.fetchone()
return row[0] if row else 1
async def increment_channel_fail(channel_id, error):
await db.execute("""INSERT INTO channel_fail_count(channel_id,consecutive_fails,last_fail)
VALUES (?,1,?) ON CONFLICT(channel_id) DO UPDATE SET
consecutive_fails=consecutive_fails+1,last_fail=excluded.last_fail""", (channel_id, error[:500]))
await db.commit()
async with db.execute("SELECT consecutive_fails FROM channel_fail_count WHERE channel_id=?", (channel_id,)) as cur:
row = await cur.fetchone()
return row[0] if row else 1
async def reset_channel_fail(channel_id):
await db.execute("DELETE FROM channel_fail_count WHERE channel_id=?", (channel_id,))
await db.commit()
async def get_channel_fail_count(channel_id):
async with db.execute("SELECT consecutive_fails FROM channel_fail_count WHERE channel_id=?", (channel_id,)) as cur:
row = await cur.fetchone()
return row[0] if row else 0
async def save_config():
async with CONFIG_LOCK:
async with aiofiles.open(CONFIG_FILE, "w", encoding="utf-8") as file:
await file.write(json.dumps(CONFIG, indent=2, ensure_ascii=False) + "\n")
async def append_message_log(path, text):
async with aiofiles.open(path, "a", encoding="utf-8") as file:
await file.write(text)
async def append_metadata(path, data):
async with aiofiles.open(path, "a", encoding="utf-8") as file:
await file.write(json.dumps(data, ensure_ascii=False) + "\n")
async def resolve_channels():
MONITORED_CHANNEL_IDS.clear()
CHANNEL_ENTITY_MAP.clear()
CHANNEL_OBJECT_MAP.clear()
overrides = CONFIG.get("channel_overrides", {})
for channel in CHANNELS:
channel_key = str(parse_channel_id(channel))
if not overrides.get(channel_key, {}).get("turnon", True):
logger.info("[RESOLVE] Skipping disabled channel %s", channel)
continue
try:
entity = await client.get_entity(parse_channel_id(channel))
entity_id = str(utils.get_peer_id(entity))
title = getattr(entity, "title", entity_id)
MONITORED_CHANNEL_IDS.add(entity_id)
CHANNEL_ENTITY_MAP[entity_id] = title
CHANNEL_OBJECT_MAP[entity_id] = entity
await reset_channel_fail(entity_id)
except Exception as error:
fail_count = await increment_channel_fail(channel_key, str(error))
logger.error("[RESOLVE ERROR] Failed to resolve channel %r: %s", channel, error)
if CHANNEL_AUTO_DISABLE_AFTER > 0 and fail_count >= CHANNEL_AUTO_DISABLE_AFTER:
logger.warning("[CHANNEL DISABLED] %r failed %d consecutive resolutions", channel, fail_count)
CONFIG.setdefault("channel_overrides", {}).setdefault(channel_key, {})["turnon"] = False
await save_config()
def apply_channel_overrides():
overrides = CONFIG.get("channel_overrides", {})
for entity_id in list(MONITORED_CHANNEL_IDS):
if not overrides.get(entity_id, {}).get("turnon", True):
MONITORED_CHANNEL_IDS.discard(entity_id)
CHANNEL_ENTITY_MAP.pop(entity_id, None)
CHANNEL_OBJECT_MAP.pop(entity_id, None)
logger.info("[OVERRIDE] Disabled channel %s", entity_id)
async def reload_config():
global CONFIG, CHANNELS, MIN_DISK_SPACE_GB, MIN_DISK_SPACE, STATUS_PORT, RESYNC_INTERVAL
global MAX_LIFETIME_RETRIES, CHANNEL_AUTO_DISABLE_AFTER, MEDIA_RECORD_TTL_DAYS, RETRY_DROP_LOG
try:
async with CONFIG_LOCK:
async with aiofiles.open(CONFIG_FILE, "r", encoding="utf-8") as file:
updated = json.loads(await file.read())
validate_config(updated)
manual_changed = updated.get("manual_downloads", {}) != CONFIG.get("manual_downloads", {})
CONFIG = updated
CHANNELS = CONFIG["channels"]
MIN_DISK_SPACE_GB = CONFIG.get("min_disk_space_gb", 6)
MIN_DISK_SPACE = MIN_DISK_SPACE_GB * 1024**3
STATUS_PORT = CONFIG.get("status_port", 0)
RESYNC_INTERVAL = CONFIG.get("resync_interval_minutes", 60)
MAX_LIFETIME_RETRIES = CONFIG.get("max_lifetime_retries", 20)
CHANNEL_AUTO_DISABLE_AFTER = CONFIG.get("channel_auto_disable_after", 5)
MEDIA_RECORD_TTL_DAYS = CONFIG.get("media_record_ttl_days", 90)
RETRY_DROP_LOG = CONFIG.get("retry_drop_log", True)
apply_channel_overrides()
await resolve_channels()
if manual_changed:
await process_manual_downloads()
logger.info("[CONFIG] Reloaded")
except Exception as error:
logger.error("[CONFIG] Reload failed: %s", error)
raise
async def config_watcher():
try:
last_mtime = CONFIG_FILE.stat().st_mtime_ns
except OSError:
last_mtime = 0
while not shutdown_event.is_set():
try:
await asyncio.sleep(10)
mtime = CONFIG_FILE.stat().st_mtime_ns
if mtime != last_mtime:
last_mtime = mtime
await reload_config()
except asyncio.CancelledError:
break
except Exception as error:
logger.error("[CONFIG WATCHER] %s", error)
async def remove_manual_download(channel_id, message_id):
entries = CONFIG.get("manual_downloads", {}).get(str(channel_id), [])
normalized = [item for item in entries if str(item) != str(message_id)]
if len(normalized) != len(entries):
CONFIG["manual_downloads"][str(channel_id)] = normalized
await save_config()
async def process_manual_downloads():
for raw_channel_id, message_ids in CONFIG.get("manual_downloads", {}).items():
try:
entity = await client.get_entity(parse_channel_id(raw_channel_id))
entity_id = str(utils.get_peer_id(entity))
title = getattr(entity, "title", entity_id)
for message_id in message_ids:
key = f"{entity_id}:{message_id}"
if key in MANUAL_IN_PROGRESS:
continue
message = await client.get_messages(entity, ids=int(message_id))
if not message:
logger.warning("[MANUAL] Message %s not found in %s", message_id, raw_channel_id)
continue
MANUAL_IN_PROGRESS.add(key)
await priority_download_queue.put((message, title, entity_id, True, raw_channel_id))
except Exception as error:
logger.error("[MANUAL] Could not queue channel %s: %s", raw_channel_id, error)
async def download_media(message, paths, channel_id):
if message.out or not (media_type := detect_media_type(message)) or await media_exists(channel_id, message.id):
return
if not download_allowed.is_set():
logger.info("[PAUSED] Waiting for disk space: message %s", message.id)
await download_allowed.wait()
if shutdown_event.is_set():
return
entity = CHANNEL_OBJECT_MAP.get(channel_id)
if not entity:
return
file_name = sanitize_filename((message.file.name if message.file else None) or
f"{message.id}{(message.file.ext if message.file else '') or ''}")
final_name = f"{message.id}_{file_name}"
target_path = paths[media_type] / final_name
temp_path = paths[media_type] / f"{final_name}.part"
if target_path.exists():
await add_media_record(channel_id, message.id, final_name, target_path.stat().st_size)
return
last_error = "Unknown error"
for attempt in range(1, DOWNLOAD_RETRY + 1):
if shutdown_event.is_set() or not client.is_connected():
break
try:
temp_path.unlink(missing_ok=True)
fresh = await client.get_messages(entity, ids=message.id)
if not fresh:
raise RuntimeError(f"Message {message.id} not found")
await asyncio.wait_for(fresh.download_media(file=temp_path), timeout=DOWNLOAD_TIMEOUT)
if not temp_path.exists() or temp_path.stat().st_size <= 0:
raise RuntimeError("Download is missing or empty")
size = temp_path.stat().st_size
os.replace(temp_path, target_path)
await add_media_record(channel_id, message.id, final_name, size)
await db.execute("DELETE FROM failed_downloads WHERE channel_id=? AND message_id=?", (channel_id, message.id))
await db.commit()
logger.info("[DOWNLOAD OK] %s (%d bytes)", final_name, size)
return
except Exception as error:
last_error = str(error)
logger.warning("[DOWNLOAD ERROR] %s attempt %d/%d: %s", final_name, attempt, DOWNLOAD_RETRY, error)
finally:
if temp_path.exists() and temp_path.stat().st_size == 0:
temp_path.unlink(missing_ok=True)
await asyncio.sleep(min(5 * attempt, 5))
if not shutdown_event.is_set():
retry_count = await add_failed_download(channel_id, message.id, last_error)
logger.error("[FAILED] %s after %d attempts (lifetime %d)", final_name, DOWNLOAD_RETRY, retry_count)
if MAX_LIFETIME_RETRIES > 0 and retry_count >= MAX_LIFETIME_RETRIES:
logger.error("[DROP] %s exceeded %d lifetime retries", final_name, MAX_LIFETIME_RETRIES)
if RETRY_DROP_LOG:
await append_metadata(LOG_DIR / "dropped_downloads.jsonl", {
"timestamp": datetime.now(timezone.utc).isoformat(), "channel_id": channel_id,
"message_id": message.id, "file_name": final_name, "error": last_error,
"retry_count": retry_count,
})
await db.execute("DELETE FROM failed_downloads WHERE channel_id=? AND message_id=?", (channel_id, message.id))
await db.commit()
async def process_message(message, entity_title, channel_id):
paths = get_channel_paths(entity_title)
sender = await message.get_sender()
first = getattr(sender, "first_name", "") or "" if sender else ""
last = getattr(sender, "last_name", "") or "" if sender else ""
sender_name = f"{first} {last}".strip() or "Unknown"
username = getattr(sender, "username", "") or "" if sender else ""
timestamp = message.date.astimezone(MY_TZ).strftime("%Y-%m-%d %I:%M:%S %p")
if message.message and not await text_logged(channel_id, message.id):
media_name = None
media_size = None
media_info = ""
if message.media and message.file:
media_name = sanitize_filename(message.file.name or f"{message.id}{message.file.ext or ''}")
media_size = getattr(message.file, "size", 0) or 0
media_info = f"\n\n[Media]\nFilename: {media_name}\nSize: {media_size / 1024**2:.3f} MB"
text = (f"[{timestamp}]\nSender Name: {sender_name}\nUsername: {username}\nMessage ID: {message.id}"
f"{media_info}\n\n{message.message}\n{'-' * 50}\n")
await append_message_log(paths["messages"], text)
await append_metadata(paths["metadata"], {"timestamp": timestamp, "message_id": message.id,
"sender_name": sender_name, "username": username, "text": message.message,
"media_filename": media_name, "media_size_bytes": media_size})
await add_logged_message(channel_id, message.id)
if message.media:
await download_media(message, paths, channel_id)
await update_last_message_id(channel_id, message.id)
async def disk_monitor():
last_warn_time = 0.0
while not shutdown_event.is_set():
try:
free = shutil.disk_usage(BASE_DIR).free
if free < MIN_DISK_SPACE:
download_allowed.clear()
now = asyncio.get_running_loop().time()
if now - last_warn_time > 300:
logger.warning("[DISK] Critical: %.2f GB free (threshold %s GB). Downloads paused automatically.",
free / 1024**3, MIN_DISK_SPACE_GB)
last_warn_time = now
elif not download_allowed.is_set():
logger.info("[DISK] Space recovered: %.2f GB free. Resuming downloads.", free / 1024**3)
download_allowed.set()
except asyncio.CancelledError:
break
except Exception as error:
logger.error("[DISK MONITOR] %s", error)
await asyncio.sleep(30)
async def worker(worker_id):
global worker_busy_count
logger.info("[WORKER %d] Started", worker_id)
while not (shutdown_event.is_set() and download_queue.empty() and priority_download_queue.empty()):
queue_used = None
task = None
item = None
try:
try:
item = priority_download_queue.get_nowait()
queue_used = priority_download_queue
except asyncio.QueueEmpty:
item = await asyncio.wait_for(download_queue.get(), timeout=1)
queue_used = download_queue
message, title, channel_id, is_manual, raw_channel_id = item
worker_busy_count += 1
task = asyncio.create_task(process_message(message, title, channel_id))
active_tasks.add(task)
await task
if is_manual:
await remove_manual_download(raw_channel_id, message.id)
MANUAL_IN_PROGRESS.discard(f"{channel_id}:{message.id}")
except asyncio.TimeoutError:
continue
except asyncio.CancelledError:
break
except Exception:
logger.exception("[WORKER %d]", worker_id)
finally:
if item is not None:
worker_busy_count = max(0, worker_busy_count - 1)
if task:
active_tasks.discard(task)
if queue_used:
queue_used.task_done()
logger.info("[WORKER %d] Exited", worker_id)
@client.on(events.NewMessage)
async def new_message_handler(event):
channel_id = str(utils.get_peer_id(event.message.peer_id))
if channel_id not in MONITORED_CHANNEL_IDS:
return
try:
await download_queue.put((event.message, CHANNEL_ENTITY_MAP[channel_id], channel_id, False, None))
except asyncio.CancelledError:
raise
except Exception:
logger.exception("[NEW MESSAGE] Could not queue message %s", event.message.id)
async def initial_sync():
active_syncs = []
try:
for entity_id, entity in CHANNEL_OBJECT_MAP.items():
last_id = await get_last_message_id(entity_id)
active_syncs.append({"entity_id": entity_id, "title": CHANNEL_ENTITY_MAP[entity_id],
"iterator": client.iter_messages(entity, min_id=max(0, last_id - 5), reverse=True).__aiter__(), "count": 0})
while active_syncs and not shutdown_event.is_set() and client.is_connected():
for sync in list(active_syncs):
exhausted = False
for _ in range(15):
try:
message = await sync["iterator"].__anext__()
await download_queue.put((message, sync["title"], sync["entity_id"], False, None))
sync["count"] += 1
except StopAsyncIteration:
exhausted = True
break
except Exception as error:
logger.error("[SYNC] %s: %s", sync["title"], error)
exhausted = True
break
if exhausted:
logger.info("[SYNC DONE] %s queued %d", sync["title"], sync["count"])
active_syncs.remove(sync)
await asyncio.sleep(0.01)
except asyncio.CancelledError:
pass
except Exception:
logger.exception("[SYNC CRITICAL ERROR]")
async def cleanup_part_files():
count = 0
for part in BASE_DIR.rglob("*.part"):
try:
part.unlink()
count += 1
except Exception as error:
logger.warning("[STARTUP CLEANUP] Could not delete %s: %s", part, error)
if count:
logger.info("[STARTUP CLEANUP] Removed %d stale .part files.", count)
async def db_cleanup():
if MEDIA_RECORD_TTL_DAYS > 0:
cutoff = (datetime.now(timezone.utc) - timedelta(days=MEDIA_RECORD_TTL_DAYS)).isoformat()
cur = await db.execute("DELETE FROM downloaded_media WHERE downloaded_at < ?", (cutoff,))
pruned = cur.rowcount
await db.commit()
if pruned:
logger.info("[DB CLEANUP] Pruned %d expired media records (TTL=%dd).", pruned, MEDIA_RECORD_TTL_DAYS)
await db.execute("VACUUM")
await db.commit()
logger.info("[DB CLEANUP] VACUUM done.")
async def periodic_resync():
if RESYNC_INTERVAL <= 0:
return
while not shutdown_event.is_set():
try:
await asyncio.sleep(RESYNC_INTERVAL * 60)
except asyncio.CancelledError:
break
if not shutdown_event.is_set():
logger.info("[RESYNC] Starting periodic re-sync...")
await initial_sync()
logger.info("[RESYNC] Periodic re-sync complete.")
async def connection_supervisor():
global sync_task
backoff = [5, 10, 30, 60, 120, 300]
attempt = 0
while not shutdown_event.is_set():
try:
logger.info("[SUPERVISOR] Monitoring connection...")
await client.run_until_disconnected()
if shutdown_event.is_set():
break
logger.warning("[SUPERVISOR] Disconnected unexpectedly.")
except asyncio.CancelledError:
break
except Exception as error:
logger.error("[SUPERVISOR] Connection error: %s", error)
delay = backoff[min(attempt, len(backoff) - 1)]
logger.info("[SUPERVISOR] Reconnecting in %ds (attempt %d)...", delay, attempt + 1)
await asyncio.sleep(delay)
attempt += 1
try:
await client.connect()
await resolve_channels()
if sync_task and not sync_task.done():
sync_task.cancel()
await asyncio.gather(sync_task, return_exceptions=True)
sync_task = asyncio.create_task(initial_sync())
attempt = 0
logger.info("[SUPERVISOR] Reconnected and re-syncing.")
except Exception as error:
logger.error("[SUPERVISOR] Reconnect attempt failed: %s", error)
async def status_server():
try:
from aiohttp import web
except ImportError:
logger.error("[STATUS] aiohttp is required when status_port is enabled")
return
async def status(_request):
free = shutil.disk_usage(BASE_DIR).free / 1024**3
async with db.execute("SELECT COUNT(*) FROM failed_downloads") as cur:
failed = await cur.fetchone()
channels = {}
for channel_id, title in CHANNEL_ENTITY_MAP.items():
channels[channel_id] = {"title": title, "enabled": channel_id in MONITORED_CHANNEL_IDS,
"fail_count": await get_channel_fail_count(channel_id)}
db_size = Path(DB_PATH).stat().st_size / 1024**2 if Path(DB_PATH).exists() else 0
return web.json_response({"uptime_seconds": int((datetime.now(timezone.utc) - START_TIME).total_seconds()),
"connected": client.is_connected(), "downloads_paused": not download_allowed.is_set(),
"workers": {"total": len(worker_tasks), "busy": worker_busy_count},
"queue": {"normal": download_queue.qsize(), "priority": priority_download_queue.qsize()},
"disk_free_gb": round(free, 2), "failed_downloads_count": failed[0],
"db_size_mb": round(db_size, 2), "channels": channels})
async def logs(request):
try:
lines = max(1, min(int(request.query.get("lines", "100")), 5000))
except ValueError:
lines = 100
try:
content = (LOG_DIR / "app.log").read_text(encoding="utf-8").splitlines()
return web.Response(text="\n".join(content[-lines:]) + ("\n" if content else ""), content_type="text/plain")
except FileNotFoundError:
return web.Response(text="", content_type="text/plain")
async def reload_endpoint(_request):
await reload_config()
return web.json_response({"status": "reloaded"})
async def cleanup_endpoint(_request):
await db_cleanup()
return web.json_response({"status": "cleaned"})
async def set_channel(request):
channel_id, enabled = request.match_info["id"], request.match_info["action"] == "enable"
CONFIG.setdefault("channel_overrides", {}).setdefault(channel_id, {})["turnon"] = enabled
await save_config()
if enabled:
await resolve_channels()
else:
apply_channel_overrides()
return web.json_response({"status": "enabled" if enabled else "disabled", "channel_id": channel_id})
app = web.Application()
app.router.add_get("/health", lambda _request: web.json_response({"status": "ok"}))
app.router.add_get("/status", status)
app.router.add_get("/logs", logs)
app.router.add_post("/reload", reload_endpoint)
app.router.add_post("/db/cleanup", cleanup_endpoint)
app.router.add_post("/channel/{id}/{action:enable|disable}", set_channel)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "0.0.0.0", STATUS_PORT)
await site.start()
logger.info("[STATUS] Listening on port %d", STATUS_PORT)
try:
await shutdown_event.wait()
finally:
await runner.cleanup()
async def shutdown():
async with shutdown_lock:
if shutdown_event.is_set():
return
shutdown_event.set()
logger.info("[SHUTDOWN] Graceful shutdown started")
for task in (config_task, sync_task, disk_task, resync_task, status_task, supervisor_task):
if task and not task.done():
task.cancel()
if client.is_connected():
try:
await asyncio.wait_for(client.disconnect(), timeout=10)
except Exception:
logger.exception("[DISCONNECT ERROR]")
for queue in (download_queue, priority_download_queue):
while not queue.empty():
try:
queue.get_nowait()
queue.task_done()
except asyncio.QueueEmpty:
break
if active_tasks:
await asyncio.gather(*active_tasks, return_exceptions=True)
for task in worker_tasks:
task.cancel()
await asyncio.gather(*worker_tasks, return_exceptions=True)
if db:
await db.commit()
await db.close()
logger.info("[SHUTDOWN COMPLETE]")
async def main():
global config_task, sync_task, disk_task, resync_task, status_task, supervisor_task, download_allowed
validate_config(CONFIG)
download_allowed = asyncio.Event()
download_allowed.set()
await init_db()
await db_cleanup()
await cleanup_part_files()
await client.start(phone=PHONE)
me = await client.get_me()
logger.info("[CONNECTED] Logged in as %s", me.first_name)
await resolve_channels()
await process_manual_downloads()
for index in range(PARALLEL_DOWNLOADS):
worker_tasks.append(asyncio.create_task(worker(index + 1)))
disk_task = asyncio.create_task(disk_monitor())
sync_task = asyncio.create_task(initial_sync())
config_task = asyncio.create_task(config_watcher())
supervisor_task = asyncio.create_task(connection_supervisor())
resync_task = asyncio.create_task(periodic_resync())
if STATUS_PORT > 0:
status_task = asyncio.create_task(status_server())
logger.info("[READY] Workers: %d, Status port: %s", PARALLEL_DOWNLOADS, STATUS_PORT or "disabled")
while not shutdown_event.is_set():
await asyncio.sleep(5)
def handle_exit(*_args):
loop.call_soon_threadsafe(lambda: asyncio.create_task(shutdown()))
if __name__ == "__main__":
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
signal.signal(signal.SIGINT, handle_exit)
if hasattr(signal, "SIGTERM"):
signal.signal(signal.SIGTERM, handle_exit)
try:
loop.run_until_complete(main())
finally:
loop.run_until_complete(shutdown())
loop.close()

4
requirements.txt Normal file
View File

@@ -0,0 +1,4 @@
telethon[cryptg]>=1.30.0
aiosqlite>=0.19.0
aiofiles>=25.1.0
aiohttp>=3.9.0