Spaces:

Sada8888
/

apk

Sleeping

apk / builder.py
Transfer Bot
Moved to Hugging Face automatically
d522aa8 about 2 months ago
25.6 kB
"""
=============================================================================
 builder.py — موتورِ ساختِ APK داخلِ خودِ اسپیس
=============================================================================
کاری که این فایل می‌کنه:
  • یه build رو توی یه ترد و پروسه‌ی جدا (Gradle) اجرا می‌کنه؛ چون کار
    سمتِ سرور انجام می‌شه، بستنِ مرورگر یا ترکِ صفحه هیچ اثری روش نداره.
  • وضعیت و لاگ رو روی دیسک می‌نویسه، پس هر وقت برگردید همون‌جا ادامه‌ی
    ماجرا رو می‌بینید.
  • APKهای ساخته‌شده رو نگه می‌داره؛ حتی وسطِ یه build جدید، نسخه‌ی قبلی
    همچنان قابلِ دانلوده.
  • کلیدِ امضا (keystore) رو مدیریت می‌کنه تا همه‌ی نسخه‌ها با یه کلید
    امضا بشن و روی هم آپدیت بشن.
=============================================================================
"""

import base64
import json
import os
import re
import shutil
import signal
import subprocess
import threading
import time
import uuid
from datetime import datetime, timezone

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
ANDROID_DIR = os.path.join(BASE_DIR, "android")
CONFIG_PATH = os.path.join(BASE_DIR, "config.json")

APK_OUT = os.path.join(ANDROID_DIR, "app", "build", "outputs", "apk", "release")


# --- محلِ نگهداریِ داده‌ها ----------------------------------------------------
def _pick_data_dir():
    """
    اگه اسپیس Persistent Storage داشته باشه (/data)، ازش استفاده می‌کنیم تا
    APKها و کلیدِ امضا بعد از ری‌استارتِ اسپیس هم بمونن. وگرنه می‌ریم سراغِ یه
    پوشه‌ی محلی که با ری‌استارتِ اسپیس پاک می‌شه (توی README توضیح داده شده).
    """
    for candidate in ("/data", os.path.join(BASE_DIR, "workspace")):
        try:
            os.makedirs(candidate, exist_ok=True)
            probe = os.path.join(candidate, ".write_test")
            with open(probe, "w") as f:
                f.write("ok")
            os.remove(probe)
            return candidate
        except OSError:
            continue
    return "/tmp/zirnavis"


DATA_DIR = _pick_data_dir()
IS_PERSISTENT = DATA_DIR == "/data"

RELEASES_DIR = os.path.join(DATA_DIR, "releases")
LOG_DIR = os.path.join(DATA_DIR, "logs")
KEYSTORE_DIR = os.path.join(DATA_DIR, "keystore")
STATE_PATH = os.path.join(DATA_DIR, "state.json")
RELEASES_PATH = os.path.join(DATA_DIR, "releases.json")
KEYSTORE_PATH = os.path.join(KEYSTORE_DIR, "zirnavis.jks")
KEYSTORE_META = os.path.join(KEYSTORE_DIR, "meta.json")

for _d in (RELEASES_DIR, LOG_DIR, KEYSTORE_DIR):
    os.makedirs(_d, exist_ok=True)

MAX_RELEASES = 10  # فقط ۱۰ نسخه‌ی آخر نگه داشته می‌شه تا دیسک پر نشه

_lock = threading.Lock()
_current_proc = None
_cancel_requested = False


# --- کمکی‌ها ------------------------------------------------------------------
def _now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _read_json(path, default):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except (OSError, json.JSONDecodeError):
        return default


def _write_json(path, data):
    """
    اول توی یه فایلِ موقت می‌نویسه بعد جابه‌جا می‌کنه، تا اگه وسطِ کار برق
    رفت فایلِ اصلی نصفه‌نیمه نشه. روی باکتِ هاگینگ‌فیس (که یه فایل‌سیستمِ
    شبکه‌ایه) گاهی rename پشتیبانی نمی‌شه، برای همین یه راهِ دومم داریم.
    """
    text = json.dumps(data, ensure_ascii=False, indent=2)
    tmp = path + ".tmp"
    try:
        with open(tmp, "w", encoding="utf-8") as f:
            f.write(text)
        os.replace(tmp, path)
    except OSError:
        with open(path, "w", encoding="utf-8") as f:
            f.write(text)
        try:
            os.remove(tmp)
        except OSError:
            pass


