auto_control/tasks/generic/publish_flow.py

"""发布计划的执行原语：**把视频推到手机**、抓分享链接、清理手机上的素材。

**分工**（用户定的）：平台只负责"素材到手机 + 状态记账"，
**抖音里怎么发（打开→点+→相册→下一步→填文案→发布）由用户在任务画布里自己写步骤** ——
所以这里**没有**写死的抖音发布流程，只有三个原语 + 一段编排：

```
push_one()            取计划 → 原子占位 → adb push 到指定目录 → 触发相册刷新
                      → 把标题写进剪贴板（供后面的「粘贴」步骤直接用）
capture_share_link()  进「我 → 作品」第一条 → 分享 → 复制链接 → 读剪贴板 → 抠出 v.douyin.com 链接
clean_phone_video()   删手机上的素材（**精确路径**，绝不 rm 通配）
```

任务步骤（`task.py`）：
  · 「推送发布视频」`push_release` —— 调 `push_one()`
  · 「标记发布结果」`mark_release` —— 用户在自己的发布流程末尾调用，回写状态（可顺带抓链接）

⚠ **失败阶段决定落 `failed` 还是 `unknown`**（见 `core/video_plan.STAGE_STATUS`）：
推送阶段失败=还没碰抖音 → `failed`（可安全重试）；推送完成后（stage=`post`）出的岔子
→ `unknown`（**可能已经发出去了，绝不自动重试**）。
"""
import os
import re
import time

from core.logger import get_logger

_log = get_logger("task.publish")

# 相册目录：**DCIM/Camera**（用户定的）。
# ⚠ 实测踩过的两件事：
#   ① 相册（含抖音的选视频页）读的是 **MediaStore 索引**，`adb push` 只放文件、不进索引
#      → 推完必须触发一次扫描（见 scan_media），否则"推送成功但相册里没有"。
#   ② 推到自建子目录 `/sdcard/DCIM/rp` 时，MIUI 会把它归进「其他相册」，
#      抖音的选视频页看不顺眼 —— 干脆放官方相机目录，哪儿都列得出来。
# 放 Camera 的代价：用户自己拍的视频也在这个目录里 → **必须靠"刚推的排最前"来定位它**
#   → 所以推完要 `touch` 把手机上那个文件的 mtime 改成"现在"（adb push 会保留**本地**修改时间，
#     推一个 3 天前上传的素材，在按时间排序的相册里就排不到最前面 —— 实测踩过）。
ALBUM_DIR = "/sdcard/DCIM/Camera"
SCAN_WAIT = 2.0          # 媒体扫描后等相册刷新（真机实测再调）
# 平台**曾经用过**的推送目录：推送前把同一计划在别处的旧副本清掉（历史遗留 + 换目录后的残留）。
# 不清的后果实测过：相册里同一个 plan id 出现 2~3 个同名视频，"点第一个"点到的可能是指向
# 已删文件的老记录 —— 这正是"相册里选错视频"的来源。
KNOWN_ALBUM_DIRS = ("/sdcard/DCIM/Camera", "/sdcard/DCIM/rp", "/sdcard/Movies/rp")


def _norm_path(p):
    """路径归一化，用于比对：去掉引号、`/storage/emulated/0` 与 `/sdcard` 等价、统一小写。

    ⚠ 大小写不是洁癖：MediaStore 的 `_data` 返回的是 **`/storage/emulated/0/dcim/rp/...`**
    （目录部分被小写了），而我们写的是 `/sdcard/DCIM/...` —— 直接字符串比对永远不相等。
    """
    s = (p or "").strip().strip("'\"")
    s = s.replace("/storage/emulated/0", "/sdcard").replace("/storage/self/primary", "/sdcard")
    return s.lower()


def _log_of(log):
    return log or (lambda s: _log.info(s))


def _sh(d, cmd):
    """跑一条 shell 并**拿回文本**。

    uiautomator2 3.x 的 `d.shell()` 返回 `ShellResponse`（**tuple 子类**，字段 .output/.exit_code），
    不是 str —— 直接 `'abc' in resp`（元组成员判断，恒 False）或 `resp.strip()`（没有这方法）
    都会得到错误结果。实测踩过：推送后的"文件大小校验"永远不通过、
    "有没有进相册"永远报没进 —— 全是这一步读错了。统一走这里取文本。
    """
    r = d.shell(cmd)
    out = getattr(r, "output", None)          # u2 3.x
    if out is None:
        out = r if isinstance(r, str) else str(r)   # 老版本 u2 / 异常返回
    return (out or "").strip()


