Import source: rub

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.

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()