hubble.py

"""
Hubble 登录同步 — 同设备 mobile → hubble 数据搬运 (v1.3)
==========================================================

v1.3 变化:
  - 修数据膨胀: 只 cp files/keva/ (账号数据 ~50M), 不 cp 整个 files/* (~640M)
    每次执行前先 rm -rf 旧 keva 避免累积
  - 拆分三个入口: run_hubble (同步+启动), run_hubble_sync (仅同步), run_hubble_launch (仅启动)

对外签名:
    run_hubble(raw, scratch, log, progress, stop_event) -> (ok, png, info)
    run_hubble_sync(raw, scratch, log, progress, stop_event) -> (ok, png, info)
    run_hubble_launch(raw, scratch, log, progress, stop_event) -> (ok, png, info)
"""

import os
import sys
import time

_HERE = os.path.dirname(os.path.abspath(__file__))
_LIB = os.path.join(_HERE, "lib")
if os.path.isdir(_LIB) and _LIB not in sys.path:
    sys.path.insert(0, _LIB)

import paramiko  # noqa: F401
from conn import Session

SRC_PKG = "com.ss.android.ugc.aweme.mobile"
DST_PKG = "com.ss.android.ugc.aweme.hubble"

CERT_URL = (
    "https%3A%2F%2Fwebcast.amemv.com%2Ffalcon%2Fwebcast_douyin%2F"
    "page%2Fcertification%2Findex.html"
    "%3Fenter_from%3Dlive_take%26hide_nav_bar%3D1%26__live_platform__%3Dwebcast"
)

INTENT_TEMPLATE = (
    "intent:#Intent;launchFlags=0x10000000;"
    "component={dst}/com.ss.android.ugc.aweme.live.LiveDummyActivity;"
    "B.hide_status_bar=false;i.intent_type=1;"
    "B.enable_transparent_background=false;"
    "i.bundle_web_view_background_color=-1;"
    "S.url={url};"
    "S.key_calling_context=com.ss.android.ugc.aweme.shortvideo.ui.scan.ScanNewActivity;"
    "l.ad_id=0;B.hide_nav_bar=false;B.show_progress=false;"
    "B.hide_more=true;B.bundle_no_hw_acceleration=false;"
    "B.disable_open_animate=false;end"
)

POLL_INTERVAL = 5
POLL_TIMEOUT = 120


class HubbleError(RuntimeError):
    pass




# ═══════════════════ 内部构建块 ═══════════════════

def _init_session(raw_input_text, log, progress, check_stop):
    """SSH + adb + root + 双包校验 + 关 multiprocess. 返回 sess."""
    log("[1] 建立 SSH 隧道 + adb ready")
    progress(5)
    check_stop()

    try:
        sess = Session(raw_input_text, log=log, hard_timeout=480)
    except ValueError as e:
        raise HubbleError(str(e))

    try:
        sess.start()
    except paramiko.AuthenticationException:
        raise HubbleError("SSH 认证失败: 用户名或密码错误")
    except Exception as e:
        raise HubbleError(f"建立连接失败: {e}")
    log(f"      serial={sess.serial}")
    progress(10)

    who = sess.sh("id", check=False).strip()
    if "uid=0" not in who:
        raise HubbleError(f"需要 root, 当前: {who}")
    brand = sess.sh("getprop ro.product.brand", check=False).strip()
    model = sess.sh("getprop ro.product.model", check=False).strip()
    log(f"      设备: {brand} {model}  ({who})")
    progress(12)

    log("[2] 校验双包 + 关 WebView 多进程")
    check_stop()
    src_path = sess.sh(f"pm path {SRC_PKG}", check=False).strip()
    dst_path = sess.sh(f"pm path {DST_PKG}", check=False).strip()
    if "package:" not in src_path:
        raise HubbleError(f"源包未安装: {SRC_PKG}")
    if "package:" not in dst_path:
        raise HubbleError(f"目标包未安装: {DST_PKG}")
    log("      双包 OK")

    sess.sh("cmd webviewupdate disable-multiprocess", check=False)
    mp = sess.sh("dumpsys webviewupdate | grep 'Multiprocess enabled'", check=False).strip()
    log(f"      {mp}")
    if "false" not in mp:
        raise HubbleError("cmd webviewupdate disable-multiprocess 未生效")
    progress(18)
    return sess