# ================== 原语 ==================
def push_video(d, serial, local_path, plan_id, album_dir=ALBUM_DIR, ext=".mp4"):
    """把平台素材推到手机相册目录，并把手机上的时间改成"现在"。返回 (remote_path, error)。

    改时间是为了**排序**：相册/抖音的选视频页按时间倒序，
    `adb push` 保留的是**本地文件的修改时间**（推一个 3 天前上传的素材 → 排在很后面），
    那"点第一个缩略图 = 刚推的那个"就不成立、可能点到别的视频。所以推完 `touch` 一下。
    """
    remote = f"{album_dir}/{plan_id}{ext}"
    try:
        local_size = os.path.getsize(local_path)
        _sh(d, f"mkdir -p {album_dir}")
        if hasattr(d, "push"):
            d.push(local_path, remote)
        else:                                   # 老版本 u2 兜底
            from core.adb_helper import _adb
            out = _adb("-s", serial, "push", local_path, remote)
            if "error" in (out or "").lower():
                return "", f"adb push 失败: {out[:120]}"
        # 校验：**比对大小列**（不是"输出里有没有这个数字"—— 那种子串匹配会把日期里的
        # 数字也当命中；u2 的 ShellResponse 更会让它恒不通过，见 _sh 的注释）
        info = _sh(d, f"ls -l '{remote}'")
        got = _remote_size(info)
        if got is None:
            return "", f"推上去的文件找不到: {info[:120] or '（ls 无输出）'}"
        if got != local_size:
            return "", f"推上去的文件大小不对: 本地 {local_size} / 手机 {got}"
        # 排到最前：把手机上这个文件的 mtime 改成现在（改不动也不拦，只是排序可能不理想）
        try:
            _sh(d, f"touch '{remote}'")
        except Exception as e:
            _log.warning(f"touch 手机上的素材失败（排序可能不理想）: {e}")
        drop_old_copies(d, plan_id, keep=remote, ext=ext)   # 老目录里的同名副本顺手清掉
        return remote, ""
    except Exception as e:
        return "", f"推文件异常: {type(e).__name__}: {str(e)[:120]}"


def _remote_size(ls_out):
    """从 `ls -l` 输出里取**大小列**（字节）。取不到返回 None。

    MIUI 的 toybox 输出形如 `-rw-rw---- 1 u0_a163 media_rw 15405006 2026-09-25 08:42 /sdcard/...`。
    判据两道：① 行首是文件类型字符（`-`/`d`/`l`…）—— 挡住 "No such file" 这类错误行；
    ② 第 5 列是纯数字 —— 大小列就在那（`ls -l` 一定有 size 列，第 4 列是 owner）。
    比"输出里有没有这个数字"这种子串匹配可靠得多（后者会被日期/权限里的数字骗到）。
    """
    for line in (ls_out or "").splitlines():
        parts = line.split()
        if len(parts) >= 6 and parts[0][:1] in "-dlcbps" and parts[4].isdigit():
            return int(parts[4])
    return None


def _indexed_paths(d, name):
    """这个文件名在相册索引里对应哪些路径（MediaStore 的 `_data` 列表）。"""
    query = ("content query --uri content://media/external/video/media --user 0 "
             f"--projection _id:_display_name:_data --where \"_display_name='{name}'\"")
    out = _sh(d, query)
    return [m.group(1) for m in re.finditer(r"_data=(\S+)", out or "")]


def scan_media(d, remote):
    """触发相册刷新，并**按路径**验证它真的进了相册索引。

    返回 `(verify, detail)`：`verify` ∈ `ok`（**这个路径**在索引里）/ `no_index`（扫完还是查不到）。

    ⚠ 为什么要验证：相册读的是 MediaStore 索引，`adb push` 只是放文件 ——
    推成功但相册看不到是个**静默失败**（"点了第一个缩略图结果点到别的东西"就是这么来的）。
    ⚠ **必须比路径而不是比文件名**：同一个 plan id 可能在老目录里还留着同名副本
    （实测 A01 上 `Movies/rp` + `DCIM/rp` 各有一份），只比名字会把"老副本在索引里"
    误判成"新文件已就绪" —— 相册里点第一个，点到的是那份指向旧文件的记录。
    ⚠ 命令只用**实测有效**的那两条：Android 12 上 `am broadcast MEDIA_SCANNER_SCAN_FILE`
    已基本无效（返回 result=0 但什么也不做），真正干活的是 `content call … scan_file`；
    老写法里那条 `MEDIA_MOUNTED` 广播在新系统上无效、还可能触发全盘重扫，已去掉。
    """
    for cmd in (f"content call --uri content://media/external/file --method scan_file --arg '{remote}'",
                f"am broadcast -a android.intent.action.MEDIA_SCANNER_SCAN_FILE -d file://{remote}"):
        try:
            _sh(d, cmd)
        except Exception as e:
            _log.warning(f"媒体扫描命令失败（继续）: {e}")
    time.sleep(SCAN_WAIT)
    name = remote.split("/")[-1]
    want = _norm_path(remote)
    others = []
    for _ in range(3):          # 索引有时慢半拍，给它一秒再查两次
        try:
            paths = _indexed_paths(d, name)
            if any(_norm_path(p) == want for p in paths):
                return "ok", f"已进相册索引（{remote}）"
            others = [p for p in paths if _norm_path(p) != want]
        except Exception as e:
            _log.warning(f"查相册索引失败（当作没进）: {e}")
            break
        time.sleep(1)
    if others:
        _log.warning(f"⚠ 相册索引里只有**同名但不同路径**的记录：{others[:2]} ——"
                     f"这个文件本身（{remote}）还没进索引")
    _log.warning(f"⚠ 文件已推到 {remote}，但**没进相册索引**（相册里可能看不到它）——"
                 f"推送目录 {ALBUM_DIR} 或系统抓取策略可能要调整，见 doc/TASK_DEV.md §4.7")
    return "no_index", "扫完仍没进相册索引（相册里可能看不到它）"


