Import source: telebot

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
This commit is contained in:
arsam
2026-09-18 23:14:27 +03:30
parent 03b7c2696c
commit f5d7ae9646
+675
View File
@@ -0,0 +1,675 @@
import os
import re
import json
from pathlib import Path
from typing import Optional
from dotenv import load_dotenv
from pyrogram import Client, filters
from pyrogram.types import Message
import asyncio
import time
from urllib.parse import urlparse
from pyrogram import idle
load_dotenv()
API_ID = int(os.getenv("API_ID", "0"))
API_HASH = os.getenv("API_HASH", "").strip()
BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip()
BASE_DIR = Path(__file__).resolve().parent
DOWNLOAD_DIR = BASE_DIR / "downloads"
QUEUE_DIR = BASE_DIR / "queue"
QUEUE_FILE = QUEUE_DIR / "tasks.jsonl"
STATUS_FILE = QUEUE_DIR / "status.jsonl"
SETTINGS_FILE = QUEUE_DIR / "settings.json"
DELETED_FILE = QUEUE_DIR / "deleted.jsonl"
CANCEL_FILE = QUEUE_DIR / "cancelled.jsonl"
PROCESSING_FILE = QUEUE_DIR / "processing.json"
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
QUEUE_DIR.mkdir(parents=True, exist_ok=True)
if not API_ID or not API_HASH or not BOT_TOKEN:
raise RuntimeError("Please set API_ID, API_HASH and BOT_TOKEN in .env")
app = Client(
"tel2rub",
api_id=API_ID,
api_hash=API_HASH,
bot_token=BOT_TOKEN,
)
def safe_filename(name: Optional[str]) -> str:
name = (name or "file.bin").strip()
name = re.sub(r'[<>:"/\\|?*\x00-\x1F]', "_", name)
name = name.rstrip(". ")
return name[:200] or "file.bin"
def split_name(filename: str) -> tuple[str, str]:
path = Path(filename)
return path.stem, path.suffix
def get_media(message: Message):
media_types = [
("document", message.document),
("video", message.video),
("audio", message.audio),
("voice", message.voice),
("photo", message.photo),
("animation", message.animation),
("video_note", message.video_note),
("sticker", message.sticker),
]
for media_type, media in media_types:
if media:
return media_type, media
return None, None
def build_download_filename(message: Message, media_type: str, media) -> str:
original_name = getattr(media, "file_name", None)
if not original_name:
file_unique_id = getattr(media, "file_unique_id", None) or "file"
default_extensions = {
"document": ".bin",
"video": ".mp4",
"audio": ".mp3",
"voice": ".ogg",
"photo": ".jpg",
"animation": ".mp4",
"video_note": ".mp4",
"sticker": ".webp",
}
original_name = f"{file_unique_id}{default_extensions.get(media_type, '.bin')}"
original_name = safe_filename(original_name)
stem, suffix = split_name(original_name)
unique_name = f"{stem}_{message.id}{suffix or '.bin'}"
return safe_filename(unique_name)
waiting_for_zip_password = False
class QueueManager:
def __init__(self):
self._cache = None
self._mtime = 0
def all(self):
mtime = QUEUE_FILE.stat().st_mtime if QUEUE_FILE.exists() else 0
if mtime == self._mtime and self._cache is not None:
return self._cache
self._cache = []
if QUEUE_FILE.exists():
with open(QUEUE_FILE, "r", encoding="utf-8") as f:
self._cache = [json.loads(l) for l in f if l.strip()]
self._mtime = mtime
return self._cache
def push(self, task):
task.setdefault("job_id", str(int(time.time() * 1000)))
with open(QUEUE_FILE, "a", encoding="utf-8") as f:
f.write(json.dumps(task, ensure_ascii=False) + "\n")
self._cache = None
def remove(self, job_id=None, message_id=None):
tasks = self.all()
kept, removed = [], None
for t in tasks:
if (job_id and str(t.get("job_id")) == str(job_id)) or \
(message_id and int(t.get("status_message_id", 0)) == message_id):
removed = t
else:
kept.append(t)
if removed:
with open(QUEUE_FILE, "w", encoding="utf-8") as f:
f.writelines(json.dumps(t, ensure_ascii=False) + "\n" for t in kept)
self._cache = None
return removed
queue = QueueManager()
def mark_deleted(task: dict):
with open(DELETED_FILE, "a", encoding="utf-8") as f:
f.write(json.dumps(task, ensure_ascii=False) + "\n")
def mark_cancelled(task: dict):
with open(CANCEL_FILE, "a", encoding="utf-8") as f:
f.write(json.dumps(task, ensure_ascii=False) + "\n")
def cancel_job(job_id: str):
with open(CANCEL_FILE, "a", encoding="utf-8") as f:
f.write(json.dumps({"job_id": str(job_id)}, ensure_ascii=False) + "\n")
def current_processing() -> dict | None:
try:
if PROCESSING_FILE.exists():
return json.loads(PROCESSING_FILE.read_text(encoding="utf-8"))
except Exception:
pass
return None
def task_label(task: dict) -> str:
if task.get("type") == "direct_url":
return task.get("url", "direct_url")
return task.get("file_name") or Path(task.get("path", "file")).name
def queue_position(job_id: str) -> int | None:
for index, item in enumerate(queue.all(), start=1):
if str(item.get("job_id")) == str(job_id):
return index
return None
def was_deleted(job_id=None, message_id=None) -> bool:
if not DELETED_FILE.exists():
return False
with open(DELETED_FILE, "r", encoding="utf-8") as f:
for line in f:
if not line.strip():
continue
item = json.loads(line)
if job_id and str(item.get("job_id")) == str(job_id):
return True
if message_id and int(item.get("status_message_id", 0)) == message_id:
return True
return False
def load_settings() -> dict:
try:
if SETTINGS_FILE.exists():
return json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
except Exception:
pass
return {"safe_mode": False, "zip_password": ""}
def save_settings(data: dict):
SETTINGS_FILE.write_text(
json.dumps(data, ensure_ascii=False, indent=2),
encoding="utf-8"
)
def is_direct_url(text: str) -> bool:
if not text:
return False
url = extract_first_url(text)
if not url:
return False
try:
parsed = urlparse(url)
except Exception:
return False
return parsed.scheme in ("http", "https") and bool(parsed.netloc)
def extract_first_url(text: str) -> Optional[str]:
if not text:
return None
match = re.search(r"https?://\S+", text)
return match.group(0) if match else None
def progress_bar(percent: float, length: int = 12) -> str:
filled = int(length * percent / 100)
return "█" * filled + "░" * (length - filled)
def pretty_size(size) -> str:
size = float(size or 0)
units = ["B", "KB", "MB", "GB"]
index = 0
while size >= 1024 and index < len(units) - 1:
size /= 1024
index += 1
return f"{size:.2f} {units[index]}"
def eta_text(seconds) -> str:
if not seconds or seconds <= 0:
return "نامشخص"
seconds = int(seconds)
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h:
return f"{h}h {m}m"
if m:
return f"{m}m {s}s"
return f"{s}s"
async def download_progress(current, total, status_message, file_name, started_at, state):
now = time.time()
if now - state.get("last_update", 0) < 3 and current < total:
return
state["last_update"] = now
percent = current * 100 / total if total else 0
elapsed = max(now - started_at, 1)
speed = current / elapsed
eta = (total - current) / speed if speed else None
text = (
f"در حال دریافت فایل از تلگرام\n\n"
f"فایل: `{file_name}`\n"
f"حجم: `{pretty_size(total)}`\n"
f"پیشرفت: `{percent:.1f}%`\n"
f"`{progress_bar(percent)}`\n"
f"سرعت: `{pretty_size(speed)}/s`\n"
f"زمان باقی‌مانده: `{eta_text(eta)}`"
)
try:
await status_message.edit_text(text)
except Exception:
pass
async def status_watcher():
pos = 0
while True:
await asyncio.sleep(1)
if not STATUS_FILE.exists():
continue
try:
with open(STATUS_FILE, "r", encoding="utf-8") as f:
f.seek(pos)
lines = f.readlines()
pos = f.tell()
for line in lines:
if not line.strip():
continue
data = json.loads(line)
chat_id = data.get("chat_id")
msg_id = data.get("message_id")
text = data.get("text", "")
percent = data.get("percent")
if not chat_id or not msg_id:
continue
if percent is not None:
bar = progress_bar(float(percent))
if bar not in text:
text += f"\n\n`{bar}` `{float(percent):.1f}%`"
try:
await app.edit_message_text(chat_id, msg_id, text)
except Exception:
pass
except Exception:
pass
@app.on_message(filters.private & filters.command("start"))
async def start_handler(client: Client, message: Message):
await message.reply_text(
"سلام، به Coaxys Uploader خوش آمدی.\n\n"
"فایل تلگرام یا لینک مستقیم بفرست. ربات فایل را دریافت می‌کند، روی S3 آپلود می‌کند، "
"و در پایان یک لینک دانلود ۱۶۸ ساعته تحویل می‌دهد.\n\n"
"برای دانلود از یوتیوب، اینستاگرام و... از این ربات استفاده کن: @Gozilla_bot\n"
"بعد خروجی را همین‌جا بفرست تا روی Coaxys ذخیره شود.\n\n"
"دستورات کاربردی:\n"
"`/status` وضعیت صف و فایل در حال پردازش\n"
"`/del <job_id>` لغو یا حذف یک مورد\n"
"`/delall` پاکسازی صف\n"
"`/safemode on` ساخت ZIP رمزدار قبل از آپلود\n"
"`/safemode off` آپلود عادی\n\n"
"برای فایل‌های حجیم، Safe Mode را خاموش نگه دار تا آپلود سریع‌تر انجام شود.\n\n"
"@Coaxys"
)
@app.on_message(filters.private & filters.command("safemode"))
async def safemode_handler(client: Client, message: Message):
global waiting_for_zip_password
args = message.text.split(maxsplit=1)
if len(args) < 2:
await message.reply_text("برای تغییر وضعیت Safe Mode از `/safemode on` یا `/safemode off` استفاده کن.")
return
action = args[1].strip().lower()
settings = load_settings()
if action == "on":
settings["safe_mode"] = True
save_settings(settings)
waiting_for_zip_password = True
await message.reply_text(
"Safe Mode فعال شد.\n\n"
"رمز ZIP را در پیام بعدی بفرست. از این به بعد فایل‌ها قبل از آپلود با همین رمز ZIP می‌شوند."
)
return
if action == "off":
settings["safe_mode"] = False
settings["zip_password"] = ""
save_settings(settings)
waiting_for_zip_password = False
await message.reply_text(
"Safe Mode غیرفعال شد.\n\n"
"از این به بعد فایل‌ها بدون ZIP رمزدار آپلود می‌شوند."
)
return
await message.reply_text("دستور نامعتبر است. از `/safemode on` یا `/safemode off` استفاده کن.")
@app.on_message(filters.private & filters.command("delall"))
async def clear_queue_handler(client: Client, message: Message):
tasks = queue.all()
processing = current_processing()
if not tasks and not processing:
await message.reply_text("صف خالی است.")
return
if processing:
cancel_job(str(processing.get("job_id")))
mark_cancelled(processing)
try:
await client.edit_message_text(
chat_id=processing["chat_id"],
message_id=processing["status_message_id"],
text="لغو برای فایل در حال پردازش ثبت شد. اگر آپلود شروع شده باشد، چند لحظه زمان می‌برد."
)
except Exception:
pass
for task in tasks:
mark_deleted(task)
old_path = task.get("path")
if old_path:
try:
path = Path(old_path)
if path.exists():
path.unlink()
except Exception:
pass
try:
await client.edit_message_text(
chat_id=task["chat_id"],
message_id=task["status_message_id"],
text="این مورد از صف حذف شد."
)
except Exception:
pass
with open(QUEUE_FILE, "w", encoding="utf-8") as f:
pass
queue._cache = None
queue._mtime = 0
if processing:
await message.reply_text("تمام موارد صف پاک شد و لغو فایل در حال پردازش هم ثبت شد.")
else:
await message.reply_text("صف پاک شد.")
@app.on_message(filters.private & filters.command("del"))
async def delete_one_handler(client: Client, message: Message):
job_id = None
reply_message_id = None
parts = message.text.split(maxsplit=1)
if len(parts) > 1:
job_id = parts[1].strip()
if message.reply_to_message:
reply_message_id = message.reply_to_message.id
tasks = queue.all()
if not tasks:
if job_id and was_deleted(job_id=job_id):
await message.reply_text("این مورد قبلاً از صف حذف شده است.")
return
if reply_message_id and was_deleted(message_id=reply_message_id):
await message.reply_text("این مورد قبلاً از صف حذف شده است.")
return
if job_id:
cancel_job(job_id)
await message.reply_text("لغو ثبت شد. اگر فایل در حال آپلود باشد، چند لحظه زمان می‌برد.")
return
await message.reply_text("موردی برای حذف در صف پیدا نشد.")
return
removed = queue.remove(job_id=job_id, message_id=reply_message_id)
if removed:
mark_deleted(removed)
old_path = removed.get("path")
if old_path:
try:
path = Path(old_path)
if path.exists():
path.unlink()
except Exception:
pass
try:
await client.edit_message_text(
chat_id=removed["chat_id"],
message_id=removed["status_message_id"],
text="این مورد از صف حذف شد."
)
except Exception:
pass
await message.reply_text("از صف حذف شد.")
return
if job_id and was_deleted(job_id=job_id):
await message.reply_text("این مورد قبلاً از صف حذف شده است.")
return
if reply_message_id and was_deleted(message_id=reply_message_id):
await message.reply_text("این مورد قبلاً از صف حذف شده است.")
return
if job_id:
cancel_job(job_id)
await message.reply_text("دستور لغو ثبت شد. اگر فایل در حال پردازش باشد، worker آن را متوقف می‌کند.")
return
@app.on_message(filters.private & filters.command("status"))
async def status_handler(client: Client, message: Message):
tasks = queue.all()
processing = current_processing()
if not tasks and not processing:
await message.reply_text("صف خالی است و فایلی در حال پردازش نیست.")
return
lines = ["وضعیت پردازش\n"]
if processing:
lines.append("در حال پردازش:")
lines.append(f"فایل: `{task_label(processing)}`")
lines.append(f"شناسه: `{processing.get('job_id')}`")
lines.append(f"لغو: `/del {processing.get('job_id')}`\n")
else:
lines.append("در حال پردازش: موردی نیست.\n")
if tasks:
lines.append(f"در صف: `{len(tasks)}` مورد")
for task in tasks[:10]:
position = queue_position(str(task.get("job_id"))) or "-"
size = task.get("file_size")
size_text = f" - {pretty_size(size)}" if size else ""
lines.append(
f"{position}. `{task_label(task)}`{size_text}\n"
f"شناسه: `{task.get('job_id')}`"
)
if len(tasks) > 10:
lines.append(f"... و `{len(tasks) - 10}` مورد دیگر")
else:
lines.append("در صف: موردی نیست.")
await message.reply_text("\n".join(lines))
@app.on_message(filters.private & filters.text & ~filters.command(["start", "safemode", "del", "delall", "status"]))
async def text_handler(client: Client, message: Message):
global waiting_for_zip_password
text = message.text or ""
if waiting_for_zip_password:
password = text.strip()
if not password:
await message.reply_text("رمز نمی‌تواند خالی باشد. لطفاً یک رمز معتبر ارسال کنید.")
return
settings = load_settings()
settings["safe_mode"] = True
settings["zip_password"] = password
save_settings(settings)
waiting_for_zip_password = False
await message.reply_text(
"رمز ذخیره شد.\n\n"
"از این به بعد فایل‌ها قبل از آپلود به‌صورت ZIP رمزدار آماده می‌شوند."
)
return
url = extract_first_url(text)
if not url or not is_direct_url(url):
return
settings = load_settings()
status = await message.reply_text(
"لینک دریافت شد.\n\n"
"در صف دانلود و آپلود قرار گرفت."
)
task = {
"type": "direct_url",
"url": url,
"chat_id": message.chat.id,
"status_message_id": status.id,
"safe_mode": settings.get("safe_mode", False),
"zip_password": settings.get("zip_password", ""),
}
queue.push(task)
await status.edit_text(
f"لینک در صف قرار گرفت.\n\n"
f"شناسه: `{task['job_id']}`\n"
f"لغو یا حذف: `/del {task['job_id']}`"
)
@app.on_message(
filters.private
& (
filters.document
| filters.video
| filters.audio
| filters.voice
| filters.photo
| filters.animation
| filters.video_note
| filters.sticker
)
)
async def media_handler(client: Client, message: Message):
media_type, media = get_media(message)
if not media:
await message.reply_text("فایل قابل پردازش نیست.")
return
download_name = build_download_filename(message, media_type, media)
download_path = DOWNLOAD_DIR / download_name
status = await message.reply_text(
"فایل دریافت شد.\n\n"
"در حال آماده‌سازی دانلود از تلگرام..."
)
try:
started_at = time.time()
progress_state = {"last_update": 0}
downloaded = await client.download_media(
message,
file_name=str(download_path),
progress=download_progress,
progress_args=(status, download_name, started_at, progress_state),
)
if not downloaded:
raise RuntimeError("Download failed.")
downloaded_path = Path(downloaded)
if not downloaded_path.exists():
raise RuntimeError("Downloaded file not found.")
file_size = downloaded_path.stat().st_size
settings = load_settings()
task = {
"type": "local_file",
"path": str(downloaded_path),
"caption": message.caption or "",
"chat_id": message.chat.id,
"status_message_id": status.id,
"file_name": download_name,
"file_size": file_size,
"safe_mode": settings.get("safe_mode", False),
"zip_password": settings.get("zip_password", ""),
}
queue.push(task)
await status.edit_text(
f"فایل در صف آپلود قرار گرفت.\n\n"
f"فایل: `{download_name}`\n"
f"حجم: `{pretty_size(file_size)}`\n"
f"شناسه: `{task['job_id']}`\n\n"
f"لغو یا حذف: `/del {task['job_id']}`"
)
except Exception as e:
await status.edit_text(f"خطا در دریافت فایل:\n{str(e)}")
def clear_old_status():
try:
if STATUS_FILE.exists():
STATUS_FILE.unlink()
except Exception:
pass
if __name__ == "__main__":
clear_old_status()
app.start()
app.loop.create_task(status_watcher())
idle()
app.stop()