147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
"""自动准备官方 scrcpy-server（与 vendor/scrcpy-server.version 版本一致），供 Web 真流 adb push 使用。"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
# 官方发行资产名：scrcpy-server-v{version}（无扩展名，实为 JAR）
|
||
_GITHUB_DL = "https://github.com/Genymobile/scrcpy/releases/download/v{tag}/scrcpy-server-v{tag}"
|
||
_MIN_BYTES = 8_000
|
||
_PER_URL_ATTEMPTS = 3
|
||
_PER_URL_DELAY_S = 2.0
|
||
|
||
|
||
def _read_version(repo_root: Path) -> str:
|
||
env = (os.environ.get("SCRCPY_SERVER_VERSION") or "").strip().lstrip("v")
|
||
if env:
|
||
return env
|
||
vf = repo_root / "vendor" / "scrcpy-server.version"
|
||
if vf.is_file():
|
||
line = vf.read_text(encoding="utf-8").splitlines()[0].strip().lstrip("v")
|
||
if line:
|
||
return line
|
||
return "3.3.3"
|
||
|
||
|
||
def _expand_download_url(template: str, tag: str) -> str:
|
||
return template.replace("{tag}", tag).replace("{version}", tag)
|
||
|
||
|
||
def _download_urls_to_try(tag: str) -> list[str]:
|
||
"""顺序：SCRCPY_SERVER_DOWNLOAD_URL → SCRCPY_SERVER_DOWNLOAD_URLS →（默认可选）GitHub。"""
|
||
urls: list[str] = []
|
||
single = (os.environ.get("SCRCPY_SERVER_DOWNLOAD_URL") or "").strip()
|
||
if single:
|
||
urls.append(_expand_download_url(single, tag))
|
||
raw = (os.environ.get("SCRCPY_SERVER_DOWNLOAD_URLS") or "").strip()
|
||
if raw:
|
||
for part in raw.split(","):
|
||
u = part.strip()
|
||
if not u:
|
||
continue
|
||
uu = _expand_download_url(u, tag)
|
||
if uu not in urls:
|
||
urls.append(uu)
|
||
no_github = os.environ.get("SCRCPY_SERVER_DOWNLOAD_NO_GITHUB", "").strip().lower() in (
|
||
"1",
|
||
"true",
|
||
"yes",
|
||
"on",
|
||
)
|
||
gh = _GITHUB_DL.format(tag=tag)
|
||
if not no_github and gh not in urls:
|
||
urls.append(gh)
|
||
seen: set[str] = set()
|
||
out: list[str] = []
|
||
for u in urls:
|
||
if u not in seen:
|
||
seen.add(u)
|
||
out.append(u)
|
||
return out
|
||
|
||
|
||
def _fetch_url(url: str) -> tuple[bool, bytes | None, str]:
|
||
req = urllib.request.Request(
|
||
url,
|
||
headers={"User-Agent": "douyinyoutube-live-hub/scrcpy-bootstrap"},
|
||
method="GET",
|
||
)
|
||
last_err = ""
|
||
for attempt in range(1, _PER_URL_ATTEMPTS + 1):
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=120) as resp:
|
||
data = resp.read()
|
||
return True, data, ""
|
||
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, OSError) as e:
|
||
last_err = str(e)
|
||
if attempt < _PER_URL_ATTEMPTS:
|
||
time.sleep(_PER_URL_DELAY_S)
|
||
return False, None, last_err
|
||
|
||
|
||
def ensure_scrcpy_jar(repo_root: Path) -> tuple[bool, str]:
|
||
"""
|
||
若 vendor/scrcpy-server.jar 不存在或过小，则按 URL 列表下载。
|
||
环境变量：
|
||
SKIP_SCRCPY_SERVER_FETCH=1 — 跳过
|
||
SCRCPY_SERVER_DOWNLOAD_URL — 优先下载地址，可用 {tag}/{version}
|
||
SCRCPY_SERVER_DOWNLOAD_URLS — 逗号分隔多条，先于 GitHub
|
||
SCRCPY_SERVER_DOWNLOAD_NO_GITHUB=1 — 仅用上面自定义 URL，不回退 GitHub
|
||
"""
|
||
if os.environ.get("SKIP_SCRCPY_SERVER_FETCH", "").strip().lower() in ("1", "true", "yes", "on"):
|
||
return True, "skip fetch (SKIP_SCRCPY_SERVER_FETCH)"
|
||
|
||
vendor_dir = repo_root / "vendor"
|
||
dest = vendor_dir / "scrcpy-server.jar"
|
||
env_jar = (os.environ.get("SCRCPY_SERVER_JAR") or "").strip()
|
||
if env_jar:
|
||
p = Path(env_jar).expanduser()
|
||
if p.is_file() and p.stat().st_size >= _MIN_BYTES:
|
||
return True, "using SCRCPY_SERVER_JAR"
|
||
|
||
if dest.is_file() and dest.stat().st_size >= _MIN_BYTES:
|
||
return True, "vendor/scrcpy-server.jar ok"
|
||
|
||
tag = _read_version(repo_root)
|
||
urls = _download_urls_to_try(tag)
|
||
if not urls:
|
||
return False, "未配置任何下载源（SCRCPY_SERVER_DOWNLOAD_NO_GITHUB=1 且未设置 SCRCPY_SERVER_DOWNLOAD_URL(S)）"
|
||
vendor_dir.mkdir(parents=True, exist_ok=True)
|
||
tmp = dest.with_suffix(".jar.part")
|
||
|
||
last_fail = ""
|
||
for url in urls:
|
||
ok, data, err = _fetch_url(url)
|
||
if not ok or data is None:
|
||
last_fail = f"{url}: {err}"
|
||
continue
|
||
if len(data) < _MIN_BYTES:
|
||
last_fail = f"{url}: too small ({len(data)} bytes)"
|
||
continue
|
||
if not data.startswith(b"PK"):
|
||
last_fail = f"{url}: not a JAR (missing PK header)"
|
||
continue
|
||
try:
|
||
tmp.write_bytes(data)
|
||
os.replace(tmp, dest)
|
||
except OSError as e:
|
||
return False, f"write vendor/scrcpy-server.jar failed: {e}"
|
||
return True, f"fetched scrcpy-server v{tag} -> vendor/scrcpy-server.jar (via {url})"
|
||
|
||
return False, f"download failed after {len(urls)} source(s): {last_fail}"
|
||
|
||
|
||
def main() -> int:
|
||
root = Path(__file__).resolve().parent.parent
|
||
ok, msg = ensure_scrcpy_jar(root)
|
||
print(msg, file=sys.stderr if not ok else sys.stdout)
|
||
return 0 if ok else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|