feat: Gitea 发布 skill:隐私检查 + 建仓 + 增量推送 + 回读远程验证

This commit is contained in:
NightStar
2026-09-20 11:46:27 +08:00
parent 3515d014b1
commit df0204dff9
2 changed files with 600 additions and 239 deletions
+66 -31
View File
@@ -1,62 +1,97 @@
--- ---
name: "publish-to-gitea" name: "publish-to-gitea"
description: "将文件/目录发布到 Gitea 仓库 — 自动隐私检查、创建公开仓库、Git 推送" description: "推到 gitea/发布到仓库/建 repo 推上去/把某个 git 仓库推上去时用:跑 gitea_push.py(隐私检查 → 建仓 → 对齐远程 → 推送)→ **回读远程验证**。含脚本覆盖不到的隐私盲区与 tea 建仓坑。"
--- ---
# Publish to Gitea — 发布到 Gitea 仓库 # Publish to Gitea — 发布到 Gitea 仓库
将文件/目录推送到 Gitea 仓库(git.yoresee.cc),自动隐私检查、创建公开仓库、一键推送。 把文件/目录或一个已有 git 仓库推送到 Gitea(git.yoresee.cc)。
**不属于本 skill**:在本机**正在开发的仓库里**提交并推分支 → 直接在该仓库 `git push <remote> <branch>`(凭据走已配好的 `!tea login helper`)。push 被拒先按 `diagnose-git-push-permission` 诊断——「仓库已存在但你是 `push:false`」是常见原因,别急着建新仓(非 admin 也建不了别人的命名空间)。
## 触发条件 ## 触发条件
用户说出类似以下话语时触发: 「把这个推送到 gitea」「发布到仓库」「创建 repo 推上去」「publish to git」「推到 git.yoresee.cc」
- "把这个推送到 gitea"
- "发布到仓库"
- "创建 repo 推上去"
- "publish to git"
- "推到 git.yoresee.cc"
## 流程 ## 流程
### 1. 确定要推送的内容 ### 1. 确定要推送的内容
从上下文中推断要推送的文件/目录。常见场景: - 用户刚写的 skill / 脚本 / 项目目录 → 推那个目录
- 用户刚创建/修改了一个 skill → 推送整个 skill 目录 - 用户指定路径 → 直接用
- 用户说"把这个脚本推上去" → 推送脚本文件 - 内容是一个**已有 git 仓库**(有 `.git`、提交历史要保)→ 用 `--from-git`(见 §3)
- 用户指定了具体路径 → 直接用
### 2. 确定仓库名和描述 ### 2. 确定仓库名和描述
- **仓库名**:从内容推断,kebab-case 格式 仓库名 kebab-case,描述一句话。用户没明说则主动确认(仓库默认**公开**)。
- **描述**:一句话说明用途
- 用户没明确说则主动确认
### 3. 执行推送 ### 3. 执行推送
```bash ```bash
cd /root/.openclaw/workspace && python3 scripts/gitea_push.py <repo_name> "<description>" <files...> cd /root/.openclaw/workspace
python3 scripts/gitea_push.py <repo_name> "<description>" [paths...]
``` ```
脚本自动完成: | 场景 | 命令 |
1. 🔍 隐私检查(检测 open_id、API key、token、IP 等敏感模式) |---|---|
2. 📦 创建 Gitea **公开**仓库(已存在则复用现有仓库) | 目录/文件 → 仓库 | `gitea_push.py my-proj "描述" ./src` |
3. 📄 收集文件(排除 .git、node_modules 等) | 推已有 git 仓库(**保留完整提交历史**) | `gitea_push.py my-proj "描述" --from-git /root/my-proj` |
4. 🚀 Git 增量提交 → push(**已存在仓库先拉取远程历史,正常 commit + push,不使用 --force**;全新仓库直接首推) | 只检查不推送 | `... --dry-run` |
5. 🔗 返回仓库 URL
### 4. 反馈 常用选项:`--from-git DIR`(推仓库自身提交,不自动提交脏工作区)、`--owner`(默认 `NightStar`)、`--branch`(默认 `master`)、`--message`(自定义提交信息)、`--private`、`--exclude PATTERN`(可重复)、`--allow-leaks`、`--no-scan`、`--quiet`。
推送成功后告诉用户仓库地址。 脚本四步:隐私检查 → 确保仓库存在(无则建,有则复用并**拉远程历史对齐**,不用 `--force`)→ 准备/提交 → 推送(被拒自动 `fetch + rebase` 重试 3 次)。
如果脚本检测到疑似隐私数据泄露,**必须停下来询问用户是否继续**——不可自动跳过。 **退出码**:`0` 成功 / `1` 一般错误 / `2` 隐私检查未通过 / `3` 推送被拒。成功后 stdout 是 JSON(`url` / `clone` / `branch`)。
## 安全约束 ### 4. 回读远程验证(必做,别信脚本自报)
- ⚠️ 推送前必须过隐私检查 ⚠️ **脚本打印「✓ 推送成功」不等于远程真的收到了。** 实测踩过:脚本报成功,回读发现远程只有一个空 README,文件一个没上去(内部 `git add` 未执行的 bug)。凡涉及「推没推上去」的结论,一律回读远程:
- 仓库默认**公开**(public),不设 private
- 不要推送 .git 目录、node_modules、二进制文件 ```bash
- 如果之前泄露过隐私并已 amend,确认后再推 GIT_SSH_COMMAND="ssh -o StrictHostKeyChecking=accept-new" \
git clone --depth 1 git@git.yoresee.cc:<owner>/<repo>.git /tmp/<repo>-verify
find /tmp/<repo>-verify -type f -not -path "*/.git/*" | sed "s|/tmp/<repo>-verify/||" | sort
```
判据:期望的文件都在、内容对(抽读 1-2 个);`git log --oneline` 能看到本次提交。`--from-git` 模式再核提交数(`git rev-list --count HEAD`)与本地一致。**验完删掉临时 clone。**
### 5. 反馈
告诉用户仓库地址 + 验证结论(哪些文件在、提交 sha)。若脚本报隐私命中,**必须停下问用户**,不可自动跳过(用户确认无误才加 `--allow-leaks`)。
## 隐私检查:脚本已覆盖的 + 仍需手工补的
脚本扫 48 种文本扩展名(含 `.html/.htm/.css/.vue/.go/.tsx/.env/.svg/.scss` 等)与常见文件名(`Dockerfile/.env/.npmrc/...`),模式含 GitHub PAT、`sk-`/`sk-sp-`、AWS `AKIA`、Slack `xox`、飞书 `cli_`/`ou_`、`Bearer`、PEM 私钥、明文口令赋值,以及**公网 IP**(`127./10./192.168./172.16-31./169.254./0./255.` 这类不算,RFC 5737 文档专用段 `192.0.2./198.51.100./203.0.113.` 也不算)。改动或新增扫描范围前先看脚本里的 `SCAN_EXTS` / `SECRET_PATTERNS`,别凭记忆断言「扫不到」。
⚠️ **仍需手工看的一条:项目自己排除的文件,正是不能公开的。**
先看目标目录的 `.dockerignore` / `.gitignore`。实测某项目把 `data/keywords.src.json`(明文关键词表)排除出部署产物,却随仓库推成了**公开可见**——等于把解谜答案公开。同类信号:`*.src.json`、`*.key`、`*.secret.*`、被 ignore 的配置或原始数据。
发现就问用户三选一:改 private(`--private`,但建仓后改可见性要去 Gitea 页面/API)、`git rm --cached` 后加进 ignore、或明确接受公开。**不要默认推送。**
判读命中的邮箱**逐条看域名**:`.example` / `.invalid` 是拟真噪音,真实域名才算泄露。
## 建仓相关的两个坑
⚠️ **`tea` 0.15.0 给自己命名空间建仓不能传 `--owner`**:`tea repos create --name X --owner NightStar ...` 回 `Error: not found`,**省略 `--owner` 才成功**(大小写无关,实测小写同样失败)。脚本已内部规避(owner == 登录用户时省略该参数);手工用 `tea` 建仓时照此办理。只有建到组织/他人命名空间才需要 `--owner`。
⚠️ **全新仓库的 `Initial commit`**:脚本用 `tea ... --init` 建仓,远程因此多一个只含 README 的 `Initial commit`。脚本靠 `git reset --mixed` 对齐远程历史(只移动 HEAD、**不碰工作区**,避免覆盖同名文件)。若脚本的对齐/重试仍失败,手工兜底:
```bash
cd <目录> && git init -b master -c user.name=<名> -c user.email=<邮箱>
git remote add origin git@git.yoresee.cc:<owner>/<repo>.git
export GIT_SSH_COMMAND="ssh -o StrictHostKeyChecking=accept-new"
git fetch origin && git reset --mixed origin/master # 不要用 --hard:会覆盖同名文件
git add -A && git commit -m "<描述>" && git push -u origin master
```
别用 `--force`(会抹掉远程那个 commit)。
## 完成判据
远程回读通过:期望文件都在、内容正确、`git log` 有本次提交(`--from-git` 模式另核提交数与本地一致);仓库 URL 已告知用户;若曾出现隐私命中,用户已明确表态。
## 脚本依赖 ## 脚本依赖
`scripts/gitea_push.py` — Gitea 推送脚本,基于 tea CLI。 `/root/.openclaw/workspace/scripts/gitea_push.py` — 基于 `tea` CLI;**不在本 skill 目录内**,所以要先 `cd /root/.openclaw/workspace` 再用相对路径调它。
+521 -195
View File
@@ -1,263 +1,589 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""推送到 Gitea 仓库脚本 """推送到 Gitea 仓库 —— 通用 / 鲁棒版。
用法: 用法:
python3 gitea_push.py <repo_name> <description> <files_dir...> python3 gitea_push.py <repo_name> <description> [paths...] [options]
典型场景:
1. 新目录 → 新仓库
python3 gitea_push.py remind-me "飞书加急提醒" ./skills/remind-me/
2. 增量更新已有仓库(同样命令,自动拉远程历史后提交)
python3 gitea_push.py remind-me "飞书加急提醒" ./skills/remind-me/
3. 推送一个已有 git 仓库(保留完整提交历史)
python3 gitea_push.py my-proj "我的项目" --from-git /root/my-proj
4. 只检查不推送
python3 gitea_push.py my-proj "我的项目" ./src --dry-run
环境变量: 环境变量:
GITEA_URL Gitea 实例地址(默认 https://git.yoresee.cc) GITEA_URL 实例地址(默认 https://git.yoresee.cc)
GITEA_USER Gitea 用户名(默认 NightStar) GITEA_USER 用户名(默认 NightStar)
GITEA_EMAIL Git 提交邮箱(默认 nightstar@yoresee.cc) GITEA_EMAIL 提交邮箱(默认 nightstar@yoresee.cc)
GITEA_SSH_HOST SSH 主机(默认 git.yoresee.cc)
示例: 退出码: 0 成功 / 1 一般错误 / 2 隐私检查未通过 / 3 推送被拒(重试后仍失败)
python3 gitea_push.py remind-me "飞书加急提醒" ./skills/remind-me/
python3 gitea_push.py my-tool "一个工具" ./src/ scripts/
""" """
from __future__ import annotations
import argparse
import fnmatch
import json import json
import sys
import subprocess
import os import os
import re import re
import shutil
import subprocess
import sys
import tempfile import tempfile
from pathlib import Path from pathlib import Path
GITEA_URL = os.environ.get("GITEA_URL", "https://git.yoresee.cc") # ─────────────────────────── 配置 ───────────────────────────
GITEA_URL = os.environ.get("GITEA_URL", "https://git.yoresee.cc").rstrip("/")
GITEA_USER = os.environ.get("GITEA_USER", "NightStar") GITEA_USER = os.environ.get("GITEA_USER", "NightStar")
GITEA_EMAIL = os.environ.get("GITEA_EMAIL", "nightstar@yoresee.cc") GITEA_EMAIL = os.environ.get("GITEA_EMAIL", "nightstar@yoresee.cc")
GITEA_SSH_HOST = os.environ.get("GITEA_SSH_HOST", "git.yoresee.cc")
DEFAULT_BRANCH = "master" DEFAULT_BRANCH = "master"
GIT_HOST = "git.yoresee.cc" PUSH_RETRIES = 3
SANITIZE_PATTERNS = [ # 隐私模式:命中即拦(--allow-leaks 可放行)
(r'ou_[a-f0-9]{20,}', 'ou_xxxxxx'), # 飞书 open_id SECRET_PATTERNS = [
(r'sk-[a-zA-Z0-9]{20,}', 'sk-xxxxxx'), # API key (r"ghp_[A-Za-z0-9]{20,}", "GitHub PAT (ghp_)"),
(r't-[a-zA-Z0-9]{20,}', 't-xxxxxx'), # token (r"github_pat_[A-Za-z0-9_]{20,}", "GitHub fine-grained PAT"),
(r'cli_[a-zA-Z0-9]{10,}', 'cli_xxxxxx'), # 飞书 app_id (r"sk-[A-Za-z0-9]{20,}", "API key (sk-)"),
(r'Bearer [a-zA-Z0-9\-_\.]{20,}', 'Bearer xxxxxx'), (r"sk-sp-[A-Za-z0-9\-_]{20,}", "Token Plan key (sk-sp-)"),
(r'Authorization: [a-zA-Z0-9\-_\.]{20,}', 'Authorization: xxxxxx'), (r"AKIA[0-9A-Z]{16}", "AWS Access Key ID"),
(r'[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}', 'x.x.x.x'), # IP (r"xox[baprs]-[A-Za-z0-9\-]{10,}", "Slack token"),
(r"cli_[a-zA-Z0-9]{10,}", "飞书 app_id (cli_)"),
(r"ou_[a-f0-9]{20,}", "飞书 open_id (ou_)"),
(r"t-[a-zA-Z0-9]{20,}", "token (t-)"),
(r"Bearer\s+[A-Za-z0-9\-_.]{20,}", "Bearer 凭据"),
(r"-----BEGIN [A-Z ]*PRIVATE KEY-----", "私钥文件"),
(r"(?i)\b(password|passwd|secret|api[_-]?key|access[_-]?token)\s*[:=]\s*[\"']?[^\s\"',{}]{12,}", "明文口令/密钥赋值"),
] ]
ALWAYS_EXCLUDE = {".git", "node_modules", "__pycache__", ".DS_Store", "*.pyc", "*.pyo"} # IP:仅拦公网地址(保留 127./10./192.168./172.16-31./0.0.0.0/255. 等常见非敏感写法)
IP_RE = re.compile(r"\b(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})\b")
# 扫描的文本类扩展名(旧版漏了 .html/.css/.vue 等,这里补齐)
SCAN_EXTS = {
".py", ".pyi", ".md", ".markdown", ".json", ".json5", ".yaml", ".yml",
".sh", ".bash", ".zsh", ".fish", ".txt", ".toml", ".ini", ".cfg", ".conf", ".properties",
".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx", ".vue", ".svelte",
".html", ".htm", ".css", ".scss", ".less", ".xml", ".svg",
".go", ".rs", ".java", ".kt", ".rb", ".php", ".sql", ".pl", ".lua",
".env", ".envrc", ".tf", ".tfvars", ".gradle", ".cmake",
}
SCAN_NAMES = {
"Dockerfile", "Makefile", "Rakefile", "Gemfile", "Procfile",
".env", ".env.local", ".env.production", ".npmrc", ".pypirc", ".netrc",
}
ALWAYS_EXCLUDE_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", ".mypy_cache",
".pytest_cache", ".ruff_cache", "dist", "build", ".next", ".nuxt",
"target", ".idea", ".vscode", "vendor"}
ALWAYS_EXCLUDE_FILES = {".DS_Store", "Thumbs.db"}
MAX_WARN_BYTES = 5 * 1024 * 1024 # 单文件超 5MB 提醒
def run_tea(*args): class PushError(Exception):
cmd = ["tea"] + list(args) def __init__(self, msg: str, code: int = 1):
result = subprocess.run(cmd, capture_output=True, text=True) super().__init__(msg)
if result.returncode != 0: self.code = code
raise RuntimeError(f"tea 命令失败: {' '.join(cmd)}\n{result.stderr}")
return result.stdout
def load_gitignore_rules(dirs): # ─────────────────────────── 日志 ───────────────────────────
"""读取所有相关目录下的 .gitignore 并返回 pathspec"""
patterns = [] QUIET = False
seen = set()
for d in dirs:
p = Path(d).resolve() def log(msg: str = "", *, err: bool = True) -> None:
while True: if not QUIET:
gitignore = p / ".gitignore" print(msg, file=sys.stderr if err else sys.stdout)
if str(gitignore) not in seen and gitignore.exists():
seen.add(str(gitignore))
def step(n: int, total: int, msg: str) -> None:
log(f"[{n}/{total}] {msg}")
# ─────────────────────────── 子进程 ───────────────────────────
def _ssh_env() -> dict:
env = dict(os.environ)
env["GIT_SSH_COMMAND"] = env.get(
"GIT_SSH_COMMAND", "ssh -o StrictHostKeyChecking=accept-new"
)
return env
def run(cmd: list[str], cwd: Path | None = None, *, check: bool = True,
env: dict | None = None) -> subprocess.CompletedProcess:
res = subprocess.run(cmd, cwd=str(cwd) if cwd else None, capture_output=True,
text=True, env=env)
if check and res.returncode != 0:
raise PushError(f"命令失败: {' '.join(cmd)}\n{res.stderr.strip()}")
return res
def git(args: list[str], cwd: Path, *, check: bool = True) -> subprocess.CompletedProcess:
return run(["git", *args], cwd=cwd, check=check, env=_ssh_env())
def tea(args: list[str], *, check: bool = True) -> subprocess.CompletedProcess:
return run(["tea", *args], check=check)
def has_commits(cwd: Path) -> bool:
return git(["rev-parse", "--verify", "HEAD"], cwd, check=False).returncode == 0
def remote_branch_exists(cwd: Path, branch: str) -> bool:
return git(["rev-parse", "--verify", f"origin/{branch}"], cwd,
check=False).returncode == 0
# ─────────────────────────── 隐私检查 ───────────────────────────
def _is_scannable(p: Path) -> bool:
if p.name in SCAN_NAMES:
return True
return p.suffix.lower() in SCAN_EXTS
# 跳过不算敏感的四类:本机/私网/保留段、以及 RFC 5737 文档专用网段
_DOC_RANGES = ((192, 0, 2), (198, 51, 100), (203, 0, 113))
def _public_ip(match: re.Match) -> bool:
"""判定是否公网 IP(跳过本机/私网/环回/广播/文档专用段等无害写法)。"""
a, b, c = int(match.group(1)), int(match.group(2)), int(match.group(3))
if a == 0 or a == 127 or a == 255:
return False
if a == 10 or (a == 192 and b == 168) or (a == 172 and 16 <= b <= 31):
return False
if a == 169 and b == 254:
return False
if (a, b, c) in _DOC_RANGES:
return False
# 其余一律报出,交人工判断(版本号式写法也宁可多报)
return True
def scan_paths(paths: list[Path]) -> list[dict]:
"""扫描文件中的疑似隐私。返回命中列表。"""
findings: list[dict] = []
for root in paths:
files = [root] if root.is_file() else [f for f in root.rglob("*") if f.is_file()]
for f in files:
if any(part in ALWAYS_EXCLUDE_DIRS for part in f.parts):
continue
if not _is_scannable(f):
continue
try: try:
with open(gitignore) as f: text = f.read_text(encoding="utf-8", errors="ignore")
for line in f: except OSError:
continue
hits: list[str] = []
for pattern, label in SECRET_PATTERNS:
n = len(re.findall(pattern, text))
if n:
hits.append(f"{label} × {n}")
pub_ips = [m.group(0) for m in IP_RE.finditer(text) if _public_ip(m)]
if pub_ips:
uniq = sorted(set(pub_ips))
hits.append(f"公网 IP × {len(uniq)}: {', '.join(uniq[:3])}"
+ (" …" if len(uniq) > 3 else ""))
if hits:
findings.append({"file": str(f), "hits": hits})
return findings
# ─────────────────────────── 文件收集 ───────────────────────────
def _fnmatch_fallback(patterns: list[str], rel: str) -> bool:
"""pathspec 不可用时的兜底匹配(够用即可)。"""
for pat in patterns:
pat = pat.rstrip("/")
if fnmatch.fnmatch(rel, pat) or fnmatch.fnmatch(rel, f"{pat}/*"):
return True
if pat.startswith("**/") and fnmatch.fnmatch(rel, pat[3:]):
return True
return False
def load_ignore_spec(sources: list[Path]):
"""收集各源目录向上查找的 .gitignore 规则。"""
patterns: list[str] = []
seen: set[str] = set()
for src in sources:
p = src.resolve()
if p.is_file():
p = p.parent
while True:
gi = p / ".gitignore"
if str(gi) not in seen and gi.is_file():
seen.add(str(gi))
try:
for line in gi.read_text(encoding="utf-8", errors="ignore").splitlines():
line = line.strip() line = line.strip()
if line and not line.startswith("#"): if line and not line.startswith("#") and line != ".gitignore":
patterns.append(line) patterns.append(line)
except Exception: except OSError:
pass pass
if p.parent == p: if p.parent == p:
break break
p = p.parent p = p.parent
try: try:
import pathspec import pathspec # type: ignore
return pathspec.PathSpec.from_lines("gitwildmatch", patterns) return pathspec.PathSpec.from_lines("gitwildmatch", patterns)
except ImportError: except ImportError:
# fallback: 简单 glob 匹配 log(" (pathspec 未安装,使用 fnmatch 兜底匹配)")
print(" (pathspec 未安装,使用简单 .gitignore 匹配)", file=sys.stderr) return patterns
patterns_set = set(patterns)
return patterns_set, None
def is_ignored(filepath, spec, src_root): def is_ignored(spec, rel: str) -> bool:
"""判断文件是否应被忽略""" if isinstance(spec, list):
name = filepath.name return _fnmatch_fallback(spec, rel)
# 内置排除
if name in ALWAYS_EXCLUDE:
return True
if name.endswith(".pyc") or name.endswith(".pyo"):
return True
if name == ".gitignore":
return True
# .gitignore 规则
try: try:
rel = filepath.relative_to(src_root) return spec.match_file(rel)
rel_str = str(rel)
except ValueError:
return False
if spec is None:
return False
try:
return spec.match_file(rel_str)
except Exception: except Exception:
return False return False
def sanitize_file(path): def collect_files(sources: list[Path], work_dir: Path, spec, extra_excludes: list[str]) -> int:
"""检查文件中的隐私信息""" """把源文件复制进工作区(保持相对结构)。返回文件数。"""
try:
content = Path(path).read_text()
except Exception:
return False
issues = []
for pattern, replacement in SANITIZE_PATTERNS:
matches = re.findall(pattern, content)
if matches:
issues.append(f" {pattern}: 发现 {len(matches)} 处疑似隐私数据")
if issues:
print(f"\n⚠️ 隐私检查 - {path}:", file=sys.stderr)
for issue in issues:
print(issue, file=sys.stderr)
print("", file=sys.stderr)
return True
return False
def collect_files(dirs, work_dir):
"""收集需要推送的文件,应用 .gitignore 过滤"""
print("[3/5] 准备推送内容(应用 .gitignore 过滤)...", file=sys.stderr)
spec = load_gitignore_rules(dirs)
count = 0 count = 0
for src in sources:
for d in dirs: src = src.resolve()
src = Path(d).resolve()
if src.is_file(): if src.is_file():
dest = work_dir / src.name (work_dir / src.name).write_bytes(src.read_bytes())
dest.write_bytes(src.read_bytes())
print(f" + {src.name}", file=sys.stderr)
count += 1 count += 1
elif src.is_dir(): continue
if not src.is_dir():
log(f" ⚠️ 跳过不存在的路径: {src}")
continue
for f in sorted(src.rglob("*")): for f in sorted(src.rglob("*")):
if not f.is_file(): if not f.is_file():
continue continue
if ".git" in f.parts: if any(part in ALWAYS_EXCLUDE_DIRS for part in f.parts):
continue continue
if is_ignored(f, spec, src): if f.name in ALWAYS_EXCLUDE_FILES or f.name.endswith((".pyc", ".pyo")):
continue
rel = str(f.relative_to(src))
if is_ignored(spec, rel):
continue
if any(fnmatch.fnmatch(rel, pat) for pat in extra_excludes):
continue continue
rel = f.relative_to(src)
dest = work_dir / rel dest = work_dir / rel
dest.parent.mkdir(parents=True, exist_ok=True) dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_bytes(f.read_bytes()) shutil.copy2(f, dest)
print(f" + {rel}", file=sys.stderr) size = f.stat().st_size
if size > MAX_WARN_BYTES:
log(f" ⚠️ 大文件 {rel} ({size // 1024 // 1024} MB)")
count += 1 count += 1
return count
print(f" 共收集 {count} 个文件", file=sys.stderr)
def main(): # ─────────────────────────── 仓库准备 ───────────────────────────
if len(sys.argv) < 3:
print("用法: gitea_push.py <repo_name> <description> <files_dir...>", file=sys.stderr)
print("示例: gitea_push.py my-tool '一个工具' ./src/", file=sys.stderr)
sys.exit(1)
repo_name = sys.argv[1] def login_user() -> str:
description = sys.argv[2] """取 tea 当前登录用户名(失败回退到 GITEA_USER)。"""
dirs = [d for d in sys.argv[3:] if d.strip()] res = tea(["login", "list", "--output", "json"], check=False)
if res.returncode == 0:
try:
entries = json.loads(res.stdout or "[]")
for e in entries:
if e.get("user"):
return str(e["user"])
except (json.JSONDecodeError, TypeError):
pass
return GITEA_USER
if not dirs:
print("❌ 请至少指定一个文件或目录", file=sys.stderr) def ensure_repo(repo: str, description: str, owner: str, branch: str,
sys.exit(1) private: bool, dry_run: bool) -> bool:
"""确保仓库存在。返回是否新建。
⚠️ tea 0.15.0 的坑:给自己命名空间建仓时传 `--owner <自己>` 会报
`Error: not found`;必须省略 --owner。只有 owner ≠ 登录用户(建到
组织/他人命名空间)时才传。
"""
if dry_run:
log(" (dry-run:跳过建仓)")
return False
existed = False
self_owner = login_user().lower()
cmd = ["repos", "create", "--name", repo, "--description", description,
"--branch", branch, "--init"]
if owner.lower() != self_owner:
cmd += ["--owner", owner]
if private:
cmd.append("--private")
res = tea(cmd, check=False)
if res.returncode == 0:
out = (res.stdout or "") + (res.stderr or "")
if "not found" in out.lower():
raise PushError(
f"创建仓库返回 not found(owner={owner} 可能不存在或无权访问):\n{out.strip()[:300]}")
return True
err = (res.stderr + res.stdout).lower()
if "already exists" in err or "409" in err:
existed = True
else:
# 二次确认:列一下 owner 的仓库
listing = tea(["repos", "list", "--owner", owner, "--output", "json",
"--limit", "200"], check=False)
if listing.returncode == 0:
try:
names = {r.get("name") for r in json.loads(listing.stdout or "[]")}
if repo in names:
existed = True
except json.JSONDecodeError:
pass
if not existed:
raise PushError(f"创建仓库失败: {res.stderr.strip() or res.stdout.strip()}")
log(f" 仓库 {owner}/{repo} 已存在,走增量提交(不 force)")
return False
def prepare_worktree(sources: list[Path], repo: str, owner: str, branch: str,
from_git: Path | None, extra_excludes: list[str]) -> tuple[Path, bool]:
"""准备工作区。返回 (路径, 是否为临时目录)。"""
if from_git:
work = from_git.resolve()
if not (work / ".git").exists():
raise PushError(f"{work} 不是 git 仓库(--from-git 需要已有 .git)")
remotes = git(["remote"], work, check=False).stdout.split()
url = f"git@{GITEA_SSH_HOST}:{owner}/{repo}.git"
if "origin" in remotes:
git(["remote", "set-url", "origin", url], work)
else:
git(["remote", "add", "origin", url], work)
return work, False
work = Path(tempfile.mkdtemp(prefix=f"gitea_{repo}_"))
run(["git", "init", "-b", branch], work)
git(["config", "user.name", GITEA_USER], work)
git(["config", "user.email", GITEA_EMAIL], work)
git(["remote", "add", "origin", f"git@{GITEA_SSH_HOST}:{owner}/{repo}.git"], work)
spec = load_ignore_spec(sources)
n = collect_files(sources, work, spec, extra_excludes)
log(f" 共收集 {n} 个文件")
return work, True
def align_with_remote(work: Path, branch: str, *, from_git: bool) -> str:
"""把本地分支对齐到远程历史,避免 'fetch first' 拒绝。
非 --from-git 场景用 `reset --mixed`(而非 --hard):只移动 HEAD、
不碰工作区,避免覆盖同名文件的用户内容。
"""
git(["fetch", "origin", branch], work, check=False)
if not remote_branch_exists(work, branch):
return "no-remote-branch"
if not has_commits(work):
# 未出生分支:继承远程历史,但保留工作区文件(--mixed 不碰文件)
git(["reset", "--mixed", f"origin/{branch}"], work)
return "adopted-remote"
if from_git:
res = git(["rebase", f"origin/{branch}"], work, check=False)
if res.returncode != 0:
git(["rebase", "--abort"], work, check=False)
raise PushError(
"rebase 冲突,请手动解决后重试:\n" + res.stderr.strip()[:500], code=3)
return "rebased"
return "kept-local"
# ─────────────────────────── 提交 & 推送 ───────────────────────────
def commit_if_changed(work: Path, message: str, *, add_all: bool = True) -> bool:
if add_all:
git(["add", "-A"], work)
if git(["diff", "--cached", "--quiet"], work, check=False).returncode == 0:
log(" 无内容变更,跳过 commit")
return False
res = git(["commit", "-m", message], work, check=False)
if res.returncode != 0:
raise PushError(f"commit 失败:\n{res.stderr.strip() or res.stdout.strip()}")
log(f" ✓ {git(['log', '--oneline', '-1'], work).stdout.strip()}")
return True
def push_with_retry(work: Path, branch: str) -> None:
last = ""
for attempt in range(1, PUSH_RETRIES + 1):
res = git(["push", "-u", "origin", branch], work, check=False)
if res.returncode == 0:
log(" ✓ 推送成功")
return
err = res.stderr.strip()
last = err
if any(k in err.lower() for k in ("rejected", "fetch first", "non-fast-forward", "behind")):
log(f" 推送被拒(第 {attempt} 次),拉取远程后重试…")
git(["fetch", "origin", branch], work, check=False)
if remote_branch_exists(work, branch):
rb = git(["rebase", f"origin/{branch}"], work, check=False)
if rb.returncode != 0:
git(["rebase", "--abort"], work, check=False)
raise PushError("rebase 冲突,无法自动重试:\n" + rb.stderr.strip()[:500],
code=3)
continue
raise PushError(f"推送失败:\n{err[:500]}", code=3)
raise PushError(f"推送失败(已重试 {PUSH_RETRIES} 次):\n{last[:500]}", code=3)
# ─────────────────────────── 主流程 ───────────────────────────
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="gitea_push.py",
description="推送到 Gitea 仓库(通用鲁棒版)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""示例:
gitea_push.py remind-me "飞书加急提醒" ./skills/remind-me/
gitea_push.py my-proj "我的项目" --from-git /root/my-proj
gitea_push.py my-proj "我的项目" ./src --dry-run
""")
p.add_argument("repo_name", help="仓库名(kebab-case)")
p.add_argument("description", help="仓库描述 / 提交信息")
p.add_argument("paths", nargs="*", help="要推送的文件或目录(--from-git 时可省)")
p.add_argument("--from-git", metavar="DIR", default=None,
help="推送一个已有 git 仓库(保留提交历史)")
p.add_argument("--owner", default=GITEA_USER, help=f"仓库所有者(默认 {GITEA_USER})")
p.add_argument("--branch", default=DEFAULT_BRANCH, help=f"分支(默认 {DEFAULT_BRANCH})")
p.add_argument("--message", default=None, help="自定义提交信息(默认用 description)")
p.add_argument("--private", action="store_true", help="建私有仓库(默认公开)")
p.add_argument("--exclude", action="append", default=[], metavar="PATTERN",
help="额外排除(glob,可重复)")
p.add_argument("--allow-leaks", action="store_true", help="隐私检查命中时仍继续")
p.add_argument("--no-scan", action="store_true", help="跳过隐私检查")
p.add_argument("--dry-run", action="store_true", help="只检查与暂存,不建仓不推送")
p.add_argument("--quiet", action="store_true", help="静默(只输出结果 JSON)")
return p
def main(argv: list[str] | None = None) -> int:
global QUIET
args = build_parser().parse_args(argv)
QUIET = args.quiet
branch = args.branch
message = args.message or f"feat: {args.description}"
dry = args.dry_run
total_steps = 4
try:
if shutil.which("git") is None:
raise PushError("找不到 git")
if shutil.which("tea") is None and not dry:
raise PushError("找不到 tea CLI(Gitea 命令行)")
# 源路径
if args.from_git:
sources = [Path(args.from_git)]
else:
if not args.paths:
raise PushError("请至少指定一个路径,或用 --from-git 指定仓库目录")
sources = [Path(p) for p in args.paths]
missing = [str(s) for s in sources if not s.exists()]
if missing:
raise PushError("路径不存在: " + ", ".join(missing))
# 1. 隐私检查 # 1. 隐私检查
print("[1/5] 隐私检查...", file=sys.stderr) step(1, total_steps, "隐私检查…")
has_issues = False if args.no_scan:
for d in dirs: log(" (已跳过)")
p = Path(d)
if p.is_file():
has_issues |= sanitize_file(p)
elif p.is_dir():
for f in p.rglob("*"):
if f.is_file() and f.suffix in ('.py', '.md', '.json', '.yaml', '.yml', '.sh', '.txt', '.toml', '.js', '.ts'):
has_issues |= sanitize_file(f)
if has_issues:
resp = input("⚠️ 发现疑似隐私数据,继续推送?(yes/no): ").strip().lower()
if resp not in ('yes', 'y'):
print("已取消推送", file=sys.stderr)
sys.exit(1)
# 2. 创建 Gitea 仓库(默认公开,默认分支 master)
print(f"[2/5] 创建公开仓库 {repo_name}(分支: {DEFAULT_BRANCH})...", file=sys.stderr)
try:
output = run_tea("repo", "create", "--name", repo_name,
"--description", description,
"--branch", DEFAULT_BRANCH,
"--init")
print(output.strip(), file=sys.stderr)
except RuntimeError as e:
if "already exists" in str(e).lower() or "409" in str(e):
print(f" 仓库 {repo_name} 已存在,使用现有仓库", file=sys.stderr)
else: else:
raise if args.from_git:
scan_targets = [p for p in Path(args.from_git).iterdir()
if p.name != ".git"]
else:
scan_targets = sources
findings = scan_paths(scan_targets)
if findings:
log("")
for item in findings:
log(f" ⚠️ {item['file']}")
for h in item["hits"]:
log(f" - {h}")
log("")
if not args.allow_leaks:
raise PushError(
f"发现 {len(findings)} 个文件含疑似隐私,已中止。"
"确认无误可加 --allow-leaks 继续。", code=2)
log(" --allow-leaks 已指定,继续推送")
else:
log(" ✓ 未发现敏感内容")
# 3. 收集文件(使用 .gitignore) # 2. 仓库准备
work_dir = Path(tempfile.mkdtemp(prefix="gitea_")) step(2, total_steps, f"确保仓库 {args.owner}/{args.repo_name}…")
collect_files(dirs, work_dir) is_new = ensure_repo(args.repo_name, args.description, args.owner,
branch, args.private, dry)
log(" ✓ 新建仓库" if is_new else " ✓ 复用现有仓库")
# 4. Git 初始化并推送到 master 分支 # 3. 工作区 + 对齐远程
print(f"[4/5] 推送到 Gitea (分支: {DEFAULT_BRANCH})...", file=sys.stderr) step(3, total_steps, "准备工作区并对齐远程历史…")
subprocess.run(["git", "init", "-b", DEFAULT_BRANCH], cwd=work_dir, capture_output=True) work, is_temp = prepare_worktree(sources, args.repo_name, args.owner,
subprocess.run(["git", "config", "user.name", GITEA_USER], cwd=work_dir, capture_output=True) branch, Path(args.from_git) if args.from_git else None,
subprocess.run(["git", "config", "user.email", GITEA_EMAIL], cwd=work_dir, capture_output=True) args.exclude)
subprocess.run(["git", "add", "-A"], cwd=work_dir, capture_output=True) if not args.from_git:
git(["config", "user.name", GITEA_USER], work)
git(["config", "user.email", GITEA_EMAIL], work)
mode = align_with_remote(work, branch, from_git=bool(args.from_git))
log(f" 远程对齐方式: {mode}")
commit_res = subprocess.run( if args.from_git:
["git", "commit", "-m", f"feat: {description}"], # --from-git:推送仓库已有的提交,不自动提交未完成的工作区改动
cwd=work_dir, capture_output=True, text=True dirty = git(["status", "--porcelain"], work, check=False).stdout.strip()
) if dirty:
if commit_res.returncode != 0: log(" ⚠️ 工作区有未提交改动,不会被推送(请先自行 commit)")
# 可能是空提交(所有文件都被过滤了) changed = False
print(f" commit 可能为空: {commit_res.stderr.strip()}", file=sys.stderr) else:
# 普通目录推送:全部纳入暂存区(含新增/修改/删除)
changed = commit_if_changed(work, message, add_all=True)
repo_url = f"git@{GIT_HOST}:{GITEA_USER}/{repo_name}.git" # 4. 推送
remote_res = subprocess.run( if dry:
["git", "remote", "add", "origin", repo_url], step(4, total_steps, "dry-run:跳过推送")
cwd=work_dir, capture_output=True, text=True log(f" 工作区: {work}")
) else:
if remote_res.returncode != 0: step(4, total_steps, f"推送到 {args.owner}/{args.repo_name} ({branch})…")
subprocess.run(["git", "remote", "set-url", "origin", repo_url], if args.from_git:
cwd=work_dir, capture_output=True) ahead = git(["rev-list", "--count", f"origin/{branch}..HEAD"],
work, check=False).stdout.strip()
push_result = subprocess.run( if ahead in ("", "0"):
["git", "push", "-u", "origin", DEFAULT_BRANCH, "--force"], log(" 本地无新提交,远程已是最新")
cwd=work_dir, else:
capture_output=True, text=True, log(f" 本地领先远程 {ahead} 个提交")
env={**os.environ, "GIT_SSH_COMMAND": "ssh -o StrictHostKeyChecking=accept-new"} push_with_retry(work, branch)
) else:
if push_result.returncode != 0: push_with_retry(work, branch)
print(f"❌ 推送失败:\n{push_result.stderr[:500]}", file=sys.stderr)
sys.exit(1)
# 5. 完成
browse_url = f"{GITEA_URL}/{GITEA_USER}/{repo_name}"
clone_url = f"git@{GIT_HOST}:{GITEA_USER}/{repo_name}.git"
print(f"\n[5/5] ✅ 推送成功!", file=sys.stderr)
print(f" 浏览: {browse_url}", file=sys.stderr)
print(f" 克隆: {clone_url}", file=sys.stderr)
browse = f"{GITEA_URL}/{args.owner}/{args.repo_name}"
log("")
log(f"✅ 完成!浏览: {browse}")
print(json.dumps({ print(json.dumps({
"ok": True, "ok": True,
"repo": repo_name, "repo": args.repo_name,
"url": browse_url, "owner": args.owner,
"clone": clone_url, "url": browse,
"branch": DEFAULT_BRANCH "clone": f"git@{GITEA_SSH_HOST}:{args.owner}/{args.repo_name}.git",
})) "branch": branch,
"new_repo": is_new,
"dry_run": dry,
"work_dir": str(work) if is_temp else None,
"from_git": bool(args.from_git),
}, ensure_ascii=False))
return 0
except PushError as exc:
log(f"\n❌ {exc}")
print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False))
return exc.code
except KeyboardInterrupt:
log("\n已中断")
return 1
if __name__ == "__main__": if __name__ == "__main__":
main() sys.exit(main())