def _gradle_defaults():
    """versionCode/versionName پیش‌فرض رو مستقیم از build.gradle.kts می‌خونه."""
    code, name = 1, "1.0.0"
    try:
        with open(os.path.join(ANDROID_DIR, "app", "build.gradle.kts"), "r", encoding="utf-8") as f:
            text = f.read()
        m = re.search(r'zirnavisVersionCode"\)\s*as\s*String\?\)\?\.toIntOrNull\(\)\s*\?:\s*(\d+)', text)
        if m:
            code = int(m.group(1))
        m = re.search(r'zirnavisVersionName"\)\s*as\s*String\?\)\s*\?:\s*"([^"]+)"', text)
        if m:
            name = m.group(1)
    except OSError:
        pass
    return code, name


# --- وضعیت -------------------------------------------------------------------
def _default_state():
    return {"status": "idle", "job_id": None, "version_name": None, "version_code": None,
            "note": "", "started_at": None, "finished_at": None, "step": "", "error": None,
            "log_file": None, "signed_release": False}


def get_state():
    return _read_json(STATE_PATH, _default_state())


def _set_state(**kw):
    state = get_state()
    state.update(kw)
    _write_json(STATE_PATH, state)
    return state


def _log_path(job_id):
    return os.path.join(LOG_DIR, f"{job_id}.log")


def _append_log(job_id, text):
    try:
        with open(_log_path(job_id), "a", encoding="utf-8", errors="replace") as f:
            f.write(text)
    except OSError:
        pass


def read_log(job_id, offset=0, limit=200_000):
    """از بایتِ offset به بعدِ لاگ رو می‌ده — برای نمایشِ زنده توی پنل."""
    path = _log_path(job_id) if job_id else None
    if not path or not os.path.exists(path):
        return {"offset": 0, "text": ""}
    size = os.path.getsize(path)
    if offset > size:  # لاگ ریست شده
        offset = 0
    with open(path, "rb") as f:
        f.seek(offset)
        chunk = f.read(limit)
    return {"offset": offset + len(chunk),
            "text": chunk.decode("utf-8", errors="replace")}


def recover_on_startup():
    """
    اگه اسپیس وسطِ یه build ری‌استارت شده باشه، پروسه‌ی گریدل هم مرده؛
    پس وضعیت رو از «در حال ساخت» به «نیمه‌کاره» تغییر می‌دیم تا پنل گیر نکنه.
    """
    state = get_state()
    if state.get("status") == "running":
        _set_state(status="interrupted", finished_at=_now(),
                   error="اسپیس وسطِ ساخت ری‌استارت شد. لطفاً دوباره «ساخت APK» رو بزنید.")


# --- نسخه‌ها ------------------------------------------------------------------
def get_releases():
    items = _read_json(RELEASES_PATH, [])
    return [it for it in items if os.path.exists(os.path.join(RELEASES_DIR, it.get("file", "")))]


def latest_release():
    rels = get_releases()
    return rels[0] if rels else None


def next_version_code():
    rels = get_releases()
    base_code, _ = _gradle_defaults()
    highest = max([r.get("version_code", 0) for r in rels] + [base_code])
    return highest + 1


def suggest_version_name():
    rels = get_releases()
    if rels:
        current = rels[0].get("version_name", "")
    else:
        _, current = _gradle_defaults()
    parts = current.split(".")
    if len(parts) == 3 and parts[-1].isdigit():
        parts[-1] = str(int(parts[-1]) + 1)
        return ".".join(parts)
    return current


def _record_release(item):
    items = _read_json(RELEASES_PATH, [])
    items.insert(0, item)
    # نسخه‌های خیلی قدیمی رو پاک می‌کنیم
    for old in items[MAX_RELEASES:]:
        try:
            os.remove(os.path.join(RELEASES_DIR, old["file"]))
        except OSError:
            pass
    items = items[:MAX_RELEASES]
    _write_json(RELEASES_PATH, items)