def _do_sync(sess, log, progress, check_stop):
    """force-stop + 清旧 keva + cp keva + 权限 + 授权."""
    dst_dir = f"/data/data/{DST_PKG}"
    src_dir = f"/data/data/{SRC_PKG}"

    log("[3] force-stop")
    check_stop()
    sess.sh(f"am force-stop {DST_PKG}", check=False)
    time.sleep(1)
    progress(25)

    log("[4] overlay cp keva (仅账号数据, 不删 hubble 自有条目)")
    check_stop()
    sess.sh(f"mkdir -p {dst_dir}/files/keva", check=False)
    # overlay: 只覆盖 mobile 有的, 保留 hubble 独有的 (plugin 配置等)
    sess.sh(f"cp -rf {src_dir}/files/keva/* {dst_dir}/files/keva/ 2>/dev/null; echo cp_done")
    sz = sess.sh(f"du -s {dst_dir}/files/keva 2>/dev/null | head -1", check=False).strip()
    log(f"      keva overlay done ({sz})")
    progress(35)

    log("[5] chown + chcon")
    check_stop()
    dst_uidgid = sess.sh(f"stat -c '%u:%g' {dst_dir}").strip()
    dst_ctx = sess.sh(f"stat -c '%C' {dst_dir}").strip()
    log(f"      uid:gid={dst_uidgid} ctx={dst_ctx}")
    sess.sh(
        f"chown {dst_uidgid} {dst_dir}/files 2>/dev/null; "
        f"chcon {dst_ctx} {dst_dir}/files 2>/dev/null; "
        f"chown -R {dst_uidgid} {dst_dir}/files/keva; "
        f"chcon -R {dst_ctx} {dst_dir}/files/keva",
        check=False,
    )
    progress(40)

    log("[6] 授权")
    sess.sh(f"pm grant {DST_PKG} android.permission.POST_NOTIFICATIONS 2>/dev/null", check=False)
    progress(42)


def _do_launch(sess, log, progress, check_stop, pct_start=45, pct_end=97):
    """暖机 + 轮询 plugin + 启动 intent + 验证. 返回 ok."""
    log("[7] 暖机: force-stop → monkey launch")
    check_stop()
    sess.sh(f"am force-stop {DST_PKG}", check=False)
    time.sleep(1)
    sess.sh(f"monkey -p {DST_PKG} -c android.intent.category.LAUNCHER 1", check=False)
    time.sleep(5)

    pid = sess.sh(
        f"ps -ef | grep {DST_PKG} | grep -v grep | head -1 | awk '{{print $2}}'",
        check=False,
    ).strip()
    log(f"      PID={pid}")
    progress(pct_start)

    # 轮询: pct_start → pct_start+35
    poll_base = pct_start
    poll_range = 35
    log(f"[8] 等待 live plugin 就绪 (最多 {POLL_TIMEOUT}s)")
    check_stop()
    elapsed = 0
    plugin_ready = False

    while elapsed < POLL_TIMEOUT:
        time.sleep(POLL_INTERVAL)
        elapsed += POLL_INTERVAL
        check_stop()

        pct = poll_base + int(elapsed / POLL_TIMEOUT * poll_range)
        progress(min(pct, poll_base + poll_range - 1))

        # .jar
        jar = sess.sh(
            f"ls /data/data/{DST_PKG}/files/.patchs/sc_m.l.live.plugin*.jar 2>/dev/null"
            " | grep -v '\\.tp' | head -1",
            check=False, timeout=10,
        ).strip()
        if jar and "No such" not in jar:
            log(f"      [{elapsed}s] live plugin jar: {jar.split('/')[-1]}")
            plugin_ready = True
            break

        # .tp
        tp = sess.sh(
            f"ls /data/data/{DST_PKG}/files/.patchs/sc_m.l.live.plugin*.tp 2>/dev/null | head -1",
            check=False, timeout=10,
        ).strip()
        if tp and "No such" not in tp:
            if elapsed % 15 == 0:
                log(f"      [{elapsed}s] plugin 下载中...")
            continue

        # TTPluginManager
        if pid:
            tt = sess.sh(
                f"logcat -d --pid={pid} 2>/dev/null | grep -c cr_TTPluginManager",
                check=False, timeout=10,
            ).strip()
            try:
                if int(tt) > 0:
                    log(f"      [{elapsed}s] cr_TTPluginManager OK")
                    plugin_ready = True
                    break
            except ValueError:
                pass

        if elapsed % 30 == 0:
            log(f"      [{elapsed}s] 仍在等待...")

    if not plugin_ready:
        log(f"      [{elapsed}s] 超时, 仍尝试启动")
    progress(poll_base + poll_range)

    # 启动
    log("[9] 启动实名认证页 (不 force-stop)")
    check_stop()
    intent = INTENT_TEMPLATE.format(dst=DST_PKG, url=CERT_URL)
    out = sess.sh(f'am start "{intent}"', check=False).strip()
    log(f"      {out}")
    progress(poll_base + poll_range + 2)

    for i in range(15):
        check_stop()
        time.sleep(1)
        progress(min(poll_base + poll_range + 2 + i, pct_end))

    # 验证
    top = sess.sh(
        "dumpsys activity activities | grep topResumedActivity | head -1",
        check=False,
    ).strip()
    log(f"      {top}")
    if "LiveDummyActivity" in top:
        log("      LiveDummyActivity 在前台 OK")
    else:
        log("      [注意] 不在前台, 请看截图确认")
    return True