def drop_old_copies(d, plan_id, keep, ext=".mp4"):
    """把**同一个计划**在平台用过别的目录里的旧副本删掉（精确文件名，绝不通配）。

    为什么要删：换了推送目录之后，老目录里的同名副本还在相册里（实测一个 plan id 躺了 3 份），
    "点第一个缩略图 = 刚推的那个"就不成立了。删完顺手扫一下，让相册索引把那行去掉。
    """
    gone = []
    for adir in KNOWN_ALBUM_DIRS:
        old = f"{adir}/{plan_id}{ext}"
        if _norm_path(old) == _norm_path(keep):
            continue
        try:
            if _remote_size(_sh(d, f"ls -l '{old}'")) is None:
                continue
            _sh(d, f"rm -f '{old}'")
            _sh(d, f"content call --uri content://media/external/file --method scan_file --arg '{old}'")
            gone.append(old)
        except Exception as e:
            _log.warning(f"清旧副本失败（不影响本次推送）: {old}: {e}")
    if gone:
        _log.info(f"已清掉同一计划在旧目录里的副本: {gone}")
    return gone


def verify_pushed(d, remote):
    """**推送后的一致性检查**：文件在不在 + 进没进相册索引。返回 (verify, detail)。

    `verify` ∈ `ok` / `no_index` / `nofile`。**这是"推送成功"的判据**：
    只看文件在不在是不够的 —— 后面的步骤要在相册里点它。
    """
    size = _remote_size(_sh(d, f"ls -l '{remote}'"))
    if size is None:
        return "nofile", "手机上找不到这个文件"
    return scan_media(d, remote)


def clean_phone_video(d, remote):
    """删手机上的素材（**精确路径**，绝不通配 —— 通配会删掉用户自己拍的东西）。"""
    if not remote:
        return False
    try:
        _sh(d, f"rm -f '{remote}'")
        if _remote_size(_sh(d, f"ls -l '{remote}'")) is not None:
            _log.warning(f"手机上的素材没删掉: {remote}")
            return False
        return True
    except Exception as e:
        _log.warning(f"删手机素材异常（不影响发布结论）: {e}")
        return False