def release_file_path(filename):
    safe = os.path.basename(filename)
    path = os.path.join(RELEASES_DIR, safe)
    return path if os.path.exists(path) else None


# --- کلیدِ امضا ---------------------------------------------------------------
def _keystore_from_env():
    """کلید رو از Secretهای اسپیس (KEYSTORE_BASE64 و ...) می‌سازه."""
    b64 = os.environ.get("KEYSTORE_BASE64", "").strip()
    password = os.environ.get("KEYSTORE_PASSWORD", "").strip()
    alias = os.environ.get("KEY_ALIAS", "zirnavis").strip()
    key_password = os.environ.get("KEY_PASSWORD", "").strip() or password
    if not b64 or not password:
        return None
    try:
        raw = base64.b64decode(b64)
    except Exception:
        return None
    path = os.path.join(KEYSTORE_DIR, "from_secret.jks")
    with open(path, "wb") as f:
        f.write(raw)
    return {"path": path, "password": password, "alias": alias,
            "key_password": key_password, "source": "secret"}


def _keystore_from_disk():
    if not os.path.exists(KEYSTORE_PATH):
        return None
    meta = _read_json(KEYSTORE_META, None)
    if not meta:
        return None
    return {"path": KEYSTORE_PATH, "password": meta["password"], "alias": meta["alias"],
            "key_password": meta.get("key_password", meta["password"]),
            "source": "uploaded" if meta.get("imported") else "generated"}


def resolve_keystore():
    return _keystore_from_env() or _keystore_from_disk()


def _keystore_fingerprint(path, password, alias):
    """اثرِ انگشتِ SHA-256 کلید — برای اینکه مطمئن بشید همون کلیدِ همیشگیه."""
    try:
        proc = subprocess.run(
            ["keytool", "-list", "-v", "-keystore", path, "-storepass", password,
             "-alias", alias, "-J-Duser.language=en"],
            capture_output=True, text=True, timeout=60)
        m = re.search(r"SHA-?256:\s*([0-9A-Fa-f:]{40,})", proc.stdout)
        return m.group(1) if m else None
    except (OSError, subprocess.TimeoutExpired):
        return None


def _keystore_aliases(path, password):
    """(ok, aliases یا پیامِ خطا)"""
    try:
        proc = subprocess.run(
            ["keytool", "-list", "-keystore", path, "-storepass", password,
             "-J-Duser.language=en"],
            capture_output=True, text=True, timeout=60)
    except (OSError, subprocess.TimeoutExpired) as exc:
        return False, f"اجرای keytool ممکن نشد: {exc}"
    out = (proc.stdout or "") + (proc.stderr or "")
    if proc.returncode != 0:
        if "password was incorrect" in out or "Keystore was tampered" in out:
            return False, "رمزِ فایلِ کلید اشتباهه."
        return False, out.strip()[-400:] or "فایلِ کلید معتبر نیست."
    aliases = []
    for line in out.splitlines():
        if "PrivateKeyEntry" in line:
            aliases.append(line.split(",")[0].strip())
    if not aliases:
        return False, "توی این فایل هیچ کلیدِ خصوصی‌ای پیدا نشد."
    return True, aliases


def import_keystore(file_bytes, password, alias=None):
    """
    فایلِ .jks آپلودشده رو بررسی و ذخیره می‌کنه. اگه رمز درست باشه، از این
    به بعد همه‌ی APKها با همین کلید امضا می‌شن — یعنی دقیقاً مثلِ نسخه‌های
    قبلی، و روی همدیگه نصب می‌شن.
    """
    if not file_bytes:
        return False, "فایلی انتخاب نشده."
    if not password:
        return False, "رمزِ فایلِ کلید رو وارد کنید."
    tmp_path = os.path.join(KEYSTORE_DIR, "upload.tmp")
    with open(tmp_path, "wb") as f:
        f.write(file_bytes)

    ok, result = _keystore_aliases(tmp_path, password)
    if not ok:
        try:
            os.remove(tmp_path)
        except OSError:
            pass
        return False, result

    aliases = result
    chosen = (alias or "").strip() or aliases[0]
    if chosen not in aliases:
        try:
            os.remove(tmp_path)
        except OSError:
            pass
        return False, f"aliasِ «{chosen}» توی فایل نیست. موجودها: {'، '.join(aliases)}"

    os.replace(tmp_path, KEYSTORE_PATH)
    _write_json(KEYSTORE_META, {
        "alias": chosen, "password": password, "key_password": password,
        "created_at": _now(), "imported": True,
        "fingerprint": _keystore_fingerprint(KEYSTORE_PATH, password, chosen),
    })
    return True, f"کلید با موفقیت بارگذاری شد (alias: {chosen})."


