#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ gitea_ops.py — Gitea 项目管理 CLI（PineSound 组织专用，也可用于任意 org/repo） 封装 Gitea 1.26 API 的常用项目管理操作： 组织 / 仓库 / 人员 / 工单 / 标签 / 里程碑 / 仓库信息（描述、topics、头像） 注意：Gitea 1.26 的「看板（Projects）」没有 REST API，必须走网页操作， 请使用同目录的 board_ops.py（Playwright）或 SKILL.md 中的网页操作指引。 环境变量： GITEA_BASE Gitea 地址（默认 http://192.168.1.3:10082） GITEA_TOKEN API token（默认读取 /app/working.secret/gitea_token） GITEA_ORG 默认组织（默认 PineSound） 用法示例： python3 gitea_ops.py org repos python3 gitea_ops.py org members python3 gitea_ops.py issue list PineSoundServer --state open python3 gitea_ops.py issue create PineSoundServer --title "测试" --body "内容" --labels "🐛 缺陷（Bug）,🟠 高（High）" --milestone "Beta-V1.1" --assignee YangYi python3 gitea_ops.py issue close PineSoundServer 8 python3 gitea_ops.py issue delete PineSoundServer 99 --yes python3 gitea_ops.py milestone list PineSoundServer python3 gitea_ops.py label list PineSoundServer python3 gitea_ops.py repo topics PineSoundServer --topics backend,go,api python3 gitea_ops.py avatar PineSoundServer /tmp/avatar.png 所有子命令都可用 --json 输出原始 JSON（便于管道处理）。 """ import argparse import base64 import json import os import sys import urllib.error import urllib.parse import urllib.request BASE = os.environ.get("GITEA_BASE", "http://192.168.1.3:10082/api/v1") TOKEN = os.environ.get("GITEA_TOKEN", "") if not TOKEN: _tf = "/app/working.secret/gitea_token" if os.path.exists(_tf): TOKEN = open(_tf).read().strip() ORG = os.environ.get("GITEA_ORG", "PineSound") def api(method, path, body=None, params=None, raw=False): """调用 Gitea API。params 会拼到 query string。返回 (status, parsed)。""" url = BASE + path if params: url += "?" + urllib.parse.urlencode(params) req = urllib.request.Request(url, method=method) req.add_header("Authorization", "token " + TOKEN) if body is not None: req.add_header("Content-Type", "application/json") data = json.dumps(body).encode() else: data = None try: with urllib.request.urlopen(req, data=data, timeout=30) as r: content = r.read().decode() return r.status, content if raw else (json.loads(content) if content else None) except urllib.error.HTTPError as e: return e.code, e.read().decode()[:500] except Exception as e: return -1, str(e) # ---------- 输出 ---------- def out(obj, as_json=False): if as_json: print(json.dumps(obj, ensure_ascii=False, indent=1)) elif isinstance(obj, list): for o in obj: print(json.dumps(o, ensure_ascii=False)) elif isinstance(obj, dict): print(json.dumps(obj, ensure_ascii=False, indent=1)) else: print(obj) def resolve_milestone_id(repo, ref): """milestone 参数支持 id 或标题，解析为 id。""" if isinstance(ref, int): return ref if str(ref).isdigit(): return int(ref) st, data = api("GET", f"/repos/{ORG}/{repo}/milestones", params={"state": "all"}) if st != 200: return None for m in data: if m.get("title") == ref: return m.get("id") return None def resolve_label_ids(repo, names): """labels 参数支持逗号分隔的标签名，解析为 id 列表。""" if not names: return [] st, data = api("GET", f"/repos/{ORG}/{repo}/labels", params={"limit": 100}) if st != 200: return [] by_name = {l.get("name"): l.get("id") for l in data} ids = [] for n in names: n = n.strip() if n in by_name: ids.append(by_name[n]) else: print(f" [warn] 标签不存在: {n}", file=sys.stderr) return ids # ---------- 子命令 ---------- def cmd_config(args): out({"base": BASE, "web_base": BASE.replace("/api/v1", ""), "org": ORG, "token_set": bool(TOKEN), "token_tail": TOKEN[-6:] if TOKEN else ""}, args.json) def cmd_org_repos(args): st, data = api("GET", f"/orgs/{args.org}/repos", params={"limit": 100}) if st != 200: out({"error": st, "msg": data}, args.json); return if args.json: out(data, True) return for r in sorted(data, key=lambda x: x["name"]): print(f"{r['name']:28s} {'私有' if r.get('private') else '公开'} stars={r.get('stars_count',0)} {r.get('description') or ''}") def cmd_org_members(args): st, data = api("GET", f"/orgs/{args.org}/members") if st != 200: out({"error": st, "msg": data}, args.json); return if args.json: out(data, True) return for u in data: print(f"{u.get('login'):20s} {u.get('full_name') or ''} {u.get('email') or ''}") def cmd_issue_list(args): params = {"limit": args.limit, "state": args.state} if args.milestone: mid = resolve_milestone_id(args.repo, args.milestone) if mid is None: out({"error": "milestone not found: " + args.milestone}, args.json); return params["milestones"] = mid if args.label: lids = resolve_label_ids(args.repo, [args.label]) if lids: params["labels"] = ",".join(str(i) for i in lids) if args.assignee: params["assignee"] = args.assignee st, data = api("GET", f"/repos/{ORG}/{args.repo}/issues", params=params) if st != 200: out({"error": st, "msg": data}, args.json); return # 过滤 PR data = [i for i in data if i.get("pull_request") is None] if args.json: out(data, True) return for i in data: labels = ",".join(l["name"] for l in (i.get("labels") or [])) ms = (i.get("milestone") or {}).get("title", "") assignees = ",".join(a["login"] for a in (i.get("assignees") or [])) print(f"#{i['number']:>4} [{i['state']:>6}] {i['title'][:50]} {ms} [{labels}] @{assignees}") def cmd_issue_get(args): st, data = api("GET", f"/repos/{ORG}/{args.repo}/issues/{args.number}") out({"error": st, "msg": data} if st != 200 else data, args.json or st != 200) def cmd_issue_create(args): body = {"title": args.title, "body": args.body or ""} if args.assignee: body["assignees"] = [a.strip() for a in args.assignee.split(",")] if args.labels: body["labels"] = resolve_label_ids(args.repo, args.labels.split(",")) if args.milestone: mid = resolve_milestone_id(args.repo, args.milestone) if mid is None: out({"error": "milestone not found: " + args.milestone}, args.json); return body["milestone"] = mid if args.deadline: body["due_date"] = args.deadline + "T23:59:59+08:00" st, data = api("POST", f"/repos/{ORG}/{args.repo}/issues", body=body) if st != 201: out({"error": st, "msg": data}, args.json); return print(f"created #{data.get('number')}: {data.get('html_url')}" if not args.json else json.dumps(data, ensure_ascii=False, indent=1)) def _ok(st): return 200 <= st < 300 def cmd_issue_edit(args): body = {} if args.title: body["title"] = args.title if args.body is not None: body["body"] = args.body if args.state: body["state"] = args.state if args.labels is not None: # Gitea 1.26 的 PATCH /issues/{index} 的 labels 字段不生效（实测！）， # 必须用 PUT /issues/{index}/labels（Replace）才能替换标签。 lid_list = resolve_label_ids(args.repo, args.labels.split(",")) if args.labels else [] api("PUT", f"/repos/{ORG}/{args.repo}/issues/{args.number}/labels", body={"labels": lid_list}) if args.assignee is not None: body["assignees"] = [a.strip() for a in args.assignee.split(",")] if args.assignee else [] body["clear_assignees"] = not bool(args.assignee) if args.milestone is not None: if args.milestone == "none": body["unset_milestone"] = True else: mid = resolve_milestone_id(args.repo, args.milestone) if mid is None: out({"error": "milestone not found: " + args.milestone}, args.json); return body["milestone"] = mid if args.deadline is not None: if args.deadline == "none": body["unset_due_date"] = True else: body["due_date"] = args.deadline + "T23:59:59+08:00" st, data = api("PATCH", f"/repos/{ORG}/{args.repo}/issues/{args.number}", body=body) if not _ok(st): out({"error": st, "msg": data}, args.json or True) else: out({"ok": True, "number": args.number, "state": data.get("state"), "title": data.get("title"), "labels": [l.get("name") for l in (data.get("labels") or [])]}, args.json or False) def cmd_issue_state(args, state): st, data = api("PATCH", f"/repos/{ORG}/{args.repo}/issues/{args.number}", body={"state": state}) if args.comment: api("POST", f"/repos/{ORG}/{args.repo}/issues/{args.number}/comments", body={"body": args.comment}) out({"error": st, "msg": data} if not _ok(st) else {"ok": True, "number": args.number, "state": state}, args.json or (not _ok(st))) def cmd_issue_comment(args): st, data = api("POST", f"/repos/{ORG}/{args.repo}/issues/{args.number}/comments", body={"body": args.body}) out({"error": st, "msg": data} if not _ok(st) else {"ok": True, "id": data.get("id")} if isinstance(data, dict) else data, args.json) def cmd_issue_delete(args): if not args.yes: print(f"危险操作！删除工单 #{args.number} 不可恢复。请加 --yes 确认。"); return st, data = api("DELETE", f"/repos/{ORG}/{args.repo}/issues/{args.number}") print({"error": st, "msg": data} if not _ok(st) else {"ok": True, "deleted": args.number}) def cmd_milestone_list(args): st, data = api("GET", f"/repos/{ORG}/{args.repo}/milestones", params={"state": args.state}) if st != 200: out({"error": st, "msg": data}, args.json); return if args.json: out(data, True); return for m in sorted(data, key=lambda x: x["id"]): d = (m.get("due_on") or "")[:10] print(f"#{m['id']:>4} [{m.get('state') or m.get('status'):>6}] {m.get('title')} 截止 {d} 完成 {m.get('closed_issues', 0)}/{m.get('open_issues', 0) + m.get('closed_issues', 0)}") def cmd_milestone_create(args): body = {"title": args.title, "description": args.description or ""} if args.deadline: body["due_on"] = args.deadline + "T23:59:59+08:00" st, data = api("POST", f"/repos/{ORG}/{args.repo}/milestones", body=body) out({"error": st, "msg": data} if not _ok(st) else {"ok": True, "id": data.get("id"), "title": data.get("title")}, args.json or (not _ok(st))) def cmd_milestone_edit(args): body = {} if args.title: body["title"] = args.title if args.description is not None: body["description"] = args.description if args.deadline is not None: body["due_on"] = args.deadline + "T23:59:59+08:00" if args.state: body["state"] = args.state st, data = api("PATCH", f"/repos/{ORG}/{args.repo}/milestones/{args.id}", body=body) out({"error": st, "msg": data} if not _ok(st) else {"ok": True, "id": args.id}, args.json or (not _ok(st))) def cmd_milestone_delete(args): if not args.yes: print("危险操作！删除里程碑不可恢复。请加 --yes 确认。"); return st, data = api("DELETE", f"/repos/{ORG}/{args.repo}/milestones/{args.id}") print({"error": st, "msg": data} if not _ok(st) else {"ok": True, "deleted": args.id}) def cmd_label_list(args): if args.org_level: st, data = api("GET", f"/orgs/{args.org}/labels", params={"limit": 100}) title = f"组织级标签 {args.org}" else: st, data = api("GET", f"/repos/{ORG}/{args.repo}/labels", params={"limit": 100}) title = f"仓库标签 {args.repo}" if st != 200: out({"error": st, "msg": data}, args.json); return if args.json: out(data, True); return print(f"== {title} ==") for l in sorted(data, key=lambda x: x["id"]): print(f"#{l['id']:>4} {l['name']} ({l['color']}) {l.get('description') or ''}") def cmd_label_create(args): if args.org_level: st, data = api("POST", f"/orgs/{args.org}/labels", body={"name": args.name, "color": args.color.lstrip('#'), "description": args.description or ""}) else: st, data = api("POST", f"/repos/{ORG}/{args.repo}/labels", body={"name": args.name, "color": args.color.lstrip('#'), "description": args.description or ""}) out({"error": st, "msg": data} if not _ok(st) else {"ok": True, "id": data.get("id"), "name": data.get("name")}, args.json or (not _ok(st))) def cmd_label_edit(args): body = {} if args.name: body["name"] = args.name if args.color: body["color"] = args.color.lstrip('#') if args.description is not None: body["description"] = args.description if args.org_level: st, data = api("PATCH", f"/orgs/{args.org}/labels/{args.id}", body=body) else: st, data = api("PATCH", f"/repos/{ORG}/{args.repo}/labels/{args.id}", body=body) out({"error": st, "msg": data} if not _ok(st) else {"ok": True, "id": args.id}, args.json or (not _ok(st))) def cmd_label_delete(args): if not args.yes: print("危险操作！删除标签不可恢复。请加 --yes 确认。"); return if args.org_level: st, data = api("DELETE", f"/orgs/{args.org}/labels/{args.id}") else: st, data = api("DELETE", f"/repos/{ORG}/{args.repo}/labels/{args.id}") print({"error": st, "msg": data} if not _ok(st) else {"ok": True, "deleted": args.id}) def cmd_repo_info(args): st, data = api("GET", f"/repos/{ORG}/{args.repo}") out({"error": st, "msg": data} if st != 200 else { "name": data.get("name"), "description": data.get("description"), "topics": data.get("topics"), "website": data.get("website"), "private": data.get("private"), "default_branch": data.get("default_branch"), "avatar_url": data.get("avatar_url"), "stars": data.get("stars_count"), }, args.json or st != 200) def cmd_repo_topics(args): topics = [t.strip() for t in args.topics.split(",") if t.strip()] st, data = api("PUT", f"/repos/{ORG}/{args.repo}/topics", body={"topics": topics}) print({"error": st, "msg": data} if st not in (200, 204) else {"ok": True, "topics": topics}) def cmd_avatar(args): with open(args.image, "rb") as f: b64 = base64.b64encode(f.read()).decode() st, data = api("POST", f"/repos/{ORG}/{args.repo}/avatar", body={"image": b64}) print({"error": st, "msg": data} if st not in (200, 204) else {"ok": True, "avatar_updated": args.repo}) def cmd_board_help(args): print("""看板（Projects）在 Gitea 1.26 没有 REST API，必须网页操作。 请使用同目录的 board_ops.py（基于 Playwright 的网页自动化 CLI）： python3 board_ops.py boards PineSoundServer python3 board_ops.py create PineSoundServer --name "开发看板" python3 board_ops.py add-issue PineSoundServer --issue 8 --board 3 python3 board_ops.py move PineSoundServer --issue 8 --board 3 --to "🛠 开发中" 详细用法见 SKILL.md 或 python3 board_ops.py -h""") # ---------- main ---------- def build_parser(): p = argparse.ArgumentParser(prog="gitea_ops.py", description="Gitea 项目管理 CLI") p.add_argument("--org", default=ORG, help=f"组织名（默认 {ORG}）") p.add_argument("--json", action="store_true", help="输出原始 JSON") sub = p.add_subparsers(dest="cmd", required=True) sub.add_parser("config", help="显示当前配置").set_defaults(func=cmd_config) pr = sub.add_parser("org", help="组织操作") orgsub = pr.add_subparsers(dest="org_cmd", required=True) orgsub.add_parser("repos", help="组织仓库列表").set_defaults(func=cmd_org_repos) orgsub.add_parser("members", help="组织成员（人员）列表").set_defaults(func=cmd_org_members) pi = sub.add_parser("issue", help="工单操作") isub = pi.add_subparsers(dest="issue_cmd", required=True) li = isub.add_parser("list", help="工单列表") li.add_argument("repo"); li.add_argument("--state", default="open", choices=["open", "closed", "all"]) li.add_argument("--limit", type=int, default=100); li.add_argument("--milestone"); li.add_argument("--label"); li.add_argument("--assignee") li.set_defaults(func=cmd_issue_list) gi = isub.add_parser("get", help="工单详情") gi.add_argument("repo"); gi.add_argument("number", type=int) gi.set_defaults(func=cmd_issue_get) ci = isub.add_parser("create", help="创建工单") ci.add_argument("repo"); ci.add_argument("--title", required=True); ci.add_argument("--body") ci.add_argument("--assignee"); ci.add_argument("--labels"); ci.add_argument("--milestone"); ci.add_argument("--deadline") ci.set_defaults(func=cmd_issue_create) ei = isub.add_parser("edit", help="编辑工单") ei.add_argument("repo"); ei.add_argument("number", type=int) ei.add_argument("--title"); ei.add_argument("--body"); ei.add_argument("--state", choices=["open", "closed"]) ei.add_argument("--assignee", help="逗号分隔用户名，传空串清空"); ei.add_argument("--labels", help="逗号分隔标签名，传空串清空") ei.add_argument("--milestone", help="id/标题，或 none 清空"); ei.add_argument("--deadline", help="YYYY-MM-DD，或 none 清空") ei.set_defaults(func=cmd_issue_edit) for name, st in (("close", "closed"), ("reopen", "open")): x = isub.add_parser(name) x.add_argument("repo"); x.add_argument("number", type=int); x.add_argument("--comment") x.set_defaults(func=lambda a, _st=st: cmd_issue_state(a, _st)) cm = isub.add_parser("comment") cm.add_argument("repo"); cm.add_argument("number", type=int); cm.add_argument("--body", required=True) cm.set_defaults(func=cmd_issue_comment) di = isub.add_parser("delete") di.add_argument("repo"); di.add_argument("number", type=int); di.add_argument("--yes", action="store_true") di.set_defaults(func=cmd_issue_delete) pm = sub.add_parser("milestone", help="里程碑操作") msub = pm.add_subparsers(dest="milestone_cmd", required=True) lm = msub.add_parser("list"); lm.add_argument("repo"); lm.add_argument("--state", default="all") lm.set_defaults(func=cmd_milestone_list) crm = msub.add_parser("create"); crm.add_argument("repo"); crm.add_argument("--title", required=True) crm.add_argument("--description"); crm.add_argument("--deadline") crm.set_defaults(func=cmd_milestone_create) em = msub.add_parser("edit"); em.add_argument("repo"); em.add_argument("id", type=int) em.add_argument("--title"); em.add_argument("--description"); em.add_argument("--deadline"); em.add_argument("--state") em.set_defaults(func=cmd_milestone_edit) dm = msub.add_parser("delete"); dm.add_argument("repo"); dm.add_argument("id", type=int) dm.add_argument("--yes", action="store_true") dm.set_defaults(func=cmd_milestone_delete) pl = sub.add_parser("label", help="标签操作（默认仓库级，--org-level 为组织级）") lsub = pl.add_subparsers(dest="label_cmd", required=True) ll = lsub.add_parser("list"); ll.add_argument("repo", nargs="?", default=""); ll.add_argument("--org-level", action="store_true") ll.set_defaults(func=cmd_label_list) cl = lsub.add_parser("create"); cl.add_argument("repo", nargs="?", default="") cl.add_argument("--name", required=True); cl.add_argument("--color", required=True, help="如 #ff0000"); cl.add_argument("--description") cl.add_argument("--org-level", action="store_true") cl.set_defaults(func=cmd_label_create) el = lsub.add_parser("edit"); el.add_argument("repo", nargs="?", default=""); el.add_argument("id", type=int) el.add_argument("--name"); el.add_argument("--color"); el.add_argument("--description"); el.add_argument("--org-level", action="store_true") el.set_defaults(func=cmd_label_edit) dl = lsub.add_parser("delete"); dl.add_argument("repo", nargs="?", default=""); dl.add_argument("id", type=int) dl.add_argument("--yes", action="store_true"); dl.add_argument("--org-level", action="store_true") dl.set_defaults(func=cmd_label_delete) pr2 = sub.add_parser("repo", help="仓库信息操作") rsub = pr2.add_subparsers(dest="repo_cmd", required=True) ir = rsub.add_parser("info"); ir.add_argument("repo"); ir.set_defaults(func=cmd_repo_info) tr = rsub.add_parser("topics"); tr.add_argument("repo"); tr.add_argument("--topics", required=True, help="逗号分隔") tr.set_defaults(func=cmd_repo_topics) ar = rsub.add_parser("avatar"); ar.add_argument("repo"); ar.add_argument("image", help="PNG 图片路径") ar.set_defaults(func=cmd_avatar) sub.add_parser("board", help="看板说明（看板无 API，用 board_ops.py）").set_defaults(func=cmd_board_help) return p def main(): args = build_parser().parse_args() if not TOKEN: print("错误：未找到 API token。请设置 GITEA_TOKEN 环境变量或写入 /app/working.secret/gitea_token", file=sys.stderr) sys.exit(1) args.func(args) if __name__ == "__main__": main()