def _first_album_item(d):
    """点相册/作品网格里的**第一个**缩略图（抖音按时间倒序，最新的在最前）。

    不用固定坐标：dump 出网格里的图片单元，按 (y, x) 排序取第一个 ——
    分辨率/刘海不同的机器都能用。仅用于**抓分享链接**（进「我」的作品第一条）。
    """
    try:
        xml = d.dump_hierarchy()
    except Exception as e:
        return False, f"dump 失败: {e}"
    cells = []
    for m in re.finditer(r"<node[^>]*>", xml):
        tag = m.group(0)

        def g(k):
            mm = re.search(k + r'="([^"]*)"', tag)
            return mm.group(1) if mm else ""
        bounds = g("bounds")
        mb = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds)
        if not mb:
            continue
        x0, y0, x1, y1 = (int(v) for v in mb.groups())
        if y0 < 200 or (y1 - y0) < 120 or (x1 - x0) < 120:
            continue
        cells.append((y0, x0, (x0 + x1) // 2, (y0 + y1) // 2))
    if not cells:
        return False, "没找到缩略图"
    cells.sort()
    d.click(cells[0][2], cells[0][3])
    return True, f"点了第一个缩略图 ({cells[0][2]},{cells[0][3]})"


def _click_any(d, sels, timeout=3):
    """按顺序试几个选择器（形如 "text=分享" / "descriptionContains=分享"），点中即返回。"""
    for sel in sels:
        try:
            key, val = sel.split("=", 1)
            el = d(**{key: val})
            if el.exists(timeout=timeout):
                el.click()
                return True, sel
        except Exception:
            continue
    return False, ""


def capture_share_link(d, serial, timeout=30):
    """抓作品的分享链接：我 → 作品第一条 → 分享 → 复制链接 → 读剪贴板。

    返回 (url, msg)。**抓不到不算失败**：链接空着，界面上标"缺链接"、素材文件不删。
    ⚠ 这一段仍是抖音 UI 流程，**按文字定位、需真机确认**；抖音改版就走不通。
    """
    from core.video_plan import extract_share_url
    from core.clipboard_helper import inject_clipboard
    try:
        _click_any(d, ["descriptionContains=我", "text=我"], timeout=4)
        time.sleep(2)
        ok, msg = _first_album_item(d)
        if not ok:
            return "", f"进作品页失败: {msg}"
        time.sleep(2.5)
        ok, sel = _click_any(d, ["descriptionContains=分享", "text=分享"], timeout=4)
        if not ok:
            return "", "没找到「分享」按钮"
        time.sleep(1.5)
        ok, sel = _click_any(d, ["text=复制链接", "text=复制"], timeout=4)
        if not ok:
            return "", "分享面板里没找到「复制链接」"
        time.sleep(1.2)
        try:
            got = d.clipboard
        except Exception as e:
            return "", f"读剪贴板失败: {e}"
        url = extract_share_url(got or "")
        return (url, "ok") if url else ("", f"剪贴板里没有抖音链接: {str(got)[:60]!r}")
    except Exception as e:
        return "", f"抓链接异常: {type(e).__name__}: {str(e)[:100]}"


# ================== 编排 ==================
def push_one(serial, plan, opts=None, log=None):
    """把一条计划的素材推到手机（**不发布**）。返回 (ok, msg)。

    做四件事：原子占位（防多设备/重跑重复推）→ 推文件 → 触发相册刷新 →
    （可选）把标题写进手机剪贴板，供你后面画的「粘贴」步骤直接用。

    占位成功后把 `stage` 置成 `post` —— 意思是"**后面的事不归平台管了**"：
    再出岔子（比如你标记失败）按 `unknown` 处理，而不是"可安全重试"，
    避免任务把一条你其实已经发出去的计划又推一遍、又发一遍。
    """
    opts = opts or {}
    log = _log_of(log)
    import uiautomator2 as u2
    from core import video_plan as vp

    plan_id = plan["id"]
    if not vp.claim(plan_id, status=vp.ST_PUSHING, stage=vp.STAGE_MANUAL):
        return False, "这条已被别的任务/设备领走了（跳过）"
    d = None
    try:
        local = vp._video_path(plan.get("video_file"))
        if not local or not os.path.exists(local):
            vp.mark_failed(plan_id, "push", "平台上的素材文件不在了（可能已被清理，要发请重传视频）")
            return False, "平台素材缺失"
        d = u2.connect(serial)
        remote, err = push_video(d, serial, local, plan_id,
                                 album_dir=opts.get("album_dir") or ALBUM_DIR)
        if err:
            vp.mark_failed(plan_id, "push", err)
            return False, err
        log(f"素材已推到手机: {remote}")
        verify, vmsg = verify_pushed(d, remote)
        if verify == "nofile":
            # 文件都不在 → 真失败（可重试），别往后走
            vp.mark_failed(plan_id, "push", vmsg)
            return False, f"推送校验失败：{vmsg}"
        log(f"推送校验：{'✅ ' + vmsg if verify == 'ok' else '⚠ ' + vmsg}")
        title = (plan.get("title") or "").strip()
        if opts.get("to_clipboard", True) and title:
            try:
                from core.clipboard_helper import inject_clipboard
                okc, msgc = inject_clipboard(serial, title, d=d)
                log(f"标题已写进剪贴板（{'成功' if okc else '失败: ' + msgc}）"
                    f"—— 后面用「粘贴」步骤就能填进抖音")
            except Exception as e:
                log(f"标题写剪贴板失败（不影响推送）: {e}")
        vp.mark_pushed(plan_id, remote=remote, verify=verify)
        return True, (f"已推送到手机：{remote}" if verify == "ok"
                      else f"已推送到手机：{remote}（⚠ {vmsg}）")
    except Exception as e:
        vp.mark_failed(plan_id, "push", f"异常: {type(e).__name__}: {str(e)[:150]}")
        return False, f"推送异常: {type(e).__name__}"
    finally:
        try:
            row = vp.get_plan(plan_id)
            if row and row["status"] == vp.ST_PUSHING and not row.get("stage") == vp.STAGE_MANUAL:
                vp.mark_failed(plan_id, "push", "执行中断（兜底降级）")
        except Exception:
            pass


def run_push(serial, plan_id, opts=None):
    """给计划页「推送到手机」按钮用的入口（在后台线程里跑）。"""
    from core import video_plan as vp
    plan = vp.get_plan(plan_id)
    if not plan:
        _log.warning(f"推送：计划 {plan_id} 不存在")
        return False, "计划不存在"
    return push_one(serial, plan, opts)