From 03b7c2696c81b964413349d4caa72a638bf06934 Mon Sep 17 00:00:00 2001 From: arsam Date: Fri, 18 Sep 2026 23:14:26 +0330 Subject: [PATCH] Import source: rub Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality. --- rub.py | 731 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 731 insertions(+) create mode 100644 rub.py diff --git a/rub.py b/rub.py new file mode 100644 index 0000000..bb5b5c1 --- /dev/null +++ b/rub.py @@ -0,0 +1,731 @@ +import json +import os +import re +import shutil +import subprocess +import time +from datetime import datetime +from pathlib import Path +from typing import Optional +from urllib.parse import urlparse + +import pyzipper +import requests +from dotenv import load_dotenv + +load_dotenv() + +BASE_DIR = Path(__file__).resolve().parent +DOWNLOAD_DIR = BASE_DIR / "downloads" +QUEUE_DIR = BASE_DIR / "queue" +QUEUE_FILE = QUEUE_DIR / "tasks.jsonl" +PROCESSING_FILE = QUEUE_DIR / "processing.json" +FAILED_FILE = QUEUE_DIR / "failed.jsonl" +STATUS_FILE = QUEUE_DIR / "status.jsonl" +URL_DIR = DOWNLOAD_DIR / "url" +CANCEL_FILE = QUEUE_DIR / "cancelled.jsonl" + +RCLONE_REMOTE = os.getenv("RCLONE_REMOTE", "coaxys:coaxys").strip().rstrip("/") +RCLONE_PREFIX = os.getenv("RCLONE_PREFIX", "").strip().strip("/") +RCLONE_TRANSFERS = os.getenv("RCLONE_TRANSFERS", "8").strip() +RCLONE_CHECKERS = os.getenv("RCLONE_CHECKERS", "16").strip() +RCLONE_S3_UPLOAD_CONCURRENCY = os.getenv("RCLONE_S3_UPLOAD_CONCURRENCY", "8").strip() +RCLONE_S3_CHUNK_SIZE = os.getenv("RCLONE_S3_CHUNK_SIZE", "32M").strip() +RCLONE_S3_UPLOAD_CUTOFF = os.getenv("RCLONE_S3_UPLOAD_CUTOFF", "16M").strip() +RCLONE_MULTI_THREAD_STREAMS = os.getenv("RCLONE_MULTI_THREAD_STREAMS", "8").strip() +DOWNLOAD_LINK_EXPIRE = os.getenv("DOWNLOAD_LINK_EXPIRE", "168h").strip() +RCLONE_RETRIES = os.getenv("RCLONE_RETRIES", "3").strip() +RCLONE_LOW_LEVEL_RETRIES = os.getenv("RCLONE_LOW_LEVEL_RETRIES", "20").strip() +RCLONE_NO_OUTPUT_TIMEOUT = int(os.getenv("RCLONE_NO_OUTPUT_TIMEOUT", "180") or 180) +RCLONE_COMMIT_TIMEOUT = int(os.getenv("RCLONE_COMMIT_TIMEOUT", "90") or 90) + +MAX_RETRIES = int(os.getenv("MAX_RETRIES", "5") or 5) +UPLOAD_TIMEOUT = int(os.getenv("UPLOAD_TIMEOUT", "0") or 0) + +DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True) +QUEUE_DIR.mkdir(parents=True, exist_ok=True) +URL_DIR.mkdir(parents=True, exist_ok=True) + + +def log_event(message: str) -> None: + timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") + print(f"[{timestamp}] {message}", flush=True) + + +def safe_filename(name: Optional[str]) -> str: + name = (name or "file").strip() + name = re.sub(r'[<>:"/\\|?*\x00-\x1F]', "_", name) + name = name.rstrip(". ") + return name[:200] or "file" + + +def safe_s3_key_name(name: Optional[str]) -> str: + name = safe_filename(name) + name = re.sub(r"[^A-Za-z0-9._-]", "_", name) + name = re.sub(r"_+", "_", name) + return name[:200] or "file" + + +def pretty_size(size) -> str: + size = float(size or 0) + units = ["B", "KB", "MB", "GB", "TB"] + 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" + + +def progress_bar(percent: float, length: int = 12) -> str: + filled = int(length * percent / 100) + return "█" * filled + "░" * (length - filled) + + +def push_status(task: dict, text: str, status: str = "working", percent: float | None = None): + payload = { + "chat_id": task.get("chat_id"), + "message_id": task.get("status_message_id"), + "status": status, + "text": text, + "percent": percent, + "time": time.time(), + } + with open(STATUS_FILE, "a", encoding="utf-8") as file: + file.write(json.dumps(payload, ensure_ascii=False) + "\n") + + job_id = task.get("job_id", "-") + first_line = text.splitlines()[0] if text else status + if status != "uploading" or percent in (None, 0, 100): + log_event(f"job={job_id} status={status} percent={percent} message={first_line}") + + +def is_cancelled(task: dict) -> bool: + job_id = str(task.get("job_id", "")) + if not job_id or not CANCEL_FILE.exists(): + return False + with open(CANCEL_FILE, "r", encoding="utf-8") as file: + for line in file: + if not line.strip(): + continue + item = json.loads(line) + if str(item.get("job_id")) == job_id: + return True + return False + + +def unique_path(path: Path) -> Path: + if not path.exists(): + return path + stem = path.stem + suffix = path.suffix + index = 1 + while True: + candidate = path.with_name(f"{stem}_{index}{suffix}") + if not candidate.exists(): + return candidate + index += 1 + + +def require_rclone() -> None: + if not shutil.which("rclone"): + raise RuntimeError("rclone نصب نیست یا در PATH پیدا نشد.") + + +def remote_object_path(file_path: Path) -> str: + name = safe_s3_key_name(file_path.name) + return f"{RCLONE_PREFIX}/{name}" if RCLONE_PREFIX else name + + +def remote_destination(file_path: Path) -> tuple[str, str]: + object_path = remote_object_path(file_path) + return f"{RCLONE_REMOTE}/{object_path}", object_path + + +def rclone_base_flags() -> list[str]: + return [ + "--s3-no-check-bucket", + "--transfers", RCLONE_TRANSFERS, + "--checkers", RCLONE_CHECKERS, + "--s3-upload-concurrency", RCLONE_S3_UPLOAD_CONCURRENCY, + "--s3-chunk-size", RCLONE_S3_CHUNK_SIZE, + "--s3-upload-cutoff", RCLONE_S3_UPLOAD_CUTOFF, + "--s3-disable-checksum", + "--s3-no-head", + "--ignore-checksum", + "--multi-thread-streams", RCLONE_MULTI_THREAD_STREAMS, + "--multi-thread-chunk-size", RCLONE_S3_CHUNK_SIZE, + "--stats", "2s", + "--stats-one-line", + "--stats-log-level", "INFO", + "--retries", RCLONE_RETRIES, + "--retries-sleep", "2s", + "--low-level-retries", RCLONE_LOW_LEVEL_RETRIES, + "--contimeout", "30s", + "--timeout", "5m", + ] + + +def build_remote_path(object_path: str) -> str: + return f"{RCLONE_REMOTE}/{object_path}" + + +def extract_percent(line: str) -> float | None: + matches = re.findall(r"(\d+(?:\.\d+)?)%", line) + if not matches: + return None + try: + return min(float(matches[-1]), 100.0) + except ValueError: + return None + + +def progress_segments(buffer: str) -> tuple[list[str], str]: + normalized = buffer.replace("\r", "\n") + parts = normalized.split("\n") + return [part.strip() for part in parts[:-1] if part.strip()], parts[-1] + + +def read_new_log_lines(log_path: Path, pos: int) -> tuple[list[str], int]: + if not log_path.exists(): + return [], pos + with open(log_path, "r", encoding="utf-8", errors="replace") as file: + file.seek(pos) + lines = [line.strip() for line in file.readlines() if line.strip()] + return lines, file.tell() + + +def upload_once(file_path: Path, task: dict, attempt: int) -> str: + destination, object_path = remote_destination(file_path) + log_path = QUEUE_DIR / f"rclone-{task.get('job_id')}-{attempt}.log" + try: + if log_path.exists(): + log_path.unlink() + except Exception: + pass + + command = [ + "rclone", + "copyto", + str(file_path), + destination, + *rclone_base_flags(), + "--log-file", str(log_path), + "--log-level", "INFO", + ] + + started = time.time() + last_status = 0.0 + last_output_at = started + last_output = "" + log_pos = 0 + completed_at = None + completion_notice_sent = False + + push_status( + task, + f"آپلود در حال شروع است\n\n" + f"فایل: `{file_path.name}`\n" + f"مسیر مقصد: `{object_path}`\n" + f"تلاش: `{attempt}/{MAX_RETRIES}`\n" + f"لغو: `/del {task.get('job_id')}`", + "uploading", + 0, + ) + log_event( + f"job={task.get('job_id')} rclone copyto {file_path} {destination} " + f"concurrency={RCLONE_S3_UPLOAD_CONCURRENCY} chunk={RCLONE_S3_CHUNK_SIZE} " + f"cutoff={RCLONE_S3_UPLOAD_CUTOFF}" + ) + + proc = subprocess.Popen( + command, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + text=True, + encoding="utf-8", + errors="replace", + bufsize=1, + ) + + try: + while True: + if is_cancelled(task): + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + raise RuntimeError("آپلود لغو شد.") + + if UPLOAD_TIMEOUT and time.time() - started > UPLOAD_TIMEOUT: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + raise RuntimeError("آپلود بیشتر از حد مجاز طول کشید و لغو شد.") + + lines, log_pos = read_new_log_lines(log_path, log_pos) + if lines: + last_output_at = time.time() + for line in lines: + last_output = line + percent = extract_percent(line) + now = time.time() + if percent is not None and percent >= 100 and completed_at is None: + completed_at = now + if percent is not None and percent >= 100 and completion_notice_sent: + continue + if percent is not None and (now - last_status >= 2 or percent >= 100): + last_status = now + completion_notice_sent = percent >= 100 + if percent >= 100: + text = ( + f"داده‌ها کامل ارسال شد؛ منتظر تایید نهایی S3 هستم\n\n" + f"فایل: `{file_path.name}`\n" + f"مسیر مقصد: `{object_path}`\n" + f"این مرحله مربوط به commit/finalize سمت S3 است.\n" + f"اگر S3 پاسخ ندهد، تا `{RCLONE_COMMIT_TIMEOUT}` ثانیه دیگر تلاش بعدی شروع می‌شود." + ) + else: + text = ( + f"در حال آپلود روی فضای Coaxys\n\n" + f"فایل: `{file_path.name}`\n" + f"مسیر مقصد: `{object_path}`\n" + f"پیشرفت: `{percent:.1f}%`\n" + f"`{progress_bar(percent)}`\n" + f"لغو: `/del {task.get('job_id')}`" + ) + push_status( + task, + text, + "uploading", + percent, + ) + continue + + if completed_at and RCLONE_COMMIT_TIMEOUT and time.time() - completed_at > RCLONE_COMMIT_TIMEOUT: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + raise RuntimeError( + f"داده‌ها کامل ارسال شدند، اما S3 تا {RCLONE_COMMIT_TIMEOUT} ثانیه پاسخ نهایی نداد. تلاش دوباره انجام می‌شود." + ) + + if RCLONE_NO_OUTPUT_TIMEOUT and time.time() - last_output_at > RCLONE_NO_OUTPUT_TIMEOUT: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + raise RuntimeError( + f"rclone بیشتر از {RCLONE_NO_OUTPUT_TIMEOUT} ثانیه هیچ خروجی نداد؛ تلاش دوباره انجام می‌شود." + ) + + if proc.poll() is not None: + break + time.sleep(0.2) + + lines, log_pos = read_new_log_lines(log_path, log_pos) + for line in lines: + last_output = line + + if proc.returncode != 0: + message = last_output or f"rclone با کد {proc.returncode} متوقف شد." + if "badgateway" in message.lower() or "bad gateway" in message.lower() or "502" in message: + message += ( + "\nS3 خطای 502 داد. آپلود دوباره با multipart و retry انجام می‌شود." + ) + raise RuntimeError(message) + finally: + pass + + verify_upload(object_path, task) + return object_path + + +def verify_upload(object_path: str, task: dict) -> None: + push_status(task, "آپلود تمام شد؛ در حال بررسی فایل روی S3 ...", "verifying") + + last_error = "" + for attempt in range(1, 31): + if is_cancelled(task): + raise RuntimeError("آپلود لغو شد.") + + result = subprocess.run( + ["rclone", "ls", RCLONE_REMOTE, "--s3-no-check-bucket"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + timeout=120, + ) + + if result.returncode != 0: + last_error = result.stderr.strip() or "بررسی فایل در S3 ناموفق بود." + log_event(f"job={task.get('job_id')} verify attempt={attempt}/30 error={last_error}") + else: + for line in result.stdout.splitlines(): + parts = line.strip().split(maxsplit=1) + if len(parts) == 2 and parts[1] == object_path: + log_event(f"job={task.get('job_id')} verified object={object_path}") + return + last_error = "فایل هنوز در لیست S3 دیده نمی‌شود." + log_event(f"job={task.get('job_id')} verify attempt={attempt}/30 pending object={object_path}") + + if attempt in (5, 15): + push_status( + task, + f"آپلود تمام شده؛ منتظر نمایش فایل در S3 هستم ...\n" + f"بررسی: `{attempt}/30`", + "verifying", + ) + time.sleep(2) + + raise RuntimeError(last_error or "آپلود کامل شد ولی فایل در خروجی rclone ls پیدا نشد.") + + +def create_download_link(object_path: str, task: dict) -> str: + remote_path = build_remote_path(object_path) + last_error = "" + + push_status( + task, + f"فایل تایید شد؛ در حال ساخت لینک دانلود {DOWNLOAD_LINK_EXPIRE} ...", + "linking", + ) + + for attempt in range(1, 6): + result = subprocess.run( + [ + "rclone", + "link", + "--expire", + DOWNLOAD_LINK_EXPIRE, + remote_path, + "--s3-no-check-bucket", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + timeout=120, + ) + + if result.returncode == 0: + lines = [line.strip() for line in result.stdout.splitlines() if line.strip()] + if lines: + link = lines[-1] + log_event(f"job={task.get('job_id')} download_link_created expire={DOWNLOAD_LINK_EXPIRE}") + return link + + last_error = result.stderr.strip() or result.stdout.strip() or "خروجی لینک خالی بود." + log_event(f"job={task.get('job_id')} link attempt={attempt}/5 error={last_error}") + time.sleep(2) + + raise RuntimeError(f"آپلود انجام شد، اما ساخت لینک دانلود ناموفق بود: {last_error}") + + +def upload_with_retry(file_path: Path, task: dict) -> str: + last_error = None + start_time = time.time() + + for attempt in range(1, MAX_RETRIES + 1): + if UPLOAD_TIMEOUT and time.time() - start_time > UPLOAD_TIMEOUT: + raise RuntimeError("آپلود بیشتر از حد مجاز طول کشید و لغو شد.") + if is_cancelled(task): + raise RuntimeError("آپلود لغو شد.") + + try: + return upload_once(file_path, task, attempt) + except Exception as e: + last_error = e + log_event(f"job={task.get('job_id')} attempt={attempt}/{MAX_RETRIES} error={e}") + error_text = str(e).lower() + transient = any( + key in error_text + for key in [ + "429", "500", "502", "503", "504", + "bad gateway", "timeout", "io timeout", + "cannot connect", "connection reset", + "connection refused", "connection timed out", + "temporarily unavailable", "broken pipe", + "error uploading chunk", + "پاسخ نهایی نداد", "تلاش دوباره", + ] + ) + + if transient and attempt < MAX_RETRIES: + if is_cancelled(task): + raise RuntimeError("آپلود لغو شد.") + push_status( + task, + f"ارتباط با S3 ناپایدار بود.\n" + f"تلاش بعدی: `{attempt + 1}/{MAX_RETRIES}`", + "uploading", + ) + time.sleep(3) + continue + break + + raise last_error if last_error else RuntimeError("Upload failed.") + + +def download_url(task: dict) -> Path: + url = task.get("url", "").strip() + if not url: + raise RuntimeError("URL خالیه") + + push_status(task, "دانلود لینک شروع شد ...", "downloading", 0) + + try: + resp = requests.get(url, stream=True, timeout=(10, 60), allow_redirects=True) + resp.raise_for_status() + except requests.exceptions.Timeout: + raise RuntimeError("لینک جواب نداد") + except requests.exceptions.ConnectionError: + raise RuntimeError("مشکل شبکه") + except requests.exceptions.HTTPError as e: + code = e.response.status_code if e.response else "نامشخص" + raise RuntimeError(f"دانلود انجام نشد. کد خطا: {code}") + + cd = resp.headers.get("content-disposition", "") + match = re.findall(r'filename="(.+?)"', cd) + name = match[0] if match else Path(urlparse(url).path).name + name = safe_filename(name or f"file_{int(time.time())}") + if "." not in name: + name += ".bin" + + target = unique_path(URL_DIR / name) + total = int(resp.headers.get("content-length") or 0) + downloaded, last_update, started = 0, 0, time.time() + + try: + with open(target, "wb") as file: + for chunk in resp.iter_content(1024 * 1024): + if is_cancelled(task): + raise RuntimeError("دانلود لغو شد.") + if not chunk: + continue + file.write(chunk) + downloaded += len(chunk) + + now = time.time() + if now - last_update < 3 and downloaded < total: + continue + last_update = now + + speed = downloaded / max(now - started, 1) + eta = (total - downloaded) / speed if total and speed else None + percent = downloaded * 100 / total if total else None + + text = f"در حال دانلود از لینک مستقیم\n\nدریافت شده: `{pretty_size(downloaded)}`" + if total: + text += f" از `{pretty_size(total)}`" + text += f"\nسرعت: `{pretty_size(speed)}/s`" + if eta: + text += f"\nزمان باقی‌مانده: `{eta_text(eta)}`" + + push_status(task, text, "downloading", percent) + except Exception: + try: + if target.exists(): + target.unlink() + except Exception: + pass + raise + + if not target.exists() or target.stat().st_size == 0: + raise RuntimeError("فایل دانلود نشد") + + task["file_name"] = target.name + task["file_size"] = target.stat().st_size + return target + + +def make_zip_with_password(file_path: Path, password: str) -> Path: + zip_path = unique_path(file_path.with_suffix(file_path.suffix + ".zip")) + with pyzipper.AESZipFile( + zip_path, + "w", + compression=pyzipper.ZIP_STORED, + encryption=pyzipper.WZ_AES, + ) as zip_file: + zip_file.setpassword(password.encode("utf-8")) + zip_file.write(file_path, arcname=file_path.name) + return zip_path + + +def pop_first_task(): + if not QUEUE_FILE.exists(): + return None + with open(QUEUE_FILE, "r", encoding="utf-8") as file: + lines = [line for line in file if line.strip()] + if not lines: + return None + + first_line = lines[0] + remaining = lines[1:] + with open(QUEUE_FILE, "w", encoding="utf-8") as file: + file.writelines(remaining) + return json.loads(first_line) + + +def recover_processing_task() -> None: + if not PROCESSING_FILE.exists(): + return + + try: + task = json.loads(PROCESSING_FILE.read_text(encoding="utf-8")) + except Exception as exc: + log_event(f"could not read processing task for recovery: {exc}") + return + + job_id = str(task.get("job_id", "")) + if not job_id: + return + + try: + existing = [] + if QUEUE_FILE.exists(): + with open(QUEUE_FILE, "r", encoding="utf-8") as file: + existing = [json.loads(line) for line in file if line.strip()] + if any(str(item.get("job_id")) == job_id for item in existing): + return + + local_path = Path(task.get("path", "")) + if task.get("type") == "local_file" and not local_path.exists(): + log_event(f"job={job_id} not recovered because local file is missing: {local_path}") + return + + with open(QUEUE_FILE, "a", encoding="utf-8") as file: + file.write(json.dumps(task, ensure_ascii=False) + "\n") + PROCESSING_FILE.unlink() + push_status( + task, + "پردازش قبلی متوقف شده بود؛ فایل دوباره به صف برگشت و ادامه داده می‌شود.", + "queued", + ) + log_event(f"job={job_id} recovered from processing.json") + except Exception as exc: + log_event(f"job={job_id} recovery failed: {exc}") + + +def save_processing(task: dict) -> None: + with open(PROCESSING_FILE, "w", encoding="utf-8") as file: + json.dump(task, file, ensure_ascii=False, indent=2) + + +def clear_processing() -> None: + if PROCESSING_FILE.exists(): + PROCESSING_FILE.unlink() + + +def append_failed(task: dict, error: str) -> None: + payload = {"task": task, "error": error} + with open(FAILED_FILE, "a", encoding="utf-8") as file: + file.write(json.dumps(payload, ensure_ascii=False) + "\n") + + +def process_task(task: dict): + task_type = task.get("type") + safe_mode = task.get("safe_mode", False) + zip_password = task.get("zip_password", "") + local_path: Path | None = None + send_path: Path | None = None + uploaded = False + + if task_type == "local_file": + local_path = Path(task.get("path", "")) + if not local_path.exists(): + raise RuntimeError("فایل محلی پیدا نشد.") + elif task_type == "direct_url": + local_path = download_url(task) + else: + raise RuntimeError("نوع تسک نامعتبر است.") + + log_event( + f"job={task.get('job_id')} processing type={task_type} " + f"file={local_path.name if local_path else '-'} size={local_path.stat().st_size if local_path and local_path.exists() else 0}" + ) + + if safe_mode and zip_password: + push_status(task, "Safe Mode فعال است؛ در حال ساخت ZIP رمزدار ...", "processing") + zipped = make_zip_with_password(local_path, zip_password) + try: + if local_path.exists(): + local_path.unlink() + except Exception: + pass + send_path = zipped + else: + send_path = local_path + + try: + if is_cancelled(task): + raise RuntimeError("آپلود لغو شد.") + + object_path = upload_with_retry(send_path, task) + download_link = create_download_link(object_path, task) + uploaded = True + push_status( + task, + f"آپلود کامل شد\n\n" + f"فایل: `{send_path.name}`\n" + f"حجم: `{pretty_size(send_path.stat().st_size)}`\n" + f"مسیر ذخیره‌سازی: `{object_path}`\n" + f"اعتبار لینک: `{DOWNLOAD_LINK_EXPIRE}`\n\n" + f"لینک دانلود:\n{download_link}", + "done", + 100, + ) + finally: + try: + if uploaded and send_path and send_path.exists(): + send_path.unlink() + except Exception: + pass + + +def worker_loop(): + require_rclone() + log_event("S3 rclone worker started.") + recover_processing_task() + + while True: + task = pop_first_task() + if not task: + time.sleep(0.2) + continue + + save_processing(task) + try: + process_task(task) + except Exception as e: + log_event(f"job={task.get('job_id')} failed error={e}") + append_failed(task, str(e)) + push_status(task, f"خطا در پردازش فایل:\n{str(e)}", "failed") + finally: + clear_processing() + + +if __name__ == "__main__": + worker_loop()