def _screenshot(sess, scratch_dir, log):
    ts_stamp = int(time.time())
    stamp = f"{os.getpid()}_{ts_stamp}"
    sess.sh(f"screencap -p /sdcard/hubble_verify_{stamp}.png", check=False)
    verify_png = None
    if scratch_dir:
        verify_png = os.path.join(scratch_dir, f"hubble_verify_{sess.local_port}_{stamp}.png")
        try:
            sess.pull(f"/sdcard/hubble_verify_{stamp}.png", verify_png)
        except Exception as e:
            log(f"      截图拉取失败: {e}")
            verify_png = None
    sess.sh(f"rm -f /sdcard/hubble_verify_{stamp}.png 2>/dev/null; true", check=False)
    return verify_png


# ═══════════════════ 三个公开入口 ═══════════════════

def run_hubble(raw_input_text, scratch_dir, log, progress, stop_event=None):
    """完整流程: 同步 + 暖机 + 启动."""
    def check_stop():
        if stop_event and stop_event.is_set():
            raise HubbleError("用户已中断")

    if scratch_dir:
        try: os.makedirs(scratch_dir, exist_ok=True)
        except: pass

    sess = _init_session(raw_input_text, log, progress, check_stop)
    try:
        _do_sync(sess, log, progress, check_stop)
        _do_launch(sess, log, progress, check_stop, pct_start=45, pct_end=97)
        png = _screenshot(sess, scratch_dir, log)
        log("\n=== Hubble 同步+启动完成 ===")
        progress(100)
        return True, png, None
    finally:
        try: sess.close()
        except: pass


def run_hubble_sync(raw_input_text, scratch_dir, log, progress, stop_event=None):
    """仅同步数据, 不暖机不启动."""
    def check_stop():
        if stop_event and stop_event.is_set():
            raise HubbleError("用户已中断")

    if scratch_dir:
        try: os.makedirs(scratch_dir, exist_ok=True)
        except: pass

    sess = _init_session(raw_input_text, log, progress, check_stop)
    try:
        _do_sync(sess, log, progress, check_stop)
        log("\n=== Hubble 仅同步完成 ===")
        log("数据已搬运, 如需验证请单独执行「启动页面」.")
        progress(100)
        return True, None, None
    finally:
        try: sess.close()
        except: pass


