diff --git a/SKILL.md b/SKILL.md index 72a3182..76ebb99 100644 --- a/SKILL.md +++ b/SKILL.md @@ -1,62 +1,97 @@ --- name: "publish-to-gitea" -description: "将文件/目录发布到 Gitea 仓库 — 自动隐私检查、创建公开仓库、Git 推送" +description: "推到 gitea/发布到仓库/建 repo 推上去/把某个 git 仓库推上去时用:跑 gitea_push.py(隐私检查 → 建仓 → 对齐远程 → 推送)→ **回读远程验证**。含脚本覆盖不到的隐私盲区与 tea 建仓坑。" --- # Publish to Gitea — 发布到 Gitea 仓库 -将文件/目录推送到 Gitea 仓库(git.yoresee.cc),自动隐私检查、创建公开仓库、一键推送。 +把文件/目录或一个已有 git 仓库推送到 Gitea(git.yoresee.cc)。 + +**不属于本 skill**:在本机**正在开发的仓库里**提交并推分支 → 直接在该仓库 `git push `(凭据走已配好的 `!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. 确定要推送的内容 -从上下文中推断要推送的文件/目录。常见场景: -- 用户刚创建/修改了一个 skill → 推送整个 skill 目录 -- 用户说"把这个脚本推上去" → 推送脚本文件 -- 用户指定了具体路径 → 直接用 +- 用户刚写的 skill / 脚本 / 项目目录 → 推那个目录 +- 用户指定路径 → 直接用 +- 内容是一个**已有 git 仓库**(有 `.git`、提交历史要保)→ 用 `--from-git`(见 §3) ### 2. 确定仓库名和描述 -- **仓库名**:从内容推断,kebab-case 格式 -- **描述**:一句话说明用途 -- 用户没明确说则主动确认 +仓库名 kebab-case,描述一句话。用户没明说则主动确认(仓库默认**公开**)。 ### 3. 执行推送 ```bash -cd /root/.openclaw/workspace && python3 scripts/gitea_push.py "" +cd /root/.openclaw/workspace +python3 scripts/gitea_push.py "" [paths...] ``` -脚本自动完成: -1. 🔍 隐私检查(检测 open_id、API key、token、IP 等敏感模式) -2. 📦 创建 Gitea **公开**仓库(已存在则复用现有仓库) -3. 📄 收集文件(排除 .git、node_modules 等) -4. 🚀 Git 增量提交 → push(**已存在仓库先拉取远程历史,正常 commit + push,不使用 --force**;全新仓库直接首推) -5. 🔗 返回仓库 URL +| 场景 | 命令 | +|---|---| +| 目录/文件 → 仓库 | `gitea_push.py my-proj "描述" ./src` | +| 推已有 git 仓库(**保留完整提交历史**) | `gitea_push.py my-proj "描述" --from-git /root/my-proj` | +| 只检查不推送 | `... --dry-run` | -### 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. 回读远程验证(必做,别信脚本自报) -- ⚠️ 推送前必须过隐私检查 -- 仓库默认**公开**(public),不设 private -- 不要推送 .git 目录、node_modules、二进制文件 -- 如果之前泄露过隐私并已 amend,确认后再推 +⚠️ **脚本打印「✓ 推送成功」不等于远程真的收到了。** 实测踩过:脚本报成功,回读发现远程只有一个空 README,文件一个没上去(内部 `git add` 未执行的 bug)。凡涉及「推没推上去」的结论,一律回读远程: + +```bash +GIT_SSH_COMMAND="ssh -o StrictHostKeyChecking=accept-new" \ + git clone --depth 1 git@git.yoresee.cc:/.git /tmp/-verify +find /tmp/-verify -type f -not -path "*/.git/*" | sed "s|/tmp/-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:/.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` 再用相对路径调它。 diff --git a/gitea_push.py b/gitea_push.py index 0f691e3..5271e29 100644 --- a/gitea_push.py +++ b/gitea_push.py @@ -1,263 +1,589 @@ #!/usr/bin/env python3 -"""推送到 Gitea 仓库脚本 +"""推送到 Gitea 仓库 —— 通用 / 鲁棒版。 + 用法: - python3 gitea_push.py + python3 gitea_push.py [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_USER Gitea 用户名(默认 NightStar) - GITEA_EMAIL Git 提交邮箱(默认 nightstar@yoresee.cc) + GITEA_URL 实例地址(默认 https://git.yoresee.cc) + GITEA_USER 用户名(默认 NightStar) + GITEA_EMAIL 提交邮箱(默认 nightstar@yoresee.cc) + GITEA_SSH_HOST SSH 主机(默认 git.yoresee.cc) -示例: - python3 gitea_push.py remind-me "飞书加急提醒" ./skills/remind-me/ - python3 gitea_push.py my-tool "一个工具" ./src/ scripts/ +退出码: 0 成功 / 1 一般错误 / 2 隐私检查未通过 / 3 推送被拒(重试后仍失败) """ +from __future__ import annotations + +import argparse +import fnmatch import json -import sys -import subprocess import os import re +import shutil +import subprocess +import sys import tempfile 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_EMAIL = os.environ.get("GITEA_EMAIL", "nightstar@yoresee.cc") +GITEA_SSH_HOST = os.environ.get("GITEA_SSH_HOST", "git.yoresee.cc") DEFAULT_BRANCH = "master" -GIT_HOST = "git.yoresee.cc" +PUSH_RETRIES = 3 -SANITIZE_PATTERNS = [ - (r'ou_[a-f0-9]{20,}', 'ou_xxxxxx'), # 飞书 open_id - (r'sk-[a-zA-Z0-9]{20,}', 'sk-xxxxxx'), # API key - (r't-[a-zA-Z0-9]{20,}', 't-xxxxxx'), # token - (r'cli_[a-zA-Z0-9]{10,}', 'cli_xxxxxx'), # 飞书 app_id - (r'Bearer [a-zA-Z0-9\-_\.]{20,}', 'Bearer xxxxxx'), - (r'Authorization: [a-zA-Z0-9\-_\.]{20,}', 'Authorization: xxxxxx'), - (r'[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}', 'x.x.x.x'), # IP +# 隐私模式:命中即拦(--allow-leaks 可放行) +SECRET_PATTERNS = [ + (r"ghp_[A-Za-z0-9]{20,}", "GitHub PAT (ghp_)"), + (r"github_pat_[A-Za-z0-9_]{20,}", "GitHub fine-grained PAT"), + (r"sk-[A-Za-z0-9]{20,}", "API key (sk-)"), + (r"sk-sp-[A-Za-z0-9\-_]{20,}", "Token Plan key (sk-sp-)"), + (r"AKIA[0-9A-Z]{16}", "AWS Access Key ID"), + (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): - cmd = ["tea"] + list(args) - result = subprocess.run(cmd, capture_output=True, text=True) - if result.returncode != 0: - raise RuntimeError(f"tea 命令失败: {' '.join(cmd)}\n{result.stderr}") - return result.stdout +class PushError(Exception): + def __init__(self, msg: str, code: int = 1): + super().__init__(msg) + self.code = code -def load_gitignore_rules(dirs): - """读取所有相关目录下的 .gitignore 并返回 pathspec""" - patterns = [] - seen = set() - for d in dirs: - p = Path(d).resolve() +# ─────────────────────────── 日志 ─────────────────────────── + +QUIET = False + + +def log(msg: str = "", *, err: bool = True) -> None: + if not QUIET: + print(msg, file=sys.stderr if err else sys.stdout) + + +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: + text = f.read_text(encoding="utf-8", errors="ignore") + 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: - gitignore = p / ".gitignore" - if str(gitignore) not in seen and gitignore.exists(): - seen.add(str(gitignore)) + gi = p / ".gitignore" + if str(gi) not in seen and gi.is_file(): + seen.add(str(gi)) try: - with open(gitignore) as f: - for line in f: - line = line.strip() - if line and not line.startswith("#"): - patterns.append(line) - except Exception: + for line in gi.read_text(encoding="utf-8", errors="ignore").splitlines(): + line = line.strip() + if line and not line.startswith("#") and line != ".gitignore": + patterns.append(line) + except OSError: pass if p.parent == p: break p = p.parent - try: - import pathspec + import pathspec # type: ignore return pathspec.PathSpec.from_lines("gitwildmatch", patterns) except ImportError: - # fallback: 简单 glob 匹配 - print(" (pathspec 未安装,使用简单 .gitignore 匹配)", file=sys.stderr) - patterns_set = set(patterns) - return patterns_set, None + log(" (pathspec 未安装,使用 fnmatch 兜底匹配)") + return patterns -def is_ignored(filepath, spec, src_root): - """判断文件是否应被忽略""" - name = filepath.name - # 内置排除 - if name in ALWAYS_EXCLUDE: - return True - if name.endswith(".pyc") or name.endswith(".pyo"): - return True - if name == ".gitignore": - return True - - # .gitignore 规则 +def is_ignored(spec, rel: str) -> bool: + if isinstance(spec, list): + return _fnmatch_fallback(spec, rel) try: - rel = filepath.relative_to(src_root) - rel_str = str(rel) - except ValueError: - return False - - if spec is None: - return False - - try: - return spec.match_file(rel_str) + return spec.match_file(rel) except Exception: return False -def sanitize_file(path): - """检查文件中的隐私信息""" - try: - content = Path(path).read_text() - except Exception: +def collect_files(sources: list[Path], work_dir: Path, spec, extra_excludes: list[str]) -> int: + """把源文件复制进工作区(保持相对结构)。返回文件数。""" + count = 0 + for src in sources: + src = src.resolve() + if src.is_file(): + (work_dir / src.name).write_bytes(src.read_bytes()) + count += 1 + continue + if not src.is_dir(): + log(f" ⚠️ 跳过不存在的路径: {src}") + continue + for f in sorted(src.rglob("*")): + if not f.is_file(): + continue + if any(part in ALWAYS_EXCLUDE_DIRS for part in f.parts): + continue + 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 + dest = work_dir / rel + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(f, dest) + size = f.stat().st_size + if size > MAX_WARN_BYTES: + log(f" ⚠️ 大文件 {rel} ({size // 1024 // 1024} MB)") + count += 1 + return count + + +# ─────────────────────────── 仓库准备 ─────────────────────────── + +def login_user() -> str: + """取 tea 当前登录用户名(失败回退到 GITEA_USER)。""" + 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 + + +def ensure_repo(repo: str, description: str, owner: str, branch: str, + private: bool, dry_run: bool) -> bool: + """确保仓库存在。返回是否新建。 + + ⚠️ tea 0.15.0 的坑:给自己命名空间建仓时传 `--owner <自己>` 会报 + `Error: not found`;必须省略 --owner。只有 owner ≠ 登录用户(建到 + 组织/他人命名空间)时才传。 + """ + if dry_run: + log(" (dry-run:跳过建仓)") 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) + 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 collect_files(dirs, work_dir): - """收集需要推送的文件,应用 .gitignore 过滤""" - print("[3/5] 准备推送内容(应用 .gitignore 过滤)...", file=sys.stderr) - spec = load_gitignore_rules(dirs) - count = 0 - - for d in dirs: - src = Path(d).resolve() - if src.is_file(): - dest = work_dir / src.name - dest.write_bytes(src.read_bytes()) - print(f" + {src.name}", file=sys.stderr) - count += 1 - elif src.is_dir(): - for f in sorted(src.rglob("*")): - if not f.is_file(): - continue - if ".git" in f.parts: - continue - if is_ignored(f, spec, src): - continue - rel = f.relative_to(src) - dest = work_dir / rel - dest.parent.mkdir(parents=True, exist_ok=True) - dest.write_bytes(f.read_bytes()) - print(f" + {rel}", file=sys.stderr) - count += 1 - - print(f" 共收集 {count} 个文件", file=sys.stderr) - - -def main(): - if len(sys.argv) < 3: - print("用法: gitea_push.py ", file=sys.stderr) - print("示例: gitea_push.py my-tool '一个工具' ./src/", file=sys.stderr) - sys.exit(1) - - repo_name = sys.argv[1] - description = sys.argv[2] - dirs = [d for d in sys.argv[3:] if d.strip()] - - if not dirs: - print("❌ 请至少指定一个文件或目录", file=sys.stderr) - sys.exit(1) - - # 1. 隐私检查 - print("[1/5] 隐私检查...", file=sys.stderr) - has_issues = False - for d in dirs: - 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) +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: - raise + git(["remote", "add", "origin", url], work) + return work, False - # 3. 收集文件(使用 .gitignore) - work_dir = Path(tempfile.mkdtemp(prefix="gitea_")) - collect_files(dirs, work_dir) + 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 - # 4. Git 初始化并推送到 master 分支 - print(f"[4/5] 推送到 Gitea (分支: {DEFAULT_BRANCH})...", file=sys.stderr) - subprocess.run(["git", "init", "-b", DEFAULT_BRANCH], cwd=work_dir, capture_output=True) - subprocess.run(["git", "config", "user.name", GITEA_USER], cwd=work_dir, capture_output=True) - subprocess.run(["git", "config", "user.email", GITEA_EMAIL], cwd=work_dir, capture_output=True) - subprocess.run(["git", "add", "-A"], cwd=work_dir, capture_output=True) - commit_res = subprocess.run( - ["git", "commit", "-m", f"feat: {description}"], - cwd=work_dir, capture_output=True, text=True - ) - if commit_res.returncode != 0: - # 可能是空提交(所有文件都被过滤了) - print(f" commit 可能为空: {commit_res.stderr.strip()}", file=sys.stderr) +def align_with_remote(work: Path, branch: str, *, from_git: bool) -> str: + """把本地分支对齐到远程历史,避免 'fetch first' 拒绝。 - repo_url = f"git@{GIT_HOST}:{GITEA_USER}/{repo_name}.git" - remote_res = subprocess.run( - ["git", "remote", "add", "origin", repo_url], - cwd=work_dir, capture_output=True, text=True - ) - if remote_res.returncode != 0: - subprocess.run(["git", "remote", "set-url", "origin", repo_url], - cwd=work_dir, capture_output=True) + 非 --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" - push_result = subprocess.run( - ["git", "push", "-u", "origin", DEFAULT_BRANCH, "--force"], - cwd=work_dir, - capture_output=True, text=True, - env={**os.environ, "GIT_SSH_COMMAND": "ssh -o StrictHostKeyChecking=accept-new"} - ) - if push_result.returncode != 0: - 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) +# ─────────────────────────── 提交 & 推送 ─────────────────────────── - print(json.dumps({ - "ok": True, - "repo": repo_name, - "url": browse_url, - "clone": clone_url, - "branch": DEFAULT_BRANCH - })) +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. 隐私检查 + step(1, total_steps, "隐私检查…") + if args.no_scan: + log(" (已跳过)") + else: + 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(" ✓ 未发现敏感内容") + + # 2. 仓库准备 + step(2, total_steps, f"确保仓库 {args.owner}/{args.repo_name}…") + is_new = ensure_repo(args.repo_name, args.description, args.owner, + branch, args.private, dry) + log(" ✓ 新建仓库" if is_new else " ✓ 复用现有仓库") + + # 3. 工作区 + 对齐远程 + step(3, total_steps, "准备工作区并对齐远程历史…") + work, is_temp = prepare_worktree(sources, args.repo_name, args.owner, + branch, Path(args.from_git) if args.from_git else None, + args.exclude) + 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}") + + if args.from_git: + # --from-git:推送仓库已有的提交,不自动提交未完成的工作区改动 + dirty = git(["status", "--porcelain"], work, check=False).stdout.strip() + if dirty: + log(" ⚠️ 工作区有未提交改动,不会被推送(请先自行 commit)") + changed = False + else: + # 普通目录推送:全部纳入暂存区(含新增/修改/删除) + changed = commit_if_changed(work, message, add_all=True) + + # 4. 推送 + if dry: + step(4, total_steps, "dry-run:跳过推送") + log(f" 工作区: {work}") + else: + step(4, total_steps, f"推送到 {args.owner}/{args.repo_name} ({branch})…") + if args.from_git: + ahead = git(["rev-list", "--count", f"origin/{branch}..HEAD"], + work, check=False).stdout.strip() + if ahead in ("", "0"): + log(" 本地无新提交,远程已是最新") + else: + log(f" 本地领先远程 {ahead} 个提交") + push_with_retry(work, branch) + else: + push_with_retry(work, branch) + + browse = f"{GITEA_URL}/{args.owner}/{args.repo_name}" + log("") + log(f"✅ 完成!浏览: {browse}") + print(json.dumps({ + "ok": True, + "repo": args.repo_name, + "owner": args.owner, + "url": browse, + "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__": - main() + sys.exit(main())