auto_control/web/video_plan_api.py

"""视频发布计划 API（「账号 → 发布计划」页）。

数据在 `video_plan` 表、服务层 `core/video_plan.py`；任务侧怎么自动发布见
`tasks/generic/task.py` 的 `_exec_publish_video`（本文件的「立即发布」调的是同一段逻辑）。

红线（与 `core/video_plan.py` 顶部一致）：发布前必须原子占位；失败要分
"推文件阶段（可重试）"与"抖音阶段（结果未知、绝不自动重试）"两类。
"""
import csv
import io

from flask import Blueprint, jsonify, request, send_file

from core import video_plan as vp
from core.logger import get_logger
from web.auth import perm_required, PERM_DEVICES, PERM_TASKS

_log = get_logger("web")
bp = Blueprint("video_plan", __name__)


@bp.route("/api/video_plan/upload_video", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_vp_upload_video():
    """上传单个视频（前端**逐文件**发，才能给每个文件一条结果与进度）。

    文件名规则 `手机号_日期_编号`（编号可省略）；解析出的手机号必须**恰好**命中台账一个账号。
    """
    f = request.files.get("file")
    if not f or not f.filename:
        return jsonify({"ok": False, "error": "没收到文件"}), 400
    replace = str(request.form.get("replace") or "") in ("1", "true", "yes")
    order = request.form.get("order")
    result = vp.upload_video(f, order=order, replace=replace)
    if not result.get("ok"):
        return jsonify(result), 400
    return jsonify(result)


@bp.route("/api/video_plan/upload_titles", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_vp_upload_titles():
    """上传标题：`{text, dry_run}`；也接受 multipart 的 txt 文件（字段名 file）。

    每行 `标题内容_手机号_日期_编号` 或 `手机号_日期_编号_标题`（都认；编号可省略、标题里可以有下划线）；
    `#` 开头的行**只有认不出手机号/日期时才当注释**（`#话题` 开头的标题照收）。
    """
    if request.files.get("file"):
        raw = request.files["file"].stream.read()
        try:
            text = raw.decode("utf-8")
        except UnicodeDecodeError:
            text = raw.decode("gbk", errors="replace")
        dry = str(request.form.get("dry_run") or "") in ("1", "true", "yes")
    else:
        data = request.json or {}
        text = data.get("text") or ""
        dry = bool(data.get("dry_run"))
    if not text.strip():
        return jsonify({"ok": False, "error": "内容是空的"}), 400
    result = vp.import_titles(text, dry_run=dry)
    return jsonify(result)


@bp.route("/api/video_plan/timeline")
@perm_required(PERM_DEVICES)
def api_vp_timeline():
    """时间线：?range=today|week|all &from=&to=&status=&account_id=&q=&limit="""
    rng = (request.args.get("range") or "").strip()
    today = vp._today()
    date_from = (request.args.get("from") or "").strip()
    date_to = (request.args.get("to") or "").strip()
    if rng == "today" and not date_from:
        date_from = date_to = today
    elif rng == "week" and not date_from:
        import time as _t
        date_from = today
        date_to = _t.strftime("%Y-%m-%d", _t.localtime(_t.time() + 7 * 86400))
    data = vp.timeline(date_from, date_to,
                       status=(request.args.get("status") or "").strip(),
                       account_id=(request.args.get("account_id") or "").strip(),
                       q=(request.args.get("q") or "").strip(),
                       limit=int(request.args.get("limit") or 500))
    return jsonify({"ok": True, **data})


@bp.route("/api/video_plan/stats")
@perm_required(PERM_DEVICES)
def api_vp_stats():
    """顶部统计（今天要发/已发/待标题/过期未发 + 素材占用与磁盘余量）。"""
    return jsonify({"ok": True, "stats": vp.stats(), "disk": vp.disk_info(),
                    "today": vp._today()})


@bp.route("/api/video_plan/<pid>", methods=["GET", "PUT", "DELETE"])
@perm_required(PERM_DEVICES)
def api_vp_one(pid):
    if request.method == "GET":
        row = vp.get_plan(pid)
        if not row:
            return jsonify({"ok": False, "error": "计划不存在"}), 404
        return jsonify({"ok": True, "plan": row})
    if request.method == "DELETE":
        if not vp.delete_plan(pid):
            return jsonify({"ok": False, "error": "计划不存在"}), 404
        return jsonify({"ok": True, "msg": "已删除（素材文件一并删除）"})
    data = request.json or {}
    if "title" in data or "body" in data:
        row, err = vp.set_title(pid, data.get("title") or "", body=data.get("body"))
        if err:
            return jsonify({"ok": False, "error": err}), 400 if "太长" in err else 404
        return jsonify({"ok": True, "msg": "标题/正文已改", "plan": row})
    return jsonify({"ok": False, "error": "没有要改的字段"}), 400


@bp.route("/api/video_plan/<pid>/video")
@perm_required(PERM_DEVICES)
def api_vp_video(pid):
    """预览平台上的素材（文件被清理后返回 410）。"""
    import os
    from flask import send_file as _send
    row = vp.get_plan(pid)
    if not row:
        return jsonify({"ok": False, "error": "计划不存在"}), 404
    path = vp._video_path(row.get("video_file"))
    if not path or not os.path.exists(path):
        return jsonify({"ok": False, "error": "素材文件已清理"}), 410
    return _send(path, conditional=True)


@bp.route("/api/video_plan/<pid>/push", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_vp_push(pid):
    """把这一条的素材**推到手机**（只推送，不在手机上发布）。

    发布动作由你写在任务里（推送步骤后面接你自己的步骤），这里只是"人工先推一条"。
    """
    import threading
    row = vp.get_plan(pid)
    if not row:
        return jsonify({"ok": False, "error": "计划不存在"}), 404
    # 地址现查（设备名 → 当前地址）：**不能用计划行里的快照** —— 设备换 IP 后快照就是错的，
    # 而且台账是粘贴导入的老记录可能压根没存上快照（实测踩过：设备池里明明有 A01，
    # 却因为快照为空点不动按钮）。快照只作为兜底。
    from core import ledger
    serial = ledger.serial_of(row.get("device_name") or "") or (row.get("serial") or "")
    if not serial:
        return jsonify({"ok": False,
                        "error": f"找不到设备『{row.get('device_name') or '?'}』的地址 ——"
                                 f"确认「工具 → 设备池」里有这台设备且已连接"}), 400
    if row.get("device_name") and not row.get("serial"):
        vp.set_serial(pid, serial)      # 顺手把空快照补上（自愈，免得下次又踩）
    if row["status"] in (vp.ST_PUSHING, vp.ST_PUBLISHING):
        return jsonify({"ok": False, "error": "这条已经推到手机上了，别重复推"}), 409
    if row["status"] == vp.ST_UNKNOWN:
        return jsonify({"ok": False,
                        "error": "这条上次结果未知（可能已经发出去了）—— 先去抖音确认，"
                                 "再用「已发出 / 未发出」裁决"}), 409
    from tasks.generic.publish_flow import run_push
    t = threading.Thread(target=run_push, args=(serial, pid), daemon=True)
    t.start()
    _log.info(f"推送素材到手机 video_plan {pid}（{row['phone']} {row['release_date']}）")
    return jsonify({"ok": True,
                    "msg": "已开始推送到手机（推完状态会变成「已推送到手机」，"
                           "之后你在手机上发；发完点「标记已发布」）"})


def _serial_for_plan(row):
    """这条计划要推给哪台设备：**设备名现查**（设备换 IP 后快照是错的），快照兜底。"""
    from core import ledger
    return (ledger.serial_of(row.get("device_name") or "") or (row.get("serial") or "")).strip()


@bp.route("/api/video_plan/push_all", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_vp_push_all():
    """**一键推送**：把某天（默认今天）所有「待发布/失败」的计划推到各自的手机上。

    并发口径：**每台设备一个线程、设备内串行**（同一个手机同时推两个文件会互相干扰：
    两条 adb push 抢带宽 + 两条 `content call scan_file` 抢索引），设备之间并行。

    立刻返回受理条数（推送是后台跑的）——进度与结果看时间线的状态徽章/
    「已推送·相册可见」标记。
    """
    import threading
    from core import ledger
    data = request.json or {}
    date = (data.get("date") or "").strip() or vp._today()
    rows = vp.pushable_plans(date, retry_failed=bool(data.get("retry_failed", True)))
    if not rows:
        return jsonify({"ok": True, "count": 0, "devices": 0,
                        "msg": f"{date} 没有待推送的计划（有标题的才推）"})
    by_dev, no_dev = {}, []
    for row in rows:
        serial = (ledger.serial_of(row.get("device_name") or "") or (row.get("serial") or "")).strip()
        (by_dev.setdefault(serial, []) if serial else no_dev).append(row)
    by_dev.pop("", None)                    # 兜底：空 key 不该出现
    if not by_dev:
        return jsonify({"ok": False, "error": f"{len(rows)} 条计划都没绑上设备地址 ——"
                                              f"到「工具 → 设备池」确认设备在线"}), 400

    from tasks.generic.publish_flow import push_one

    def _push_device(serial, plans):
        for row in plans:
            try:
                push_one(serial, row)
            except Exception as e:
                _log.warning(f"一键推送 {row.get('id')} 异常: {e}")

    for serial, plans in by_dev.items():
        threading.Thread(target=_push_device, args=(serial, plans), daemon=True).start()
    _log.info(f"一键推送 {date}：{len(rows)} 条 → {len(by_dev)} 台设备"
              f"（没绑设备的 {len(no_dev)} 条已跳过）")
    return jsonify({"ok": True, "count": len(rows), "devices": len(by_dev),
                    "no_device": len(no_dev),
                    "msg": f"已开始推送 {len(rows)} 条（{len(by_dev)} 台设备，每台串行）——"
                           f"推完状态会变成「已推送·相册可见/相册未见」，刷新看结果"})


@bp.route("/api/video_plan/<pid>/mark", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_vp_mark(pid):
    """人工标记结果：`{status: done|failed|unknown, why}`（推送到手机之后用）。"""
    data = request.json or {}
    st = (data.get("status") or "").strip()
    if st not in (vp.ST_DONE, vp.ST_FAILED, vp.ST_UNKNOWN):
        return jsonify({"ok": False, "error": "status 只能是 done / failed / unknown"}), 400
    why = (data.get("why") or "").strip()
    row, err = vp.set_status(pid, st, note=why)
    if err:
        return jsonify({"ok": False, "error": err}), 400
    _log.info(f"人工标记 video_plan {pid} → {st}")
    return jsonify({"ok": True, "msg": "已标记", "plan": row})


@bp.route("/api/video_plan/<pid>/<action>", methods=["POST"])
@perm_required(PERM_DEVICES)
def api_vp_action(pid, action):
    """单条操作：skip 跳过 / retry 重试（failed→ready）/ resolve 裁决 unknown。"""
    data = request.json or {}
    if action == "skip":
        row, err = vp.set_status(pid, vp.ST_SKIPPED, note=data.get("reason") or "")
        msg = "已跳过"
    elif action == "retry":
        row, err = vp.set_status(pid, vp.ST_READY, note="人工重试")
        msg = "已放回待发布（尝试次数已清零）"
    elif action == "resolve":
        row, err = vp.resolve_unknown(pid, bool(data.get("published")),
                                      note=data.get("note") or "")
        msg = "已记为「已发出」" if data.get("published") else "已记为「未发出」（可重试）"
    else:
        return jsonify({"ok": False, "error": f"未知操作 {action}"}), 400
    if err:
        return jsonify({"ok": False, "error": err}), 400
    return jsonify({"ok": True, "msg": msg, "plan": row})


def _norm_target(target):
    """规范化任务目标。**分组键必须是 `group_name`** —— `TaskJob.resolve_serials()`
    只认这个键，写成 `group` 会静默解析出 0 台设备（任务"跑完了"但谁也没跑）。"""
    t = dict(target or {})
    mode = (t.get("mode") or "all").strip()
    out = {"mode": mode}
    if mode == "group":
        out["group_name"] = (t.get("group_name") or t.get("group") or "").strip()
    elif mode == "serial":
        out["serial"] = (t.get("serial") or "").strip()
    return out


def _target_error(target):
    if target["mode"] not in ("all", "group", "serial"):
        return "目标只能是 全部/分组/单设备"
    if target["mode"] == "group" and not target.get("group_name"):
        return "选了「设备分组」但没指定分组名"
    if target["mode"] == "serial" and not target.get("serial"):
        return "选了「单台设备」但没指定设备"
    return ""


def _parse_hhmm(text):
    try:
        h, m = [int(x) for x in str(text).split(":")]
        assert 0 <= h < 24 and 0 <= m < 60
        return h, m, ""
    except Exception:
        return 0, 0, "时间格式应为 HH:MM"


def _count_steps(steps):
    """步骤总数（**含嵌套**：条件判断的 then/else、循环块里的）—— 界面上给人看的数字，
    只数顶层会让人以为任务很短（骨架顶层 4 步、实际 14 步）。"""
    n = 0
    for s in steps or []:
        if not isinstance(s, dict):
            continue
        n += 1
        p = s.get("params") or {}
        for k in ("then", "else", "children"):
            n += _count_steps(p.get(k) or [])
    return n


def _job_brief(j, mgr):
    """任务摘要（列表用）+ 步骤（就地编辑用）。"""
    steps = (j.params or {}).get("steps") or []
    try:
        nxt = mgr.next_run_of(j)
        nxt = nxt.strftime("%Y-%m-%d %H:%M") if nxt else ""
    except Exception:
        nxt = ""
    sch = j.schedule or {}
    return {"id": j.id, "name": j.name, "enabled": bool(j.enabled),
            "mode": sch.get("mode", "once"), "cron": sch.get("cron", ""),
            "next_run": nxt, "target": j.target, "steps": steps,
            "step_count": _count_steps(steps),
            "retry": j.retry or {},
            "schedule": sch}


@bp.route("/api/video_plan/tasks")
@perm_required(PERM_DEVICES)
def api_vp_tasks():
    """**可编辑的任务列表** —— 「发布计划」页顶部那一块。

    返回**全部通用步骤任务**，每条带 `is_release`（含「推送/标记发布结果」= 发布任务）：
    - 只列发布任务的话，手写的抖音发布流程（自己点相册/输入框/发布，没有平台的
      `push_release` 步骤）就会**一条都不显示** —— 人会以为"这页没有能改的地方"
    - 界面上按 `is_release` 分成两组：上面「发布任务」，下面「其它任务」也能就地改步骤

    带上完整 `steps`：这一页要**就地编辑**这些步骤（抓选择器），不用再跳「任务」页。
    """
    from web import context
    release, other = [], []
    try:
        jobs = list((context.mgr.jobs or {}).values())
    except Exception:
        jobs = []
    for j in jobs:
        # 只有通用步骤任务有步骤可编（其它任务类型的 params 是另一套）
        if getattr(j, "task_type", "generic_steps") != "generic_steps":
            continue
        b = _job_brief(j, context.mgr)
        b["is_release"] = vp.is_release_task((j.params or {}).get("steps") or [])
        b["has_push"] = vp.has_push_step((j.params or {}).get("steps") or [])
        (release if b["is_release"] else other).append(b)
    release.sort(key=lambda x: x["name"])
    other.sort(key=lambda x: x["name"])
    return jsonify({"ok": True, "tasks": release + other,
                    "release_count": len(release), "other_count": len(other)})


@bp.route("/api/video_plan/tasks", methods=["POST"])
@perm_required(PERM_TASKS)
def api_vp_task_create():
    """一键新建「标准发布任务」：账号校验 → 推送 → 抖音点击（占位待你抓）→ 填标题 → 标记。

    请求：{"name": "发布视频", "target": {"mode":"all"|"group"|"serial", …},
           "time": "10:00", "enabled": true}
    **走 TaskManager 建**（内存 + 库一起更新）—— 直接写库的话调度器不认。
    """
    from web import context
    data = request.json or {}
    name = (data.get("name") or "发布视频").strip()[:120]
    target = _norm_target(data.get("target"))
    err = _target_error(target)
    if err:
        return jsonify({"ok": False, "error": err}), 400
    h, m, err = _parse_hhmm((data.get("time") or "10:00").strip())
    if err:
        return jsonify({"ok": False, "error": err}), 400
    schedule = {"mode": "cron", "cron": f"{m} {h} * * *", "stop_cron": "",
                "window": {"start": "", "end": ""}}
    params = {"steps": vp.build_release_steps(),
              "dedup_reset": "day", "dedup_hours": 6,
              "preempt": False, "skip_offline": True, "max_duration": 0}
    try:
        job = context.mgr.add_job(name=name, task_type="generic_steps", target=target,
                                  params=params, schedule=schedule,
                                  retry={"max_attempts": 1, "delay": 60},
                                  enabled=bool(data.get("enabled", True)))
    except Exception as e:
        _log.warning(f"新建发布任务失败: {e}")
        return jsonify({"ok": False, "error": f"创建失败: {str(e)[:120]}"}), 500
    _log.info(f"新建发布任务 {job.name}({job.id})，每天 "
              f"{h:02d}:{m:02d}，目标 {target}")
    return jsonify({"ok": True, "msg": f"已创建「{name}」（每天 {h:02d}:{m:02d} 跑）——"
                                       f"就在下面这一块把抖音那几步的选择器抓一下就能用了",
                    "job_id": job.id, "task": _job_brief(job, context.mgr)})


@bp.route("/api/video_plan/tasks/<job_id>/adopt", methods=["POST"])
@perm_required(PERM_TASKS)
def api_vp_task_adopt(job_id):
    """给**已有任务**插上平台那两步（`push_release` 放最前、`mark_release` 放最后）。

    手写的抖音发布流程常缺这两步：不插的话计划页的视频永远不会被推到手机、
    发布结果也不会回写状态。**中间你自己的步骤一步不动**（位置理由见
    `core.video_plan.insert_release_steps`）。
    """
    from web import context
    job = (context.mgr.jobs or {}).get(job_id)
    if not job:
        return jsonify({"ok": False, "error": "任务不存在"}), 404
    if getattr(job, "task_type", "generic_steps") != "generic_steps":
        return jsonify({"ok": False, "error": "只有「通用步骤」任务能插平台发布步骤"}), 400
    steps = (job.params or {}).get("steps") or []
    if not steps:
        return jsonify({"ok": False, "error": "这个任务没有步骤，先去「任务」页把抖音流程画好"}), 400
    if vp.has_push_step(steps) and vp.has_mark_step(steps):
        return jsonify({"ok": False, "error": "这个任务已经有平台发布步骤了"}), 409
    params = dict(job.params or {})
    params["steps"] = vp.insert_release_steps(steps)
    try:
        context.mgr.update_job(job_id, params=params)
    except Exception as e:
        _log.warning(f"给任务 {job_id} 插平台发布步骤失败: {e}")
        return jsonify({"ok": False, "error": f"保存失败: {str(e)[:120]}"}), 500
    _log.info(f"给任务 {job.name}({job_id}) 插上平台发布步骤")
    return jsonify({"ok": True, "msg": "已插入「推送发布视频」（最前）与「标记发布结果」（最后）——"
                                       "你原来的步骤没动；到「账号 → 发布计划」里核对一下顺序",
                    "task": _job_brief(job, context.mgr)})


@bp.route("/api/video_plan/tasks/<job_id>", methods=["GET", "PUT"])
@perm_required(PERM_TASKS)
def api_vp_task_one(job_id):
    """就地编辑发布任务（「账号 → 发布计划」页里，不用跳「任务」页）。

    只放开这一页用得上的字段：`name` / `time`(HH:MM) / `target` / `steps` / `enabled`。
    **没传的字段原样保留**（深合并：`params` 里还有去重有效期、抢占等通用参数，
    整块覆盖会把它们抹掉）。底层仍是 `context.mgr.update_job()` —— 内存 + 库一起改，
    否则调度器不认（直接写库的坑见 `core/task_manager` 的 `_save_jobs`）。
    """
    from web import context
    job = (context.mgr.jobs or {}).get(job_id)
    if not job:
        return jsonify({"ok": False, "error": "任务不存在"}), 404
    if request.method == "GET":
        return jsonify({"ok": True, "task": _job_brief(job, context.mgr)})
    data = request.json or {}
    fields = {}
    if "name" in data:
        nm = (data.get("name") or "").strip()[:120]
        if not nm:
            return jsonify({"ok": False, "error": "任务名不能为空"}), 400
        fields["name"] = nm
    if "target" in data:
        target = _norm_target(data.get("target"))
        err = _target_error(target)
        if err:
            return jsonify({"ok": False, "error": err}), 400
        fields["target"] = target
    if "enabled" in data:
        fields["enabled"] = bool(data.get("enabled"))
    if "time" in data:
        h, m, err = _parse_hhmm((data.get("time") or "").strip())
        if err:
            return jsonify({"ok": False, "error": err}), 400
        sch = dict(job.schedule or {})
        sch["mode"] = "cron"
        sch["cron"] = f"{m} {h} * * *"
        sch.setdefault("stop_cron", "")
        sch.setdefault("window", {"start": "", "end": ""})
        fields["schedule"] = sch
    if "steps" in data:
        steps = data.get("steps")
        if not isinstance(steps, list):
            return jsonify({"ok": False, "error": "steps 必须是数组"}), 400
        params = dict(job.params or {})
        params["steps"] = steps
        fields["params"] = params
    if not fields:
        return jsonify({"ok": False, "error": "没有要改的字段"}), 400
    try:
        context.mgr.update_job(job_id, **fields)
    except Exception as e:
        _log.warning(f"改发布任务 {job_id} 失败: {e}")
        return jsonify({"ok": False, "error": f"保存失败: {str(e)[:120]}"}), 500
    _log.info(f"改发布任务 {job.name}({job_id})：{'/'.join(fields)}")
    return jsonify({"ok": True, "msg": "已保存", "task": _job_brief(job, context.mgr)})


@bp.route("/api/video_plan/links")
@perm_required(PERM_DEVICES)
def api_vp_links():
    """已发布作品的分享链接（界面展示 / `format=csv` 导出，给后续铺评论用）。"""
    rows = vp.links(date_from=(request.args.get("from") or "").strip(),
                    date_to=(request.args.get("to") or "").strip(),
                    device=(request.args.get("device") or "").strip(),
                    phone=(request.args.get("phone") or "").strip())
    if (request.args.get("format") or "") == "csv":
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(["发布日期", "设备号", "手机号", "抖音号",
                    "编号", "标题", "正文", "分享链接", "发布时间"])
        for r in rows:
            w.writerow([r["release_date"], r["device_name"], r["phone"],
                        r["douyin_id"], r["seq"], r["title"], r.get("body", ""),
                        r["share_url"], r["published_at"]])
        data = ("" + buf.getvalue()).encode("utf-8")   # BOM：Excel 打开中文不乱码
        return send_file(io.BytesIO(data), mimetype="text/csv", as_attachment=True,
                         download_name=f"video_links_{vp._today()}.csv")
    return jsonify({"ok": True, "links": rows, "count": len(rows)})