def _quick_launch(sess, log, progress, check_stop):
    """快速尝试: 直接 am start intent, 不暖机. 返回 True=成功 False=白屏需兜底."""
    log("[快速] 尝试直接启动 (不暖机)...")
    progress(22)
    check_stop()

    # 先 force-stop 再启动
    sess.sh(f"am force-stop {DST_PKG}", check=False)
    time.sleep(2)

    intent = INTENT_TEMPLATE.format(dst=DST_PKG, url=CERT_URL)
    sess.sh(f'am start "{intent}"', check=False)
    log("      intent 已发送, 等待 15s...")
    progress(30)

    for i in range(15):
        check_stop()
        time.sleep(1)
        progress(30 + i)

    # 验证: top activity + WebView 日志
    top = sess.sh(
        "dumpsys activity activities | grep topResumedActivity | head -1",
        check=False,
    ).strip()

    if "LiveDummyActivity" not in top:
        log("      [快速] Activity 不在前台, 需兜底")
        return False

    # 检查 WebView 实例是否创建 (cr_AwContents / cr_TTPlugin)
    pid = sess.sh(
        f"ps -ef | grep {DST_PKG} | grep -v grep | head -1 | awk '{{print $2}}'",
        check=False,
    ).strip()
    if pid:
        wv = sess.sh(
            f"logcat -d --pid={pid} 2>/dev/null | grep -iE 'cr_AwContents|cr_TTPlugin' | tail -1",
            check=False, timeout=10,
        ).strip()
        if wv:
            log("      [快速] WebView 已创建, 成功!")
            return True

    # Activity 在前台但 WebView 未检测到 — 可能白屏
    log("      [快速] WebView 未检测到, 可能白屏, 进入兜底...")
    return False


def run_hubble_launch(raw_input_text, scratch_dir, log, progress, stop_event=None):
    """先快速直接启动, 失败则暖机+轮询兜底."""
    def check_stop():
        if stop_event and stop_event.is_set():
            raise HubbleError("用户已中断")

    if scratch_dir:
        try: os.makedirs(scratch_dir, exist_ok=True)
        except: pass

    sess = _init_session(raw_input_text, log, progress, check_stop)
    try:
        # 第一轮: 快速尝试
        if _quick_launch(sess, log, progress, check_stop):
            progress(95)
            png = _screenshot(sess, scratch_dir, log)
            log("\n=== 快速启动成功 ===")
            progress(100)
            return True, png, None

        # 第二轮: 完整暖机兜底
        log("[兜底] 进入完整暖机+轮询流程...")
        _do_launch(sess, log, progress, check_stop, pct_start=50, pct_end=97)
        png = _screenshot(sess, scratch_dir, log)
        log("\n=== Hubble 启动页面完成 (兜底) ===")
        progress(100)
        return True, png, None
    finally:
        try: sess.close()
        except: pass


def run_hubble_quick(raw_input_text, scratch_dir, log, progress, stop_event=None):
    """快速模式: 只发 intent, 不暖机不轮询不验证."""
    def check_stop():
        if stop_event and stop_event.is_set():
            raise HubbleError("用户已中断")

    if scratch_dir:
        try: os.makedirs(scratch_dir, exist_ok=True)
        except: pass

    sess = _init_session(raw_input_text, log, progress, check_stop)
    try:
        log("[快速] force-stop + am start intent")
        progress(30)
        sess.sh(f"am force-stop {DST_PKG}", check=False)
        time.sleep(1)
        intent = INTENT_TEMPLATE.format(dst=DST_PKG, url=CERT_URL)
        sess.sh(f'am start "{intent}"', check=False)
        log("      intent 已发送")
        progress(60)

        log("[快速] 等待 10s...")
        for i in range(10):
            check_stop()
            time.sleep(1)
            progress(60 + i * 3)

        png = _screenshot(sess, scratch_dir, log)
        log("\n=== 快速启动完成 ===")
        progress(100)
        return True, png, None
    finally:
        try: sess.close()
        except: pass