ykxmz/_build/clean_dup_assets.py

# -*- coding: utf-8 -*-
r'''统一清理两个平台发行版里的重复附件。

背景：
    重发时如果只"复用已有发行版"，附件会一次次堆积。
    Gitee 尤其绕：`GET /releases/tags/{tag}` 返回的 `assets`
    **只有 name 和 download_url，没有 id**，所以没法删。
    但另一个接口有 id：
        GET    /repos/{o}/{r}/releases/{id}/attach_files  -> 带 id
        DELETE /repos/{o}/{r}/releases/{id}/attach_files/{attach_id}

    Gitea/EzHub 则直接有 assets[].id。

策略：同名附件只保留最新的一份，其余删掉。
支持 --dry 先看要删什么。
'''
from __future__ import annotations

import json
import sys
import time
import urllib.error
import urllib.request
from collections import defaultdict
from pathlib import Path

ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote")
GT = (ROOT / "_build" / ".gitee_token").read_text(encoding="utf-8").strip()
ET = (Path.home() / ".ykxmz" / "ezhub_token").read_text(
    encoding="utf-8").strip()
TAG = "v1.8.0"

# 这两个是 Gitee 自动生成的源码包，不是我们传的，不要动
KEEP_ALWAYS = {"v1.8.0.zip", "v1.8.0.tar.gz"}


def http(url: str, method: str = "GET", kind: str = "gitea",
         token: str = "", tries: int = 3):
    """带重试的请求。网络会抖，不重试会误判。"""
    headers = {"User-Agent": "ykxmz-clean"}
    if kind == "gitee":
        sep = "&" if "?" in url else "?"
        url = url + sep + "access_token=" + token
    else:
        headers["Authorization"] = "token " + token
    last = None
    for attempt in range(1, tries + 1):
        try:
            req = urllib.request.Request(url, method=method, headers=headers)
            with urllib.request.urlopen(req, timeout=90) as resp:
                text = resp.read().decode("utf-8", "replace")
                return resp.status, (json.loads(text) if text.strip() else {})
        except urllib.error.HTTPError as exc:
            return exc.code, exc.read().decode("utf-8", "replace")[:200]
        except Exception as exc:                      # noqa: BLE001
            last = exc
            if attempt < tries:
                time.sleep(4)
    return 0, str(last)


def list_gitee_assets(rid: int) -> list:
    """Gitee 要单独调 attach_files 才拿得到 id。"""
    status, data = http(
        "https://gitee.com/api/v5/repos/xmz7891/ykxmz/releases/%s"
        "/attach_files" % rid, kind="gitee", token=GT)
    if status != 200 or not isinstance(data, list):
        print("    [失败] 取 Gitee 附件列表 HTTP %s: %s" % (status, data))
        return []
    return data


def list_gitea_assets(rid: int) -> list:
    status, data = http(
        "https://git.261826.xyz/api/v1/repos/xmz/ykxmz/releases/%s"
        % rid, kind="gitea", token=ET)
    if status != 200:
        print("    [失败] 取 EzHub 发行版 HTTP %s" % status)
        return []
    return data.get("assets") or []


def clean(platform: str, rid: int, assets: list, dry: bool) -> int:
    groups = defaultdict(list)
    for a in assets:
        name = a.get("name") or "?"
        if name in KEEP_ALWAYS:
            continue
        groups[name].append(a)

    removed = 0
    for name, items in sorted(groups.items()):
        if len(items) <= 1:
            continue
        # 新的排前面
        items.sort(key=lambda x: str(x.get("created_at") or ""), reverse=True)
        keep = items[0]
        print("    %-32s %d 份 -> 保留 id=%s" % (name, len(items),
                                                keep.get("id")))
        for extra in items[1:]:
            aid = extra.get("id")
            if dry:
                print("        （演练）会删 id=%s" % aid)
                removed += 1
                continue
            if platform == "gitee":
                url = ("https://gitee.com/api/v5/repos/xmz7891/ykxmz"
                       "/releases/%s/attach_files/%s" % (rid, aid))
            else:
                url = ("https://git.261826.xyz/api/v1/repos/xmz/ykxmz"
                       "/releases/%s/assets/%s" % (rid, aid))
            status, body = http(url, method="DELETE",
                                kind=("gitee" if platform == "gitee"
                                      else "gitea"),
                                token=(GT if platform == "gitee" else ET))
            if status in (200, 204):
                print("        ✓ 删了 id=%s" % aid)
                removed += 1
            else:
                print("        ✗ id=%s HTTP %s %s"
                      % (aid, status, str(body)[:80]))
    return removed


def main() -> int:
    dry = "--dry" in sys.argv
    total = 0

    print("  ═══ Gitee ═══")
    status, rel = http("https://gitee.com/api/v5/repos/xmz7891/ykxmz"
                       "/releases/tags/" + TAG, kind="gitee", token=GT)
    if status == 200:
        rid = rel.get("id")
        assets = list_gitee_assets(rid)
        print("    发行版 id=%s，可删附件 %d 个" % (rid, len(assets)))
        total += clean("gitee", rid, assets, dry)
    else:
        print("    [失败] HTTP %s" % status)
    print()

    print("  ═══ EzHub ═══")
    status, rel = http("https://git.261826.xyz/api/v1/repos/xmz/ykxmz"
                       "/releases/tags/" + TAG, kind="gitea", token=ET)
    if status == 200:
        rid = rel.get("id")
        assets = list_gitea_assets(rid)
        print("    发行版 id=%s，附件 %d 个" % (rid, len(assets)))
        total += clean("gitea", rid, assets, dry)
    else:
        print("    [失败] HTTP %s" % status)

    print()
    print("  %s %d 个" % ("演练：会删" if dry else "共删", total))
    return 0


if __name__ == "__main__":
    sys.exit(main())