def delete_keystore():
    removed = False
    for p in (KEYSTORE_PATH, KEYSTORE_META):
        try:
            os.remove(p)
            removed = True
        except OSError:
            pass
    return removed


def keystore_status():
    ks = resolve_keystore()
    if not ks:
        return {"ready": False, "source": None, "persistent": IS_PERSISTENT}
    meta = _read_json(KEYSTORE_META, {}) or {}
    return {"ready": True, "source": ks["source"], "alias": ks["alias"],
            "persistent": IS_PERSISTENT,
            "imported": bool(meta.get("imported")),
            "fingerprint": meta.get("fingerprint"),
            "saved_as_secret": ks["source"] == "secret"}


def generate_keystore(alias="zirnavis", password=None, dname=None):
    """
    یه کلیدِ امضای تازه با keytool می‌سازه. فقط یه بار در طولِ عمرِ برنامه
    باید انجام بشه — بعدش باید base64 همین فایل رو توی Secrets ذخیره کنید.
    """
    if os.path.exists(KEYSTORE_PATH):
        return False, "کلید از قبل ساخته شده. اگه واقعاً می‌خواید عوضش کنید، اول فایلِ فعلی رو پاک کنید."
    password = password or base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip("=")
    dname = dname or "CN=Zirnavis, OU=Mobile, O=Zirnavis, L=Tehran, C=IR"
    cmd = [
        "keytool", "-genkeypair", "-v",
        "-keystore", KEYSTORE_PATH,
        "-alias", alias,
        "-keyalg", "RSA", "-keysize", "2048",
        "-validity", "10950",              # حدود ۳۰ سال
        "-storepass", password,
        "-keypass", password,
        "-dname", dname,
        "-storetype", "JKS",
    ]
    try:
        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    except (OSError, subprocess.TimeoutExpired) as exc:
        return False, f"اجرای keytool ناموفق بود: {exc}"
    if proc.returncode != 0:
        return False, (proc.stderr or proc.stdout or "خطای نامشخص در keytool")[-800:]
    _write_json(KEYSTORE_META, {
        "alias": alias, "password": password, "key_password": password,
        "created_at": _now(), "imported": False,
        "fingerprint": _keystore_fingerprint(KEYSTORE_PATH, password, alias),
    })
    return True, "کلید ساخته شد."


def keystore_export():
    """base64 کلید + رمزها، برای ذخیره‌کردن توی Secretهای اسپیس."""
    ks = _keystore_from_disk()
    if not ks:
        return None
    with open(ks["path"], "rb") as f:
        b64 = base64.b64encode(f.read()).decode()
    return {"KEYSTORE_BASE64": b64, "KEYSTORE_PASSWORD": ks["password"],
            "KEY_ALIAS": ks["alias"], "KEY_PASSWORD": ks["key_password"]}


# --- آپلود اختیاری روی هاگینگ‌فیس --------------------------------------------
def _upload_to_hf(local_path, filename, log):
    """
    اگه HF_TOKEN و RELEASE_REPO ست شده باشن، APK روی یه ریپوی dataset آپلود
    می‌شه تا یه لینکِ دائمی داشته باشه (چون فایل‌های داخلِ خودِ اسپیس با
    ری‌استارت پاک می‌شن، مگه اینکه Persistent Storage داشته باشید).
    """
    token = os.environ.get("HF_TOKEN", "").strip()
    repo = os.environ.get("RELEASE_REPO", "").strip()
    if not token or not repo:
        return None
    try:
        from huggingface_hub import HfApi
        api = HfApi(token=token)
        api.create_repo(repo_id=repo, repo_type="dataset", exist_ok=True, private=False)
        api.upload_file(path_or_fileobj=local_path, path_in_repo=filename,
                        repo_id=repo, repo_type="dataset")
        url = f"https://huggingface.co/datasets/{repo}/resolve/main/{filename}"
        log(f"\n📤 روی هاگینگ‌فیس آپلود شد: {url}\n")
        return url
    except Exception as exc:  # noqa: BLE001
        log(f"\n⚠️ آپلود روی هاگینگ‌فیس ناموفق بود: {exc}\n")
        return None


