""" ============================================================================= 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, "درخواستِ لغو ارسال شد."