#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
=============================================================================
ورکر اختصاصی دانلود ویدیوهای بالای ۲۰ مگابایت تلگرام با اکانت کاربر (Telethon MTProto)
طراحی شده برای هاست‌های سی‌پنل اشتراکی و سرورهای لینوکس / مک
=============================================================================
شناسه برنامه (API ID): 26297341
هش برنامه (API HASH): 321267735440187bdd1f54fcb1f25612
آیدی عددی اکانت دانلودر: 7537792690
"""
import sys
import os
import re
import time
import json
import asyncio
# پلی‌فیل‌های سازگاری کامل پایتون ۳.۶ با متدهای ناهمگام پایتون ۳.۷+
if not hasattr(asyncio, "get_running_loop"):
def _compat_get_running_loop():
try:
return asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop
asyncio.get_running_loop = _compat_get_running_loop
if not hasattr(asyncio, "create_task"):
def _compat_create_task(coro):
loop = asyncio.get_event_loop()
return loop.create_task(coro)
asyncio.create_task = _compat_create_task
if not hasattr(asyncio, "run"):
def _compat_run(coro):
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop.run_until_complete(coro)
asyncio.run = _compat_run
if not hasattr(asyncio, "all_tasks"):
def _compat_all_tasks(loop=None):
return asyncio.Task.all_tasks(loop=loop)
asyncio.all_tasks = _compat_all_tasks
import argparse
import tempfile
import urllib.request
import urllib.parse
from pathlib import Path
try:
import fcntl
except ImportError:
fcntl = None
# پشتیبانی از ماژول‌های مستقل محلی در صورت عدم دسترسی به ترمینال هاست
LIBS_DIR = Path(__file__).resolve().parent / "libs"
if LIBS_DIR.exists():
sys.path.insert(0, str(LIBS_DIR))
# کتابخانه قدرتمند تلتون
try:
from telethon import TelegramClient, events
from telethon.sessions import StringSession
from telethon.tl.types import DocumentAttributeVideo, DocumentAttributeFilename
except ImportError as err:
print(f"❌ خطا در بارگذاری ماژول‌های تلگرام: {err}")
sys.exit(1)
# موتور دانلود چند اتصاله و موازی FastTelethon (ارتقای سرعت تا ۱۰ الی ۴۰ مگابایت بر ثانیه)
try:
from fast_telethon import download_file as fast_download_file
except ImportError:
try:
from libs.fast_telethon import download_file as fast_download_file
except ImportError:
fast_download_file = None
# مسیرهای پروژه
BASE_DIR = Path(__file__).resolve().parent
CONFIG_FILE = BASE_DIR / "config.php"
UPLOAD_DIR = BASE_DIR / "uploads" / "videos"
LOCAL_DATA_DIR = Path(__file__).resolve().parent / "data"
LOCAL_QUEUE_DIR = LOCAL_DATA_DIR / "queue"
LOCAL_TASK_DIR = LOCAL_DATA_DIR / "tasks"
TEMP_QUEUE_DIR = Path(os.environ.get("TMPDIR", "/tmp")) / "tg_userbot_queue"
TEMP_TASK_DIR = Path(os.environ.get("TMPDIR", "/tmp")) / "tg_uploader_tasks"
SYS_TEMP_QUEUE_DIR = Path(tempfile.gettempdir()) / "tg_userbot_queue"
SYS_TEMP_TASK_DIR = Path(tempfile.gettempdir()) / "tg_uploader_tasks"
# اطمینان از وجود پوشه‌های محلی
LOCAL_QUEUE_DIR.mkdir(parents=True, exist_ok=True)
LOCAL_TASK_DIR.mkdir(parents=True, exist_ok=True)
# حداکثر عمر فایل صف (ثانیه). قبلاً ۲ ساعت بود و ویدیوهای آخرِ صف‌های طولانی بی‌صدا حذف می‌شدند.
QUEUE_FILE_MAX_AGE = 24 * 3600
def touch_worker_lock():
"""زنده نگه‌داشتن زمان تغییر فایل قفل تا PHP آن را قدیمی/منقضی فرض نکند و ورکر دوم راه نیندازد"""
try:
lp = LOCAL_DATA_DIR / "worker.lock"
if lp.exists():
os.utime(str(lp), None)
except Exception:
pass
class WorkerProcessLock:
"""قفل انحصاری پروسس ورکر برای صف‌بندی ترتیبی و جلوگیری از دانلود همزمان چند ویدیو"""
def __init__(self, lock_path=None):
self.lock_path = lock_path or (LOCAL_DATA_DIR / "worker.lock")
self.fp = None
def acquire(self):
try:
self.lock_path.parent.mkdir(parents=True, exist_ok=True)
self.fp = open(str(self.lock_path), "a+")
if fcntl:
fcntl.flock(self.fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
self.fp.seek(0)
self.fp.truncate()
self.fp.write(str(os.getpid()))
self.fp.flush()
return True
except (IOError, BlockingIOError, OSError):
if self.fp:
try:
self.fp.close()
except Exception:
pass
self.fp = None
return False
except Exception:
return True
COMPLETED_MSGS_FILE = LOCAL_DATA_DIR / "completed_msg_ids.json"
def load_completed_msg_ids():
"""واکشی شناسه‌های پیام‌های دانلود شده جهت جلوگیری از دانلود تکراری پارت‌های قبلی"""
try:
if COMPLETED_MSGS_FILE.exists():
return set(json.loads(COMPLETED_MSGS_FILE.read_text(encoding="utf-8")))
except Exception:
pass
return set()
def save_completed_msg_id(mid):
"""ثبت شناسه پیام تکمیل‌شده جهت اطمینان از عدم پردازش دوباره"""
try:
current = load_completed_msg_ids()
current.add(int(mid))
if len(current) > 500:
current = set(list(current)[-500:])
COMPLETED_MSGS_FILE.parent.mkdir(parents=True, exist_ok=True)
COMPLETED_MSGS_FILE.write_text(json.dumps(list(current)), encoding="utf-8")
except Exception:
pass
def release(self):
if self.fp:
try:
if fcntl:
fcntl.flock(self.fp.fileno(), fcntl.LOCK_UN)
self.fp.close()
except Exception:
pass
self.fp = None
def upload_file_ftp(local_path: Path, remote_filename: str, ftp_conf: dict, progress_cb=None, cancel_cb=None) -> bool:
"""انتقال مستقیم فایل دانلود شده به هاست اصلی با پروتکل پرسرعت FTP و گزارش لایو پیشرفت"""
import ftplib
host = ftp_conf.get("host", "94.130.44.82")
port = int(ftp_conf.get("port", 21))
user = ftp_conf.get("username", "")
passwd = ftp_conf.get("password", "")
use_ssl = bool(ftp_conf.get("ssl", False))
passive = bool(ftp_conf.get("passive", True))
remote_dir = ftp_conf.get("remote_dir", "/public_html/zahra/panel/uploads/videos")
total_bytes = local_path.stat().st_size
sent_bytes = 0
t_start = time.time()
last_cb_time = 0
last_cb_bytes = 0
print(f"🚀 در حال اتصال به FTP هاست اصلی ({host}:{port}) جهت ارسال فایل {remote_filename}...")
try:
if use_ssl:
ftp = ftplib.FTP_TLS(timeout=60)
else:
ftp = ftplib.FTP(timeout=60)
ftp.connect(host, port, timeout=30)
ftp.login(user, passwd)
if use_ssl and hasattr(ftp, 'prot_p'):
ftp.prot_p()
ftp.set_pasv(passive)
# اطمینان از وجود پوشه مقصد روی هاست اصلی
if remote_dir:
cleaned_dir = remote_dir.strip('/')
try:
ftp.cwd(cleaned_dir)
except Exception:
for part in cleaned_dir.split('/'):
if not part:
continue
try:
ftp.cwd(part)
except Exception:
try:
ftp.mkd(part)
ftp.cwd(part)
except Exception:
pass
blocksize = 128 * 1024
with open(local_path, 'rb') as f:
def callback(chunk):
nonlocal sent_bytes, last_cb_time, last_cb_bytes
if cancel_cb and cancel_cb():
raise Exception("FTP upload cancelled by user")
sent_bytes += len(chunk)
now = time.time()
if progress_cb and (now - last_cb_time >= 2.5 or sent_bytes >= total_bytes):
dt = now - last_cb_time
db = sent_bytes - last_cb_bytes
spd = int(db / dt) if dt > 0 else int(sent_bytes / (now - t_start + 0.001))
last_cb_time = now
last_cb_bytes = sent_bytes
progress_cb(sent_bytes, total_bytes, spd)
ftp.storbinary(f'STOR {remote_filename}', f, blocksize=blocksize, callback=callback)
try:
ftp.quit()
except Exception:
ftp.close()
print(f"✅ انتقال FTP فایل {remote_filename} به هاست اصلی با موفقیت تمام شد.")
return True
except Exception as e:
print(f"❌ خطای انتقال مستقیم FTP به هاست اصلی: {e}")
return False
# بارگذاری مقادیر از config.php
def load_php_config():
conf = {
"api_id": 26297341,
"api_hash": "321267735440187bdd1f54fcb1f25612",
"account_id": 7537792690,
"session_string": "",
"api_secret": "nima_ub_sec_321267735440187bdd1f54fcb1f25612",
"bot_token": "8900585918:AAEqoKYv7bUHRpGeaP2c4Qbiz_zZ9tTK2XA"
}
if CONFIG_FILE.exists():
try:
# روش ۱: از طریق خروجی php json_encode
import subprocess
out = subprocess.check_output(["php", "-r", f"echo json_encode(require('{CONFIG_FILE}'));"], text=True, stderr=subprocess.DEVNULL)
data = json.loads(out)
ub = data.get("userbot", {})
conf["api_id"] = int(ub.get("api_id", conf["api_id"]))
conf["api_hash"] = str(ub.get("api_hash", conf["api_hash"]))
conf["account_id"] = int(ub.get("account_id", conf["account_id"]))
conf["session_string"] = str(ub.get("session_string", ""))
conf["api_secret"] = str(ub.get("api_secret", conf["api_secret"]))
conf["bot_token"] = str(data.get("telegram_bot", {}).get("token", conf["bot_token"]))
conf["ftp"] = data.get("ftp", {})
return conf
except Exception:
pass
# روش ۲: تحلیل دستی فایل با عبارات باقاعده (Regex)
try:
content = CONFIG_FILE.read_text(encoding="utf-8")
m_id = re.search(r"'api_id'\s*=>\s*(\d+)", content)
if m_id: conf["api_id"] = int(m_id.group(1))
m_hash = re.search(r"'api_hash'\s*=>\s*['\"]([^'\"]+)['\"]", content)
if m_hash: conf["api_hash"] = m_hash.group(1)
m_acc = re.search(r"'account_id'\s*=>\s*(\d+)", content)
if m_acc: conf["account_id"] = int(m_acc.group(1))
m_sess = re.search(r"'session_string'\s*=>\s*['\"]([^'\"]*)['\"]", content)
if m_sess: conf["session_string"] = m_sess.group(1)
m_sec = re.search(r"'api_secret'\s*=>\s*['\"]([^'\"]+)['\"]", content)
if m_sec: conf["api_secret"] = m_sec.group(1)
except Exception:
pass
return conf
# ذخیره رشته سشن جدید در config.php
def save_session_string_to_config(session_str):
if not CONFIG_FILE.exists():
return False
try:
content = CONFIG_FILE.read_text(encoding="utf-8")
if "'session_string'" in content:
updated = re.sub(r"('session_string'\s*=>\s*['\"])[^'\"]*(['\"])", rf"\g<1>{session_str}\g<2>", content)
CONFIG_FILE.write_text(updated, encoding="utf-8")
return True
except Exception as e:
print("خطا در به‌روزرسانی config.php:", e)
return False
def find_php_binary():
import shutil
p = shutil.which("php")
if p:
return p
candidates = [
"/usr/local/bin/php",
"/usr/bin/php",
"/opt/cpanel/ea-php83/root/usr/bin/php",
"/opt/cpanel/ea-php82/root/usr/bin/php",
"/opt/cpanel/ea-php81/root/usr/bin/php",
"/opt/cpanel/ea-php80/root/usr/bin/php",
"/opt/cpanel/ea-php74/root/usr/bin/php"
]
for c in candidates:
if os.path.isfile(c) and os.access(c, os.X_OK):
return c
return "php"
def direct_telegram_api(method, params, bot_token=None):
"""ارسال مستقیم درخواست به API ربات تلگرام در صورت بروز مشکل در وب‌سرور لوکال"""
if not bot_token:
conf = load_php_config()
bot_token = conf.get("bot_token", "")
if not bot_token:
return None
try:
url = f"https://api.telegram.org/bot{bot_token}/{method}"
data = json.dumps(params).encode("utf-8")
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode("utf-8"))
except Exception:
return None
_PENDING_API_THREADS = []
def call_internal_api_async(action, payload):
"""ارسال ناهمگام درخواست به API در ترد پس‌زمینه جهت جلوگیری از قفل شدن ایونت‌لوپ دانلود"""
try:
import threading
# تردهای تمام‌شده را پاک کن تا لیست بی‌نهایت رشد نکند
_PENDING_API_THREADS[:] = [x for x in _PENDING_API_THREADS if x.is_alive()]
# جلوگیری از انباشت پروسه‌های موازی PHP در گزارش پیشرفت
if action == "update_progress" and len(_PENDING_API_THREADS) > 0:
return
t = threading.Thread(target=call_internal_api, args=(action, payload), daemon=True)
t.start()
_PENDING_API_THREADS.append(t)
except Exception:
pass
def wait_pending_api_calls(timeout=15):
"""صبر تا پایان آپدیت‌های درصد در حال ارسال؛ تا پیام قدیمی (۹۹٪) روی پیام موفقیت نشیند"""
deadline = time.time() + timeout
for t in list(_PENDING_API_THREADS):
remaining = deadline - time.time()
if remaining <= 0:
break
try:
t.join(remaining)
except Exception:
pass
# ارسال درخواست به API وب‌سایت (پشتیبانی سه‌گانه از PHP CLI هوشمند، درخواست وب، و تلگرام مستقیم)
def call_internal_api(action, payload):
conf = load_php_config()
payload["action"] = action
payload["secret"] = conf["api_secret"]
# روش ۱: اجرای مستقیم از طریق PHP CLI (سازگار با ۱۰۰٪ هاست‌های سی‌پنل)
api_script = BASE_DIR / "userbot_api.php"
if not api_script.exists():
api_script = BASE_DIR / "telegram_bot" / "userbot_api.php"
if api_script.exists():
try:
import subprocess
php_bin = find_php_binary()
proc = subprocess.run(
[php_bin, str(api_script), json.dumps(payload, ensure_ascii=False)],
capture_output=True,
text=True,
timeout=180 if action == "complete" else 30
)
if proc.stdout:
json_match = re.search(r"\{.*\}", proc.stdout, re.DOTALL)
if json_match:
return json.loads(json_match.group(0))
except Exception:
pass
# روش ۲: درخواست وب از طریق URL در صورت وجود دامنه یا وب‌سرور لوکال
site_url = payload.get("site_url") or conf.get("site_url", "")
urls_to_try = []
if site_url:
urls_to_try.append(site_url.rstrip("/") + "/telegram_bot/userbot_api.php")
urls_to_try.append("http://127.0.0.1/telegram_bot/userbot_api.php")
for url in urls_to_try:
try:
data = urllib.parse.urlencode(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, method="POST")
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode("utf-8"))
except Exception:
continue
# روش ۳ (پشتیبان نهایی): ارسال مستقیم به تلگرام در صورتی که اکشن مربوط به ادیت پیام کاربر باشد
chat_id = payload.get("chat_id")
msg_id = payload.get("status_msg_id") or payload.get("msg_id")
bot_token = conf.get("bot_token", "")
if action == "notify_start" and chat_id and msg_id:
v_title = payload.get("title", "ویدیو سرور")
t_msg = "⚡ نوبت دانلود این ویدیو در صف رسید!\n\n"
t_msg += f"╭─ اطلاعات تسک\n"
t_msg += f"│ نام ویدیو: {v_title}\n"
t_msg += "│ وضعیت: در حال اتصال به اکانت و دریافت استریم ویدیو روی سرور...\n"
t_msg += "│ پیشرفت: ▱▱▱▱▱▱▱▱▱▱ 0%\n"
t_msg += "╰─────────────────────────────\n"
t_msg += "فایل با سرعت بالا مستقیماً روی سرور دانلود خواهد شد."
tid = payload.get("task_id", "")
kb = {'inline_keyboard': [[{'text': 'به‌روزرسانی وضعیت', 'callback_data': f"refresh_dl_{tid}", 'style': 'primary'}, {'text': 'حذف از صف دانلود', 'callback_data': f"cancel_dl_{tid}", 'style': 'danger'}]]}
direct_telegram_api("editMessageText", {
"chat_id": chat_id,
"message_id": msg_id,
"text": t_msg,
"parse_mode": "HTML",
"reply_markup": kb
}, bot_token)
return None
# توابع کمکی کار با صف‌ها و وضعیت تسک در تمامی مسیرها
def get_all_queue_files(specific_task_id=None, queue_file_arg=None):
files = []
seen_ids = set()
now_ts = time.time()
if queue_file_arg:
p = Path(queue_file_arg)
if p.exists() and p.is_file():
try:
data = json.loads(p.read_text(encoding="utf-8"))
tid = str(data.get("task_id", p.stem))
if not is_task_cancelled(tid):
seen_ids.add(tid)
files.append(p)
else:
remove_queue_file_all(tid, str(p))
except Exception:
pass
if specific_task_id:
clean_id = re.sub(r'[^a-zA-Z0-9_-]', '', str(specific_task_id))
if not is_task_cancelled(clean_id):
candidates = [
LOCAL_QUEUE_DIR / f"queue_{clean_id}.json",
TEMP_QUEUE_DIR / f"queue_{clean_id}.json",
SYS_TEMP_QUEUE_DIR / f"queue_{clean_id}.json",
Path("/tmp/tg_userbot_queue") / f"queue_{clean_id}.json"
]
for c in candidates:
if c.exists() and c not in files:
seen_ids.add(clean_id)
files.append(c)
# پیمایش تمامی دایرکتوری‌های صف جهت واکشی سایر تسک‌ها به ترتیب
dirs = [LOCAL_QUEUE_DIR, TEMP_QUEUE_DIR, SYS_TEMP_QUEUE_DIR, Path("/tmp/tg_userbot_queue")]
for d in dirs:
if d.exists() and d.is_dir():
for f in d.glob("queue_*.json"):
try:
st = f.stat()
# پاکسازی خودکار فایل‌های صفر بایتی یا قدیمی‌تر از ۲۴ ساعت
if st.st_size == 0 or (now_ts - st.st_mtime > QUEUE_FILE_MAX_AGE):
try:
f.unlink()
except Exception:
pass
continue
data = json.loads(f.read_text(encoding="utf-8"))
tid = str(data.get("task_id", f.stem))
# اگر تسک لغو شده است بلافاصله پاکسازی شود
if is_task_cancelled(tid):
try:
f.unlink()
except Exception:
pass
continue
if tid not in seen_ids:
seen_ids.add(tid)
files.append(f)
except Exception:
pass
# مرتب‌سازی قطعی FIFO بر اساس زمان ایجاد تسک (اولین ارسالی = اولین دانلود)
def _sort_key(p):
try:
d = json.loads(p.read_text(encoding="utf-8"))
return float(d.get("created_at", p.stat().st_mtime))
except Exception:
return p.stat().st_mtime
files.sort(key=_sort_key)
return files
def save_task_state_multi(task_id, data):
clean_id = re.sub(r'[^a-zA-Z0-9_-]', '', str(task_id))
content = json.dumps(data, ensure_ascii=False)
paths = [
LOCAL_TASK_DIR / f"task_{clean_id}.json",
TEMP_TASK_DIR / f"task_{clean_id}.json",
SYS_TEMP_TASK_DIR / f"task_{clean_id}.json",
Path("/tmp/tg_uploader_tasks") / f"task_{clean_id}.json"
]
for p in paths:
try:
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content, encoding="utf-8")
except Exception:
pass
def is_task_cancelled(task_id, sync_task_id=None):
clean_id = re.sub(r'[^a-zA-Z0-9_-]', '', str(task_id))
paths = [
LOCAL_TASK_DIR / f"task_{clean_id}.json",
TEMP_TASK_DIR / f"task_{clean_id}.json",
SYS_TEMP_TASK_DIR / f"task_{clean_id}.json",
Path("/tmp/tg_uploader_tasks") / f"task_{clean_id}.json"
]
for p in paths:
if p.exists():
try:
data = json.loads(p.read_text(encoding="utf-8"))
if data.get("cancelled"):
return True
except Exception:
pass
# بررسی وضعیت لغو در تسک‌های همگام‌سازی کانال
if sync_task_id:
clean_sync = re.sub(r'[^a-zA-Z0-9_-]', '', str(sync_task_id))
sync_paths = [
BASE_DIR / "data" / "sync_tasks" / f"sync_{clean_sync}.json",
Path("data/sync_tasks") / f"sync_{clean_sync}.json",
Path("/tmp/sync_tasks") / f"sync_{clean_sync}.json"
]
for sp in sync_paths:
if sp.exists():
try:
sdata = json.loads(sp.read_text(encoding="utf-8"))
if sdata.get("cancelled"):
return True
except Exception:
pass
return False
def remove_queue_file_all(task_id, queue_file_path=None):
clean_id = re.sub(r'[^a-zA-Z0-9_-]', '', str(task_id))
candidates = [
LOCAL_QUEUE_DIR / f"queue_{clean_id}.json",
TEMP_QUEUE_DIR / f"queue_{clean_id}.json",
SYS_TEMP_QUEUE_DIR / f"queue_{clean_id}.json",
Path("/tmp/tg_userbot_queue") / f"queue_{clean_id}.json"
]
if queue_file_path:
candidates.append(Path(queue_file_path))
for c in candidates:
try:
if c.exists():
c.unlink()
except Exception:
pass
def remove_task_state_all(task_id):
clean_id = re.sub(r'[^a-zA-Z0-9_-]', '', str(task_id))
paths = [
LOCAL_TASK_DIR / f"task_{clean_id}.json",
TEMP_TASK_DIR / f"task_{clean_id}.json",
SYS_TEMP_TASK_DIR / f"task_{clean_id}.json",
Path("/tmp/tg_uploader_tasks") / f"task_{clean_id}.json"
]
for p in paths:
try:
if p.exists():
p.unlink()
except Exception:
pass
# ۱. مود لاگین اولیه اکانت
async def run_login(conf):
print("=" * 65)
print("🔐 احراز هویت و اتصال اکانت دانلودر به تلگرام (Userbot Login)")
print(f"📌 API ID: {conf['api_id']}")
print(f"📌 API HASH: {conf['api_hash']}")
print("=" * 65)
client = TelegramClient(StringSession(), conf["api_id"], conf["api_hash"])
await client.connect()
if not await client.is_user_authorized():
phone = input("📞 شماره موبایل اکانت را به همراه کد کشور وارد کنید (مثال: +989123456789): ").strip()
await client.send_code_request(phone)
code = input("📩 کد ۵ رقمی ارسال شده به تلگرام را وارد کنید: ").strip()
try:
await client.sign_in(phone, code)
except Exception as e:
if "Two-step verification" in str(e) or "SessionPasswordNeededError" in str(type(e)):
pwd = input("🔒 رمز دومرحله‌ای (2FA Password) اکانت را وارد کنید: ").strip()
await client.sign_in(password=pwd)
else:
raise e
me = await client.get_me()
session_str = client.session.save()
print("=" * 65)
print(f"✅ ورود موفقیت‌آمیز بود! مشخصات اکانت متصل شده:")
print(f"👤 نام: {me.first_name} {me.last_name or ''}")
print(f"🆔 شناسه عددی (User ID): {me.id}")
print(f"📱 شماره: +{me.phone}")
print("=" * 65)
# ذخیره خودکار در config.php
saved = save_session_string_to_config(session_str)
if saved:
print("💾 رشته نشست سشن (Session String) با موفقیت در config.php ذخیره شد.")
else:
print("⚠️ لطفاً رشته سشن زیر را در بخش 'userbot' داخل config.php در 'session_string' قرار دهید:")
print(f"\n{session_str}\n")
# ذخیره فایل فیزیکی سشن نیز جهت بک‌آپ
Path("userbot.session").write_text(session_str, encoding="utf-8")
await client.disconnect()
def is_valid_video_message(m, expected_size=0):
if not m or not getattr(m, "media", None):
return False
file_obj = getattr(m, "file", None)
media_obj = getattr(m, "media", None)
f_size = getattr(file_obj, "size", 0) or 0
if f_size == 0 and hasattr(media_obj, "document") and getattr(media_obj.document, "size", None):
f_size = media_obj.document.size
# ۱. رد قطعی استیکرها، ایموجی‌ها و تصاویر ثابت
if hasattr(media_obj, "document") and getattr(media_obj.document, "attributes", None):
for attr in media_obj.document.attributes:
attr_cls = type(attr).__name__
if attr_cls in ("DocumentAttributeSticker", "DocumentAttributeCustomEmoji"):
return False
mime = (getattr(file_obj, "mime_type", "") or getattr(getattr(media_obj, "document", None), "mime_type", "") or "").lower()
if "webp" in mime or "tgsticker" in mime:
return False
# ۲. فیلتر حجم: در صورتی که حجم مورد انتظار بالای ۱ مگابایت باشد، فایل‌های زیر ۱ مگابایت به هیچ عنوان ویدیو نیستند
if expected_size > 1024 * 1024:
if f_size < 1024 * 1024:
return False
else:
# حتی اگر حجم تسک صفر باشد، فایل‌های زیر ۵۰۰ کیلوبایت بدون MIME صریح ویدیو رد می‌شوند
if f_size < 512 * 1024 and not mime.startswith("video/"):
return False
# ۳. بررسی نشانه‌های قطعی ویدیو
# الف) دارا بودن اتریبیوت DocumentAttributeVideo
if hasattr(media_obj, "document") and getattr(media_obj.document, "attributes", None):
for attr in media_obj.document.attributes:
if type(attr).__name__ == "DocumentAttributeVideo":
return True
# ب) MIME type ویدیو (مانند video/mp4, video/mkv, video/webm, ...)
if mime.startswith("video/"):
return True
# ج) پسوند ویدیویی معتبر
file_name = getattr(file_obj, "name", "") or ""
if file_name and "." in file_name:
ext = file_name.split(".")[-1].lower()
if ext in ("mp4", "mkv", "webm", "mov", "avi", "flv", "m4v", "3gp", "ts"):
return True
# د) حجم بالای ۲ مگابایت برای اسنادی که به صورت Document (بدون فشرده‌سازی) ارسال شده‌اند
if f_size >= 2 * 1024 * 1024 and not mime.startswith("image/") and not mime.startswith("audio/"):
return True
return False
# ۲. تابع دانلود استریم فایل با سرعت بالا و گزارش زنده پیشرفت
async def download_task_media(client, task_info, used_message_ids=None):
if used_message_ids is None:
used_message_ids = set()
task_id = str(task_info.get("task_id", ""))
fwd_msg_id = int(task_info.get("forwarded_message_id", 0))
chat_id = int(task_info.get("user_chat_id", 0))
status_msg_id = int(task_info.get("status_message_id", 0))
title = task_info.get("video_title", "ویدیو سرور")
ext = task_info.get("clean_ext", "mp4")
site_url = task_info.get("site_url", "")
bot_id = int(task_info.get("bot_id", 0))
bot_username = task_info.get("bot_username", "")
queue_file_path = task_info.get("queue_file", "")
task_file_size = int(task_info.get("file_size", 0))
original_name = str(task_info.get("original_name", "")).strip()
conf = load_php_config()
if not bot_id and ":" in conf.get("bot_token", ""):
bot_id = int(conf["bot_token"].split(":")[0])
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] ⏳ شروع پردازش تسک {task_id} برای ویدیوی '{title}'")
print(f"📌 پیام فوروارد: {fwd_msg_id} | حجم: {task_file_size} بایت | ربات هدف: {bot_id} (@{bot_username}) | چت: {chat_id}")
if is_task_cancelled(task_id, task_info.get("sync_task_id")):
print(f"🛑 تسک {task_id} پیش از شروع دانلود توسط کاربر لغو شده است.")
remove_queue_file_all(task_id, queue_file_path)
return False
used_message_ids = used_message_ids if used_message_ids is not None else set()
used_message_ids.update(load_completed_msg_ids())
# اطمینان از برقراری اتصال زنده به کلاینت تلگرام
if not client.is_connected():
try:
print("🔄 اتصال تلگرام قطع بود؛ در حال اتصال مجدد...")
await client.connect()
except Exception as e:
print(f"⚠️ اتصال مجدد با خطا مواجه شد: {e}")
message = None
target_peer = None
target_dialog = None
# ۱. بررسی اولیه: واکشی مستقیم پیام از کانال مرجع بدون نیاز به جستجو در گفتگوی ربات
src_chat_id = int(task_info.get("source_chat_id", 0))
src_msg_id = int(task_info.get("source_message_id", 0))
if src_chat_id and src_msg_id:
try:
chan_msg = await client.get_messages(src_chat_id, ids=src_msg_id)
if chan_msg and is_valid_video_message(chan_msg):
message = chan_msg
print(f"✅ پیام ویدیو مستقیماً از کانال {src_chat_id} (شناسه پیام {src_msg_id}) واکشی شد.")
except Exception as e:
print(f"ℹ️ دریافت مستقیم از کانال مقدور نشد ({e})؛ جستجو در گفتگوی ربات انجام می‌شود...")
# ۲. در صورتی که پیام مستقیم یافت نشد، واکشی گفتگوهای اخیر جهت جستجو در چت ربات
if not message:
try:
dialogs = await client.get_dialogs(limit=50)
except Exception as e:
print(f"⚠️ دریافت لیست گفتگوها با خطا مواجه شد: {e}")
dialogs = []
# شناسایی دقیق گفتگوی ربات در گفتگوهای اکانت بر اساس bot_id یا bot_username
for d in dialogs:
ent = d.entity
ent_id = getattr(ent, 'id', None)
ent_uname = getattr(ent, 'username', None)
if bot_id and ent_id == bot_id:
target_peer = ent
target_dialog = d
break
if bot_username and ent_uname and str(ent_uname).lower() == str(bot_username).lower():
target_peer = ent
target_dialog = d
break
# اگر در گفتگوها نبود، دریافت مستقیم موجودیت ربات با get_entity
if not target_peer and (bot_id or bot_username):
try:
target_peer = await client.get_entity(bot_id or bot_username)
if target_peer:
print(f"💬 گفتگوی ربات مستقیماً از طریق get_entity پیدا شد: {bot_id or bot_username}")
except Exception as e:
print(f"⚠️ یافتن مستقیم ربات با get_entity ناموفق بود: {e}")
if not target_peer:
err_msg = f"❌ گفتگوی ربات ({bot_id or bot_username}) در اکانت دانلودر یافت نشد."
print(err_msg)
call_internal_api("fail", {
"task_id": task_id,
"error": "گفتگوی ربات در تلگرام اکانت دانلودر یافت نشد. لطفاً با اکانت دانلودر به آیدی ربات رفته و یک‌بار دستور /start را ارسال کنید.",
"chat_id": chat_id,
"status_msg_id": status_msg_id,
"site_url": site_url
})
remove_queue_file_all(task_id, queue_file_path)
return False
d_name = getattr(target_dialog, 'name', None) or getattr(target_peer, 'first_name', None) or getattr(target_peer, 'title', None) or 'نامشخص'
d_id = getattr(target_peer, 'id', None)
print(f"💬 گفتگوی هدف ربات شناسایی شد: '{d_name}' (شناسه: {d_id})")
# ۳. جستجو و انطباق پیام ویدیوی معتبر در چت ربات با حلقه تلاش مجدد هوشمند
for attempt in range(4):
try:
recent_msgs = await client.get_messages(target_peer, limit=100)
chat_messages = list(recent_msgs) if recent_msgs else []
except Exception as e:
print(f"⚠️ خطا در واکشی پیام‌های گفتگوی ربات: {e}")
chat_messages = []
# فیلتر سخت‌گیرانه: فقط ویدیوهای معتبر که قبلاً توسط تسک‌های دیگر مصرف نشده‌اند
valid_video_candidates = [
m for m in chat_messages
if m and m.id not in used_message_ids and is_valid_video_message(m, expected_size=task_file_size)
]
# الف) تطابق با شناسه دقیق پیام فوروارد شده (fwd_msg_id)
if fwd_msg_id and fwd_msg_id not in used_message_ids:
for m in valid_video_candidates:
if m.id == fwd_msg_id:
message = m
print(f"✅ پیام ویدیو با شناسه دقیق {fwd_msg_id} در گفتگوی ربات یافت شد.")
break
if not message:
try:
direct_msg = await client.get_messages(target_peer, ids=fwd_msg_id)
if direct_msg and direct_msg.id not in used_message_ids and is_valid_video_message(direct_msg, expected_size=task_file_size):
message = direct_msg
print(f"✅ پیام ویدیو مستقیماً با شناسه {fwd_msg_id} دریافت شد.")
except Exception as e:
print(f"⚠️ دریافت مستقیم پیام با شناسه {fwd_msg_id}: {e}")
# ب) تطابق با حجم دقیق یا بسیار نزدیک بایت‌های فایل (File Size Match)
if not message and task_file_size > 0:
for m in valid_video_candidates:
cand_size = getattr(m.file, "size", 0) or 0
if cand_size == task_file_size or abs(cand_size - task_file_size) <= 102400:
message = m
print(f"✅ پیام ویدیو بر اساس تطابق حجم فایل ({cand_size} بایت) با شناسه {m.id} شناسایی شد.")
break
# ج) تطابق با نام اصلی فایل
if not message and original_name:
for m in valid_video_candidates:
c_name = getattr(m.file, "name", "") or ""
if c_name and c_name.strip().lower() == original_name.lower():
message = m
print(f"✅ پیام ویدیو بر اساس تطابق نام فایل ('{c_name}') با شناسه {m.id} شناسایی شد.")
break
# د) تطابق با ترتیب صف (FIFO): قدیمی‌ترین ویدیوی معتبر مصرف‌نشده
if not message and valid_video_candidates:
valid_video_candidates.sort(key=lambda m: m.id)
message = valid_video_candidates[0]
print(f"✅ پیام ویدیو بر اساس ترتیب زمانی صف (FIFO) با شناسه {message.id} انتخاب شد.")
if message:
break
if attempt < 3:
print(f"⏳ پیام ویدیوی معتبر هنوز در چت ربات دریافت نشده؛ تلاش مجدد ({attempt + 1}/3) پس از ۳ ثانیه...")
await asyncio.sleep(3)
if not message or not getattr(message, "media", None):
err_msg = f"❌ هیچ فایل ویدیویی معتبری برای تسک {task_id} در گفتگوی ربات تلگرام یافت نشد."
print(err_msg)
call_internal_api("fail", {
"task_id": task_id,
"error": "پیام ویدیو در گفتگوی ربات دریافت نشد. لطفاً مطمئن شوید اکانت ربات را بلاک نکرده و یک‌بار با آن /start زده باشد.",
"chat_id": chat_id,
"status_msg_id": status_msg_id,
"site_url": site_url
})
remove_queue_file_all(task_id, queue_file_path)
return False
# ثبت شناسه پیام مصرف شده تا هیچ تسک دیگری دوباره این پیام را دانلود نکند
used_message_ids.add(message.id)
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
# تشخیص دقیق پسوند فایل از مشخصات مدیا در تلگرام
if message.file and getattr(message.file, "name", None) and "." in str(message.file.name):
detected_ext = str(message.file.name).split(".")[-1].lower()
if detected_ext in ["mp4", "mkv", "webm", "mov", "avi", "flv", "m4v"]:
ext = detected_ext
elif message.file and getattr(message.file, "ext", None):
detected_ext = str(message.file.ext or "").lstrip(".").lower()
if detected_ext in ["mp4", "mkv", "webm", "mov", "avi", "flv", "m4v"]:
ext = detected_ext
file_name = f"vid_{time.strftime('%Y%m%d_%H%M%S')}_{os.urandom(3).hex()}.{ext}"
target_path = UPLOAD_DIR / file_name
total_size = message.file.size or 0
start_time = time.time()
last_update_time = 0
last_edit_time = 0
last_reported_percent = -1
last_speed_time = time.time()
last_speed_bytes = 0
current_speed = 0
current_download_mode = "fast"
print(f"🚀 در حال دانلود مدیا '{title}' به حجم {total_size} بایت ({round(total_size / (1024*1024), 2)} MB)...")
def progress_callback(current, total):
nonlocal last_update_time, last_edit_time, last_reported_percent, last_speed_time, last_speed_bytes, current_speed
now_time = time.time()
# بررسی لغو تسک
if is_task_cancelled(task_id):
print(f"❌ دانلود تسک {task_id} توسط دکمه لغو کاربر متوقف شد.")
raise asyncio.CancelledError("User cancelled download")
# محاسبه سرعت با میانگین متحرک پایدار (EMA) بر پایه پنجره زمانی ۱.۲ ثانیه‌ای
dt = now_time - last_speed_time
if dt >= 1.2 or current >= total:
db = current - last_speed_bytes
instant_speed = int(db / dt) if dt > 0 else 0
total_dt = now_time - start_time
overall_speed = int(current / total_dt) if total_dt > 0.8 else instant_speed
if current_speed <= 0:
current_speed = instant_speed or overall_speed
else:
# میانگین متحرک وزنی نمایی (EMA) با ثبات بالا
current_speed = int(0.35 * instant_speed + 0.65 * current_speed)
# در صورتی که میانگین کلی بالای ۱ مگابایت است اما به دلیل وقفه فاصله‌ای چانک‌ها افت لحظه‌ای رخ داد، افت کاذب را فیلتر کن
if overall_speed > 1024 * 1024 and current_speed < 0.6 * overall_speed:
current_speed = int(0.75 * overall_speed + 0.25 * current_speed)
last_speed_time = now_time
last_speed_bytes = current
# ثبت سبک در دیسک محلی
if now_time - last_update_time >= 1.0 or current >= total:
last_update_time = now_time
touch_worker_lock()
speed = max(100 * 1024, current_speed)
rem = int((total - current) / speed) if speed > 0 else 0
percent = int((current / total) * 100) if total > 0 else 0
percent = 100 if current >= total else min(percent, 99)
# ادیت سبک پیام تلگرام هر ۳.۵ ثانیه و با تغییر حداقل ۲٪ پیشرفت برای کاهش شدید مصرف CPU
should_edit = (now_time - last_edit_time >= 3.5 and abs(percent - last_reported_percent) >= 2) and current < total
if should_edit:
last_edit_time = now_time
last_reported_percent = percent
# بررسی مجدد لغو تسک قبل از ذخیره وضعیت
if is_task_cancelled(task_id, task_info.get("sync_task_id")):
raise asyncio.CancelledError("User cancelled download")
# ثبت وضعیت تسک در تمامی مسیرهای محلی
task_data = {
"taskId": task_id,
"chatId": chat_id,
"messageId": status_msg_id,
"title": title,
"now": current,
"total": total,
"percent": percent,
"speed": speed,
"remSeconds": rem,
"cancelled": False,
"updatedAt": int(now_time),
"engine": "userbot",
"mode": current_download_mode
}
save_task_state_multi(task_id, task_data)
if should_edit or current >= total:
speed_str = f"{round(speed/(1024*1024), 2)} MB/s" if speed >= 1024*1024 else f"{round(speed/1024, 1)} KB/s"
print(f"📊 پیشرفت تسک {task_id}: {percent}% ({round(current/(1024*1024), 2)}/{round(total/(1024*1024), 2)} MB) - سرعت: {speed_str} [{current_download_mode}]")
# ارسال وضعیت به userbot_api.php جهت ادیت پیام تلگرام به صورت ناهمگام در ترد پس‌زمینه
call_internal_api_async("update_progress", {
"task_id": task_id,
"now": current,
"total": total,
"speed": speed,
"rem_seconds": rem,
"percent": percent,
"title": title,
"chat_id": chat_id,
"msg_id": status_msg_id,
"site_url": site_url,
"should_edit_message": 1,
"mode": current_download_mode,
"is_sync": 1 if task_info.get("is_sync") else 0,
"sync_task_id": task_info.get("sync_task_id", ""),
"sync_index": task_info.get("sync_index", 1),
"sync_total": task_info.get("sync_total", 1),
"sync_done_count": task_info.get("sync_done_count", 0),
"sync_done_bytes": task_info.get("sync_done_bytes", 0),
"sync_total_bytes": task_info.get("sync_total_bytes", 0),
"sync_start_time": task_info.get("sync_start_time", 0),
"channel_title": task_info.get("channel_title", ""),
"course_title": task_info.get("course_title", ""),
"session_num": task_info.get("session_num", 1),
"part_num": task_info.get("part_num", 1)
})
try:
# دانلود چنداتصاله موازی با FastTelethon برای بیشترین سرعت ممکن روی سرور
fast_success = False
target_size = total_size or task_file_size or 0
if fast_download_file and target_size > 1024 * 1024:
for fast_attempt in range(2):
try:
print(f"⚡ شروع دانلود چنداتصاله موازی FastTelethon (تنظیم‌شده روی سقف I/O ۶ مگابایت سرور) برای تسک {task_id} ({round(target_size/(1024*1024), 2)} MB) - نوبت تلاش {fast_attempt + 1}...")
current_download_mode = "fast"
await fast_download_file(
client=client,
location=message,
out=str(target_path),
file_size=target_size,
progress_callback=progress_callback,
connection_count=8
)
is_fully_downloaded = target_path.exists() and (target_path.stat().st_size >= target_size or (target_size - target_path.stat().st_size <= 65536))
if is_fully_downloaded:
fast_success = True
print(f"✅ دانلود فوق‌سریع موازی با موفقیت به پایان رسید ({target_path.stat().st_size} بایت).")
break
else:
print(f"⚠️ فایل دانلود شده کامل نیست ({target_path.stat().st_size} از {target_size} بایت)؛ در حال دانلود بخش‌های نهایی...")
except asyncio.CancelledError:
raise
except Exception as fast_err:
import traceback
tb_str = traceback.format_exc()
if target_path.exists():
try:
target_path.unlink()
except Exception:
pass
if fast_attempt == 0:
print(f"⚠️ خطای موقت اتصال اول ({fast_err})؛ تلاش مجدد با آماده‌سازی کانکشن‌ها...")
await asyncio.sleep(0.5)
else:
print(f"⚠️ خطای نهایی دانلود موازی ({fast_err})؛ بازگشت به دانلود استاندارد تلتون...\n{tb_str}")
if not fast_success:
current_download_mode = "standard"
start_time = time.time()
# دانلود از طریق کلاینت تلتون استاندارد (تک‌اتصاله)
downloaded_file = await client.download_media(message, file=str(target_path), progress_callback=progress_callback)
except asyncio.CancelledError:
if target_path.exists():
target_path.unlink()
remove_queue_file_all(task_id, queue_file_path)
return False
except Exception as e:
print("❌ خطا در حین دانلود:", e)
call_internal_api("fail", {
"task_id": task_id,
"error": str(e),
"chat_id": chat_id,
"status_msg_id": status_msg_id,
"site_url": site_url
})
remove_queue_file_all(task_id, queue_file_path)
return False
if not target_path.exists() or target_path.stat().st_size == 0:
print("❌ فایل دانلود شده روی دیسک خالی است.")
remove_queue_file_all(task_id, queue_file_path)
return False
actual_size = target_path.stat().st_size
if task_file_size > 1024 * 1024 and actual_size < 1024 * 1024:
if target_path.exists():
try:
target_path.unlink()
except Exception:
pass
err_msg = f"❌ حجم فایل دانلود شده ({actual_size} بایت) بسیار کمتر از حجم ویدیو ({task_file_size} بایت) است. فایل نامعتبر حذف شد."
print(err_msg)
call_internal_api("fail", {
"task_id": task_id,
"error": err_msg,
"chat_id": chat_id,
"status_msg_id": status_msg_id,
"site_url": site_url
})
remove_queue_file_all(task_id, queue_file_path)
return False
duration = 0
if message.file and hasattr(message.file, "duration"):
duration = int(message.file.duration or 0)
# بررسی تنظیمات ذخیره‌سازی و FTP
current_conf = load_php_config()
ftp_cfg = current_conf.get("ftp") or {}
storage_cfg = current_conf.get("storage") or {}
storage_mode = str(storage_cfg.get("mode", "local")).lower()
# در صورتی که حالت ذخیره‌سازی روی FTP باشد و تنظیمات FTP فعال باشد
ftp_enabled = (storage_mode == "ftp") and bool(ftp_cfg.get("enabled")) and bool(ftp_cfg.get("host")) and bool(ftp_cfg.get("username"))
ftp_transferred = False
if ftp_enabled:
print(f"🌐 انتقال مستقیم FTP فعال است؛ در حال استریم ویدیو به هاست مقصد...")
is_sync_task = 1 if task_info.get("is_sync") else 0
sync_payload_extra = {}
if is_sync_task:
sync_payload_extra = {
"is_sync": 1,
"sync_task_id": task_info.get("sync_task_id", ""),
"sync_index": task_info.get("sync_index", 1),
"sync_total": task_info.get("sync_total", 1),
"sync_done_count": task_info.get("sync_done_count", 0),
"sync_done_bytes": task_info.get("sync_done_bytes", 0),
"sync_total_bytes": task_info.get("sync_total_bytes", 0),
"sync_start_time": task_info.get("sync_start_time", 0),
"channel_title": task_info.get("channel_title", ""),
"course_title": task_info.get("course_title", ""),
"session_num": task_info.get("session_num", 1),
"part_num": task_info.get("part_num", 1)
}
# گزارش شروع مرحله انتقال در تلگرام
ftp_start_payload = {
"task_id": task_id,
"now": 0,
"total": actual_size,
"speed": 0,
"rem_seconds": 0,
"percent": 0,
"title": title,
"chat_id": chat_id,
"msg_id": status_msg_id,
"site_url": site_url,
"should_edit_message": 1,
"mode": "ftp",
"is_ftp": 1
}
ftp_start_payload.update(sync_payload_extra)
call_internal_api_async("update_progress", ftp_start_payload)
def ftp_progress_callback(sent, total, speed):
rem = int((total - sent) / speed) if speed > 0 else 0
pct = int((sent / total) * 100) if total > 0 else 0
pct = 100 if sent >= total else min(pct, 99)
speed_str = f"{round(speed/(1024*1024), 2)} MB/s" if speed >= 1024*1024 else f"{round(speed/1024, 1)} KB/s"
print(f"🚀 انتقال FTP تسک {task_id}: {pct}% ({round(sent/(1024*1024), 2)}/{round(total/(1024*1024), 2)} MB) - سرعت: {speed_str}")
cb_payload = {
"task_id": task_id,
"now": sent,
"total": total,
"speed": speed,
"rem_seconds": rem,
"percent": pct,
"title": title,
"chat_id": chat_id,
"msg_id": status_msg_id,
"site_url": site_url,
"should_edit_message": 1,
"mode": "ftp",
"is_ftp": 1
}
cb_payload.update(sync_payload_extra)
call_internal_api_async("update_progress", cb_payload)
ftp_transferred = upload_file_ftp(
target_path,
file_name,
ftp_cfg,
progress_cb=ftp_progress_callback,
cancel_cb=lambda: is_task_cancelled(task_id, task_info.get("sync_task_id"))
)
if ftp_transferred:
# پاکسازی ایمن فایل محلی پس از انتقال موفقیت‌آمیز به هاست مقصد
try:
target_path.unlink()
print(f"🗑️ فایل محلی {file_name} پس از انتقال موفقیت‌آمیز FTP پاک شد.")
except Exception:
pass
else:
print(f"📦 حالت ذخیره‌سازی محلی (Local): ویدیو روی همین سرور دانلودر نگهداری می‌شود و هیچ فایلی با FTP به هاست دوم ارسال نمی‌گردد.")
# صبر برای پایان تردهای ناهمگام ادیت درصد، تا هیچ پیام درصدی بعد از پیام موفقیت ارسال نشود
wait_pending_api_calls(timeout=15)
# تکمیل و ثبت نهایی در دیتابیس
call_internal_api("complete", {
"task_id": task_id,
"file_name": file_name,
"original_name": task_info.get("original_name", file_name),
"file_size": actual_size,
"duration": duration,
"title": title,
"chat_id": chat_id,
"status_msg_id": status_msg_id,
"site_url": site_url,
"channel_id": task_info.get("source_chat_id", 0),
"channel_message_id": task_info.get("source_message_id", 0),
"session_id": task_info.get("session_id", 0),
"session_num": task_info.get("session_num", 0),
"part_num": task_info.get("part_num", 0),
"part_title": task_info.get("video_title", ""),
"is_ftp": 1 if ftp_transferred else 0,
"storage_mode": "ftp" if ftp_transferred else "local",
"is_sync": 1 if task_info.get("is_sync") else 0,
"sync_task_id": task_info.get("sync_task_id", ""),
"sync_index": task_info.get("sync_index", 1),
"sync_total": task_info.get("sync_total", 1),
"sync_done_count": task_info.get("sync_done_count", 0),
"sync_done_bytes": task_info.get("sync_done_bytes", 0),
"sync_total_bytes": task_info.get("sync_total_bytes", 0),
"channel_title": task_info.get("channel_title", ""),
"course_title": task_info.get("course_title", "")
})
# پاکسازی فایل صف از تمامی مسیرها
remove_queue_file_all(task_id, queue_file_path)
# ثبت دائمی شناسه پیام جهت جلوگیری قطعی از دانلود مجدد پارت‌های تکمیل شده
if message and hasattr(message, "id") and message.id:
save_completed_msg_id(message.id)
used_message_ids.add(message.id)
print(f"🎉 دانلود و ثبت با موفقیت تمام شد: {file_name} ({actual_size} بایت)")
return True
# ۳. پردازش ترتیبی تسک‌های در صف (Sequential FIFO Queue)
async def process_queue(conf, specific_task_id=None, queue_file=None, _retry=0):
session_str = conf.get("session_string") or (Path("userbot.session").read_text() if Path("userbot.session").exists() else "")
if not session_str:
print("❌ خطا: اکانت لاگین نشده است. لطفاً رشته سشن را در setup.php بررسی فرمایید.")
return
# سیستم قفل انحصاری پروسس: اگر ورکر دیگری هم‌اکنون در حال پردازش صف است، تداخل ایجاد نشود
finished_normally = False
lock = WorkerProcessLock()
if not lock.acquire():
print("ℹ️ ورکر دانلودر دیگری هم‌اکنون در پس‌زمینه در حال دانلود است. تسک در صف ثبت شد و به ترتیب نوبت دانلود خواهد شد.")
return
try:
tasks_to_run = get_all_queue_files(specific_task_id=specific_task_id, queue_file_arg=queue_file)
if not tasks_to_run:
print("ℹ️ صف دانلود خالی است (هیچ تسک منتظری یافت نشد).")
return
print(f"🔄 تعداد {len(tasks_to_run)} تسک در صف شناسایی شد. در حال اتصال به کلاینت MTProto تلگرام...")
try:
from telethon.crypto import libssl
if libssl.decrypt_ige:
print("⚡ شتاب‌دهنده سخت‌افزاری C OpenSSL (AES-NI) فعال است.")
else:
print("⚠️ شتاب‌دهنده C OpenSSL لود نشد؛ در حال استفاده از رمزگشایی پایتون.")
except Exception:
pass
client = TelegramClient(StringSession(session_str), conf["api_id"], conf["api_hash"])
try:
await asyncio.wait_for(client.connect(), timeout=15)
except Exception as e:
print(f"❌ خطا در اتصال به تلگرام: {e}")
return
if not await client.is_user_authorized():
print("❌ نشست سشن منقضی یا نامعتبر است. لطفاً در setup.php مجدداً وارد اکانت شوید.")
await client.disconnect()
return
try:
me = await client.get_me()
print(f"🟢 اکانت متصل است: {getattr(me, 'first_name', '')} (شناسه: {getattr(me, 'id', '')})")
except Exception:
pass
# پردازش ترتیبی صف تا خالی شدن کامل آن
first_pass = True
used_message_ids = set()
consecutive_empty = 0
worker_file_path = Path(__file__).resolve()
worker_initial_mtime = worker_file_path.stat().st_mtime if worker_file_path.exists() else 0
while True:
# اگر فایل اسکریپت ورکر روی سرور آپدیت شده، خروج تمیز انجام می‌دهیم تا نسخه جدید لود شود
if worker_file_path.exists() and worker_file_path.stat().st_mtime > worker_initial_mtime:
print("🔄 فایل ورکر پایتون روی دیسک آپدیت شده است؛ خروج تمیز جهت اجرای نسخه جدید...")
break
pending_tasks = get_all_queue_files(
specific_task_id=specific_task_id if first_pass else None,
queue_file_arg=queue_file if first_pass else None
)
first_pass = False
specific_task_id = None
queue_file = None
if not pending_tasks:
consecutive_empty += 1
if consecutive_empty >= 2:
print("🏁 تمام تسک‌های موجود در صف با موفقیت دانلود و ذخیره شدند.")
break
# فرصت کوتاه ۳ ثانیه‌ای برای ثبت تسک‌های جدید توسط وب‌هوک‌های موازی
await asyncio.sleep(3)
continue
consecutive_empty = 0
touch_worker_lock()
current_task_file = pending_tasks[0]
tid = None
try:
try:
data = json.loads(current_task_file.read_text(encoding="utf-8"))
except Exception:
try:
current_task_file.unlink()
except Exception:
pass
continue
tid = str(data.get("task_id", current_task_file.stem))
chat_id = data.get("user_chat_id") or data.get("chat_id")
status_msg_id = data.get("status_message_id") or data.get("status_msg_id")
v_title = data.get("video_title") or data.get("title", "")
# اگر تسک قبل از رسیدن نوبت لغو شده باشد
if is_task_cancelled(tid):
print(f"⏩ تسک {tid} در صف لغو شده است؛ پاکسازی از صف...")
remove_queue_file_all(tid, str(current_task_file))
try:
current_task_file.unlink()
except Exception:
pass
continue
if chat_id and status_msg_id:
call_internal_api("notify_start", {
"task_id": tid,
"chat_id": chat_id,
"status_msg_id": status_msg_id,
"title": v_title,
"site_url": data.get("site_url", "")
})
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 🚀 شروع دانلود تسک: {tid} ('{v_title}')")
await download_task_media(client, data, used_message_ids=used_message_ids)
except Exception as e:
print(f"❌ خطا در پردازش تسک {current_task_file.name}:", e)
finally:
# اطمینان قطعی از حذف تسک جاری از صف تحت هر شرایطی
try:
if current_task_file.exists():
current_task_file.unlink()
except Exception:
pass
if tid:
remove_queue_file_all(tid, str(current_task_file))
await asyncio.sleep(1)
await client.disconnect()
finished_normally = True
finally:
lock.release()
# race: تسکی که درست در لحظهٔ پایان ورکر ثبت شده، ورکر جدیدی نمی‌گرفت چون قفل هنوز آزاد نشده بود.
# پس بعد از آزاد شدن قفل یک‌بار دیگر صف را نگاه می‌کنیم و در صورت وجود تسک دوباره شروع می‌کنیم.
if finished_normally and _retry < 20:
try:
await asyncio.sleep(1)
if get_all_queue_files():
print("🔁 تسک جدیدی همزمان با پایان ورکر ثبت شده بود؛ ادامهٔ پردازش صف...")
await process_queue(conf, _retry=_retry + 1)
except Exception as e:
print(f"⚠️ خطا در بررسی مجدد صف: {e}")
# ۴. حالت گوش دادن مداوم (Daemon Mode)
async def run_daemon(conf):
session_str = conf.get("session_string") or (Path("userbot.session").read_text() if Path("userbot.session").exists() else "")
if not session_str:
print("❌ خطا: اکانت لاگین نشده است. ابتدا با دستور python3 userbot_worker.py --login وارد شوید.")
return
client = TelegramClient(StringSession(session_str), conf["api_id"], conf["api_hash"])
try:
await asyncio.wait_for(client.connect(), timeout=15)
except Exception as e:
print(f"❌ خطا در اتصال به تلگرام: {e}")
return
if not await client.is_user_authorized():
print("❌ نشست سشن منقضی یا نامعتبر است. لطفاً رشته سشن را در setup.php بررسی فرمایید.")
await client.disconnect()
return
print("=" * 65)
print("🚀 دیمن دانلودر خودکار اکانت فعال شد.")
print(f"👤 متصل به اکانت ID: {conf['account_id']}")
print("🔄 منتظر دریافت ویدیوهای فوروارد شده از ربات...")
print("=" * 65)
daemon_used_ids = set()
while True:
try:
for qf in get_all_queue_files():
try:
data = json.loads(qf.read_text(encoding="utf-8"))
await download_task_media(client, data, used_message_ids=daemon_used_ids)
except Exception as e:
print("❌ خطا در پردازش فایل صف دیمن:", e)
except Exception as e:
print("❌ خطای حلقه دیمن:", e)
await asyncio.sleep(2)
# ۵. بررسی وضعیت اتصال اکانت برای وب و پنل (JSON Output)
async def check_status(conf):
session_str = conf.get("session_string") or (Path("userbot.session").read_text() if Path("userbot.session").exists() else "")
if not session_str:
print(json.dumps({"ok": False, "status": "no_session", "error": "سشن اکانت هنوز تنظیم نشده است."}, ensure_ascii=False))
return
client = TelegramClient(StringSession(session_str), conf["api_id"], conf["api_hash"])
try:
if hasattr(asyncio, "wait_for"):
await asyncio.wait_for(client.connect(), timeout=6)
else:
await client.connect()
if not await client.is_user_authorized():
print(json.dumps({"ok": False, "status": "unauthorized", "error": "سشن منقضی یا نامعتبر شده است."}, ensure_ascii=False))
await client.disconnect()
return
me = await client.get_me()
print(json.dumps({
"ok": True,
"status": "connected",
"account_id": me.id,
"first_name": me.first_name or "",
"last_name": me.last_name or "",
"username": me.username or "",
"phone": me.phone or ""
}, ensure_ascii=False))
await client.disconnect()
except Exception as e:
print(json.dumps({"ok": False, "status": "error", "error": f"خطا در اتصال: {e}"}, ensure_ascii=False))
# ۶. ارسال کد ورود به تلگرام از طریق وب
async def web_send_code(conf, phone):
client = TelegramClient(StringSession(), conf["api_id"], conf["api_hash"])
try:
await client.connect()
res = await client.send_code_request(phone)
temp_state = {
"phone": phone,
"phone_code_hash": res.phone_code_hash,
"temp_session": client.session.save(),
"time": time.time()
}
state_file = Path(os.environ.get("TMPDIR", "/tmp")) / "tg_userbot_web_login.json"
state_file.write_text(json.dumps(temp_state), encoding="utf-8")
print(json.dumps({
"ok": True,
"message": "کد تأیید تلگرام با موفقیت ارسال شد.",
"phone": phone
}, ensure_ascii=False))
await client.disconnect()
except Exception as e:
print(json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False))
# ۷. تأیید کد ورود و دریافت سشن استرینگ از طریق وب
async def web_verify_code(conf, code, password=None):
from telethon.errors import SessionPasswordNeededError
state_file = Path(os.environ.get("TMPDIR", "/tmp")) / "tg_userbot_web_login.json"
if not state_file.exists():
print(json.dumps({"ok": False, "error": "درخواست ارسال کد یافت نشد یا منقضی شده است. لطفاً مجدداً شماره را ارسال کنید."}, ensure_ascii=False))
return
try:
data = json.loads(state_file.read_text(encoding="utf-8"))
phone = data["phone"]
phone_code_hash = data["phone_code_hash"]
temp_session = data["temp_session"]
client = TelegramClient(StringSession(temp_session), conf["api_id"], conf["api_hash"])
await client.connect()
try:
await client.sign_in(phone=phone, code=code, phone_code_hash=phone_code_hash)
except SessionPasswordNeededError:
if not password:
print(json.dumps({
"ok": False,
"need_password": True,
"message": "اکانت شما دارای رمز تأیید دومرحله‌ای (2FA) است. لطفاً رمز عبور را وارد کنید."
}, ensure_ascii=False))
await client.disconnect()
return
await client.sign_in(password=password)
me = await client.get_me()
session_str = client.session.save()
save_session_string_to_config(session_str)
try:
Path("userbot.session").write_text(session_str, encoding="utf-8")
except Exception:
pass
if state_file.exists():
state_file.unlink()
print(json.dumps({
"ok": True,
"message": "ورود به اکانت با موفقیت انجام شد و نشست ذخیره گردید!",
"session_string": session_str,
"account_id": me.id,
"name": f"{me.first_name} {me.last_name or ''}".strip()
}, ensure_ascii=False))
await client.disconnect()
except Exception as e:
print(json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False))
def convert_digits(text):
if not text:
return ""
fa_digits = "۰۱۲۳۴۵۶۷۸۹"
ar_digits = "٠١٢٣٤٥٦٧٨٩"
en_digits = "0123456789"
trans = str.maketrans(fa_digits + ar_digits, en_digits + en_digits)
return str(text).translate(trans)
# ۵. اسکن کامل تاریخچه کانال و صدور پیش‌فاکتور (Channel History Scan)
async def scan_channel_history(conf, channel_id, out_file=None, status_msg_id=0, admin_chat_id=0):
session_str = conf.get("session_string") or (Path("userbot.session").read_text() if Path("userbot.session").exists() else "")
if not session_str:
err = {"ok": False, "error": "نشست اکانت دانلودر تنظیم نشده است"}
if out_file: Path(out_file).write_text(json.dumps(err, ensure_ascii=False), encoding="utf-8")
print(json.dumps(err, ensure_ascii=False))
return
client = TelegramClient(StringSession(session_str), conf["api_id"], conf["api_hash"])
await client.connect()
if not await client.is_user_authorized():
err = {"ok": False, "error": "اکانت دانلودر احراز هویت نشده است"}
if out_file: Path(out_file).write_text(json.dumps(err, ensure_ascii=False), encoding="utf-8")
print(json.dumps(err, ensure_ascii=False))
await client.disconnect()
return
try:
entity = await client.get_entity(int(channel_id))
except Exception as e:
err = {"ok": False, "error": f"عدم دسترسی به کانال: {e}"}
if out_file: Path(out_file).write_text(json.dumps(err, ensure_ascii=False), encoding="utf-8")
print(json.dumps(err, ensure_ascii=False))
await client.disconnect()
return
channel_title = getattr(entity, 'title', 'کانال')
# واکشی ترتیبی پیام‌ها از ابتدا به انتها
all_msgs = []
async for msg in client.iter_messages(entity, reverse=True):
all_msgs.append(msg)
# جستجوی پیام «شروع»
start_found = False
start_idx = 0
for idx, msg in enumerate(all_msgs):
m_text = (msg.message or msg.text or "").strip()
if "شروع" in m_text:
start_idx = idx
start_found = True
break
sessions = []
current_session = None
part_counter = 0
for msg in all_msgs[start_idx:]:
raw_text = (msg.message or msg.text or "").strip()
has_video = False
media_size = 0
media_duration = 0
media_name = ""
if msg.video:
has_video = True
media_size = getattr(msg.video, 'size', 0)
media_duration = getattr(msg.video, 'duration', 0)
media_name = getattr(msg.file, 'name', '') or f"vid_{msg.id}.mp4"
elif msg.document:
mime = getattr(msg.document, 'mime_type', '') or ''
attrs = getattr(msg.document, 'attributes', [])
is_vid_doc = mime.startswith('video/') or any(type(a).__name__ == 'DocumentAttributeVideo' for a in attrs)
doc_name = getattr(msg.file, 'name', '') or ''
if is_vid_doc or doc_name.lower().endswith(('.mp4', '.mkv', '.webm', '.mov')):
has_video = True
media_size = getattr(msg.document, 'size', 0)
media_duration = getattr(msg.file, 'duration', 0) if msg.file else 0
media_name = doc_name or f"vid_{msg.id}.mp4"
# پیام متنی شروع جلسه جدید
if raw_text and not has_video:
clean_t = convert_digits(raw_text)
sess_num = None
sess_match = re.search(r'(?:جلسه|session|sess|jalse)\s*[:\-_#]?\s*(\d+)', clean_t, re.IGNORECASE)
if sess_match:
sess_num = int(sess_match.group(1))
elif "جلسه" in clean_t:
sess_num = (current_session["session_num"] + 1) if current_session else 1
if sess_num is not None:
current_session = {
"session_num": sess_num,
"session_title": raw_text,
"videos": []
}
sessions.append(current_session)
part_counter = 0
continue
# ویدیوی پارت جلسه
if has_video:
if current_session is None:
current_session = {
"session_num": 1,
"session_title": "جلسه 1",
"videos": []
}
sessions.append(current_session)
part_counter = 0
part_counter += 1
# بررسی کپشن (با حذف پسوند mp4 جهت جلوگیری از خطای p4)
clean_cap = re.sub(r'\.(mp4|mkv|webm|mov)$', '', raw_text, flags=re.I)
part_match = re.search(r'(?:پارت|part|بخش|قسمت)\s*[:\-_#]?\s*(\d+)', clean_cap, re.IGNORECASE)
p_num = int(part_match.group(1)) if part_match else part_counter
current_session["videos"].append({
"message_id": msg.id,
"part_num": p_num,
"part_title": f"پارت {p_num}",
"caption": raw_text,
"file_size": media_size,
"duration": media_duration,
"file_name": media_name
})
total_videos = sum(len(s["videos"]) for s in sessions)
total_size = sum(v["file_size"] for s in sessions for v in s["videos"])
avg_speed = 12 * 1024 * 1024 # میانگین ۱۲ مگابایت بر ثانیه
est_seconds = int(total_size / avg_speed) if avg_speed > 0 else 0
res = {
"ok": True,
"channel_id": int(channel_id),
"channel_title": channel_title,
"start_found": start_found,
"total_sessions": len(sessions),
"total_videos": total_videos,
"total_size": total_size,
"estimated_seconds": est_seconds,
"sessions": sessions
}
if out_file:
Path(out_file).parent.mkdir(parents=True, exist_ok=True)
Path(out_file).write_text(json.dumps(res, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(res, ensure_ascii=False))
# رندر بلافاصله پیش‌فاکتور در پیوی ادمین
if admin_chat_id > 0 and status_msg_id > 0:
call_internal_api("render_scan_invoice", {
"channel_id": int(channel_id),
"admin_chat_id": admin_chat_id,
"status_msg_id": status_msg_id,
"scan_data": res
})
await client.disconnect()
# ۶. اجرای ترتیبی همگام‌سازی کامل کانال با یک پیام زنده (Channel Full Sync Runner)
async def run_channel_sync(conf, sync_task_file):
sync_file = Path(sync_task_file)
if not sync_file.exists():
print(f"❌ فایل همگام‌سازی یافت نشد: {sync_task_file}")
return
data = json.loads(sync_file.read_text(encoding="utf-8"))
channel_id = int(data.get("channel_id", 0))
chat_id = int(data.get("chat_id", 0))
status_msg_id = int(data.get("status_msg_id", 0))
items = data.get("items", [])
total_videos = len(items)
session_str = conf.get("session_string") or (Path("userbot.session").read_text() if Path("userbot.session").exists() else "")
if not session_str:
print("❌ نشست اکانت دانلودر تنظیم نشده است.")
return
client = TelegramClient(StringSession(session_str), conf["api_id"], conf["api_hash"])
await client.connect()
if not await client.is_user_authorized():
print("❌ نشست اکانت دانلودر معتبر نیست.")
await client.disconnect()
return
try:
cur_data = json.loads(sync_file.read_text(encoding="utf-8"))
cur_data["pid"] = os.getpid()
sync_file.write_text(json.dumps(cur_data, ensure_ascii=False, indent=2), encoding="utf-8")
except Exception:
pass
print(f"🚀 شروع همگام‌سازی کانال {channel_id} با {total_videos} ویدیو (PID: {os.getpid()})...")
start_time = time.time()
done_count = 0
total_bytes = sum(it.get("file_size", 0) for it in items)
done_bytes = 0
for idx, item in enumerate(items):
# بررسی لغو از طریق کاربر
cur_data = json.loads(sync_file.read_text(encoding="utf-8"))
if cur_data.get("cancelled"):
print("🛑 عملیات همگام‌سازی متوقف شد.")
break
if item.get("status") == "completed":
done_count += 1
done_bytes += item.get("file_size", 0)
continue
item_task_id = f"sync_{channel_id}_{item['message_id']}"
task_info = {
"task_id": item_task_id,
"user_chat_id": chat_id,
"status_message_id": status_msg_id,
"source_chat_id": channel_id,
"source_message_id": item["message_id"],
"forwarded_message_id": 0,
"video_title": item.get("part_title", f"پارت {item.get('part_num', 1)}"),
"clean_ext": "mp4",
"site_url": data.get("site_url", ""),
"file_size": item.get("file_size", 0),
"original_name": item.get("file_name", f"vid_{item['message_id']}.mp4"),
"session_id": item.get("session_id", 0),
"session_num": item.get("session_num", 1),
"part_num": item.get("part_num", 1),
"is_channel": 1,
"is_sync": 1,
"sync_task_id": data.get("sync_task_id", ""),
"sync_index": idx + 1,
"sync_total": total_videos,
"sync_done_count": done_count,
"sync_done_bytes": done_bytes,
"sync_total_bytes": total_bytes,
"sync_start_time": start_time,
"channel_title": data.get("channel_title", ""),
"course_title": data.get("course_title", ""),
}
try:
success = await download_task_media(client, task_info)
except asyncio.CancelledError:
print("🛑 عملیات همگام‌سازی به دستور کاربر متوقف شد.")
break
except Exception as e:
print(f"⚠️ خطای پردازش پارت {idx+1}: {e}")
success = False
# بررسی مجدد وضعیت لغو بلافاصله پس از اتمام پارت
cur_data = json.loads(sync_file.read_text(encoding="utf-8"))
if cur_data.get("cancelled"):
print("🛑 عملیات همگام‌سازی متوقف شد.")
break
if success:
done_count += 1
done_bytes += item.get("file_size", 0)
cur_data["items"][idx]["status"] = "completed"
cur_data["done_count"] = done_count
cur_data["done_bytes"] = done_bytes
sync_file.write_text(json.dumps(cur_data, ensure_ascii=False, indent=2), encoding="utf-8")
else:
print(f"⚠️ خطای پردازش پارت {idx+1}")
if done_count >= total_videos:
total_duration = time.time() - start_time
call_internal_api("sync_all_completed", {
"sync_task_id": data.get("sync_task_id", ""),
"chat_id": chat_id,
"status_msg_id": status_msg_id,
"channel_id": channel_id,
"channel_title": data.get("channel_title", ""),
"course_title": data.get("course_title", ""),
"total_videos": total_videos,
"total_bytes": total_bytes,
"total_duration": total_duration,
"total_sessions": data.get("total_sessions", 1)
})
print(f"🎉 همگام‌سازی تمامی {total_videos} ویدیوی کانال با موفقیت کامل شد!")
await client.disconnect()
def run_coroutine(coro):
if hasattr(asyncio, "run"):
return asyncio.run(coro)
try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop.run_until_complete(coro)
def main():
parser = argparse.ArgumentParser(description="Telegram Userbot Downloader for Large Videos")
parser.add_argument("--login", action="store_true", help="Login to Telegram account interactively")
parser.add_argument("--daemon", action="store_true", help="Run background worker daemon")
parser.add_argument("--cron", action="store_true", help="Process pending queue and exit (for cPanel Cron)")
parser.add_argument("--task", type=str, help="Process a specific task ID")
parser.add_argument("--queue-file", type=str, help="Process a specific queue file path")
parser.add_argument("--status", action="store_true", help="Check account connection status (JSON)")
parser.add_argument("--send-code", type=str, help="Send login verification code to phone (Web Auth)")
parser.add_argument("--verify-code", type=str, help="Verify login code (Web Auth)")
parser.add_argument("--password", type=str, help="2FA password if required")
parser.add_argument("--scan-channel", type=int, help="Scan channel history and output structure")
parser.add_argument("--scan-file", type=str, help="Output file path for scan result")
parser.add_argument("--scan-msg-id", type=int, default=0, help="Status message ID for scan")
parser.add_argument("--admin-chat", type=int, default=0, help="Admin chat ID for scan")
parser.add_argument("--run-sync", type=str, help="Run full channel sync task file")
args = parser.parse_args()
conf = load_php_config()
if args.scan_channel:
run_coroutine(scan_channel_history(conf, args.scan_channel, out_file=args.scan_file, status_msg_id=args.scan_msg_id, admin_chat_id=args.admin_chat))
elif args.run_sync:
run_coroutine(run_channel_sync(conf, args.run_sync))
elif args.status:
run_coroutine(check_status(conf))
elif args.send_code:
run_coroutine(web_send_code(conf, args.send_code))
elif args.verify_code:
run_coroutine(web_verify_code(conf, args.verify_code, password=args.password))
elif args.login:
run_coroutine(run_login(conf))
elif args.daemon:
run_coroutine(run_daemon(conf))
elif args.task or args.queue_file:
run_coroutine(process_queue(conf, specific_task_id=args.task, queue_file=args.queue_file))
elif args.cron:
run_coroutine(process_queue(conf))
else:
run_coroutine(process_queue(conf))
if __name__ == "__main__":
main()