def _update_config_json(version_code, apk_url, changelog, log):
    """بخشِ update داخلِ config.json رو با نسخه‌ی تازه هماهنگ می‌کنه."""
    try:
        import config_store  # importِ تنبل تا حلقه‌ی import پیش نیاد
        cfg = config_store.load()
        cfg.setdefault("update", {})
        cfg["update"]["latest_version_code"] = version_code
        if apk_url:
            cfg["update"]["apk_url"] = apk_url
        if changelog:
            cfg["update"]["changelog"] = changelog
        ok, msg = config_store.save(cfg)
        log("\n📝 بخشِ update داخلِ config.json به‌روزرسانی شد.\n" if ok
            else f"\n⚠️ به‌روزرسانیِ config.json ناموفق بود: {msg}\n")
    except Exception as exc:  # noqa: BLE001
        log(f"\n⚠️ به‌روزرسانیِ config.json ناموفق بود: {exc}\n")


# --- خودِ build ---------------------------------------------------------------
def start_build(version_name, version_code, note="", update_config=False, public_base=""):
    global _cancel_requested
    with _lock:
        if get_state().get("status") == "running":
            return False, "یه build دیگه همین الان در حال اجراست."
        try:
            version_code = int(version_code)
        except (TypeError, ValueError):
            return False, "versionCode باید عدد باشه."
        version_name = (version_name or "").strip()
        if not version_name:
            return False, "versionName خالیه."

        rels = get_releases()
        if any(r.get("version_code") == version_code for r in rels):
            return False, f"نسخه‌ای با versionCode={version_code} قبلاً ساخته شده. یه عددِ بزرگ‌تر بذارید."

        job_id = datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:6]
        _cancel_requested = False
        _set_state(status="running", job_id=job_id, version_name=version_name,
                   version_code=version_code, note=note, started_at=_now(),
                   finished_at=None, step="آماده‌سازی…", error=None,
                   log_file=job_id, signed_release=False)

        thread = threading.Thread(
            target=_run_build,
            args=(job_id, version_name, version_code, note, update_config, public_base),
            daemon=False,
        )
        thread.start()
        return True, job_id


def _run_build(job_id, version_name, version_code, note, update_config, public_base):
    global _current_proc

    def log(text):
        _append_log(job_id, text)

    started = time.time()
    log(f"══════════════════════════════════════════════\n")
    log(f"  ساختِ نسخه {version_name} (versionCode={version_code})\n")
    log(f"  شروع: {_now()}\n")
    log(f"══════════════════════════════════════════════\n\n")

    ks = resolve_keystore()
    env = os.environ.copy()

    # نکته: کشِ گریدل عمداً روی /data گذاشته نمی‌شه. باکتِ هاگینگ‌فیس یه
    # فایل‌سیستمِ شبکه‌ایه و گریدل ده‌ها هزار فایلِ کوچیک می‌سازه — روی شبکه
    # به‌شدت کند می‌شه. کش داخلِ خودِ کانتینر می‌مونه.

    # تنظیماتِ ساختنی (دسترسی‌ها، نامِ برنامه، آیکون) قبل از هر build روی
    # فایل‌های پروژه نوشته می‌شن.
    try:
        import app_settings
        ok_apply, msg_apply = app_settings.apply()
        log(("⚙️  تنظیماتِ برنامه اعمال شد (دسترسی‌ها، نام، آیکون).\n"
             if ok_apply else f"⚠️  اعمالِ تنظیماتِ برنامه: {msg_apply}\n"))
        st = app_settings.load()
        on = [k for k, v in st.get("permissions", {}).items() if v]
        log("    دسترسی‌های فعال: " + ("، ".join(on) if on else "هیچ‌کدام") + "\n\n")
    except Exception as exc:  # noqa: BLE001
        log(f"⚠️  اعمالِ تنظیماتِ برنامه ناموفق بود: {exc}\n\n")
    if ks:
        env["ZIRNAVIS_KEYSTORE"] = ks["path"]
        env["ZIRNAVIS_KEYSTORE_PASSWORD"] = ks["password"]
        env["ZIRNAVIS_KEY_ALIAS"] = ks["alias"]
        env["ZIRNAVIS_KEY_PASSWORD"] = ks["key_password"]
        log(f"🔑 امضا با کلیدِ اختصاصی (منبع: {ks['source']}, alias: {ks['alias']})\n\n")
    else:
        log("⚠️  کلیدِ امضای اختصاصی پیدا نشد؛ با کلیدِ debug امضا می‌شه.\n"
            "    این APK نصب می‌شه، ولی نسخه‌های بعدی ممکنه روش آپدیت نشن.\n"
            "    از بخشِ «کلید امضا» توی پنل، یه کلید بسازید.\n\n")

    _set_state(step="اجرای Gradle…")

    cmd = [
        "gradle", "--no-daemon", "--console=plain", "--stacktrace",
        "-p", ANDROID_DIR, "assembleRelease",
        f"-PzirnavisVersionCode={version_code}",
        f"-PzirnavisVersionName={version_name}",
    ]

    ok = False
    error = None
    try:
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                env=env, text=True, bufsize=1, cwd=ANDROID_DIR,
                                start_new_session=True)
        _current_proc = proc
        for line in proc.stdout:
            log(line)
        proc.wait()
        _current_proc = None
        if _cancel_requested:
            error = "ساخت توسط کاربر لغو شد."
        elif proc.returncode != 0:
            error = f"Gradle با کد {proc.returncode} شکست خورد. انتهای لاگ رو ببینید."
        else:
            ok = True
    except OSError as exc:
        error = f"اجرای Gradle ممکن نشد: {exc}"

    if not ok:
        _set_state(status="cancelled" if _cancel_requested else "error",
                   finished_at=_now(), step="", error=error)
        log(f"\n❌ {error}\n")
        return

    # --- برداشتنِ خروجی ------------------------------------------------------
    _set_state(step="ذخیره‌ی خروجی…")
    apk_src = os.path.join(APK_OUT, "app-release.apk")
    if not os.path.exists(apk_src):
        alt = [f for f in os.listdir(APK_OUT)] if os.path.isdir(APK_OUT) else []
        candidates = [f for f in alt if f.endswith(".apk")]
        if candidates:
            apk_src = os.path.join(APK_OUT, candidates[0])
        else:
            _set_state(status="error", finished_at=_now(), step="",
                       error="Gradle موفق بود ولی فایلِ APK پیدا نشد.")
            log("\n❌ فایلِ APK توی خروجی پیدا نشد.\n")
            return

    filename = f"zirnavis-{version_name}-{version_code}.apk"
    dest = os.path.join(RELEASES_DIR, filename)
    shutil.copy2(apk_src, dest)
    size = os.path.getsize(dest)

    apk_url = _upload_to_hf(dest, filename, log)
    if not apk_url and public_base:
        apk_url = f"{public_base.rstrip('/')}/apk/{filename}"

    _record_release({
        "file": filename,
        "version_name": version_name,
        "version_code": version_code,
        "note": note,
        "size": size,
        "built_at": _now(),
        "signed_release": bool(ks),
        "apk_url": apk_url or "",
        "duration": int(time.time() - started),
        "job_id": job_id,
    })

    if update_config:
        _update_config_json(version_code, apk_url, note, log)

    log(f"\n✅ آماده شد: {filename}  ({size / 1024 / 1024:.1f} مگابایت)"
        f"  در {int(time.time() - started)} ثانیه\n")
    _set_state(status="success", finished_at=_now(), step="", error=None,
               signed_release=bool(ks))


def cancel_build():
    global _cancel_requested, _current_proc
    if get_state().get("status") != "running":
        return False, "هیچ buildی در حال اجرا نیست."
    _cancel_requested = True
    proc = _current_proc
    if proc and proc.poll() is None:
        try:
            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
            time.sleep(3)
            if proc.poll() is None:
                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
        except (OSError, ProcessLookupError):
            pass
    return True, "درخواستِ لغو ارسال شد."