替换 gh CLI 为 Gitea REST API，新增 fetch_gitea_data.py
- 移除对 gh CLI 的依赖，改用 Gitea /api/v1 接口 - fetch_gitea_data.py：自动解析 git remote 获取实例地址和仓库路径，支持分页，输出结构化 JSON - 私有仓库通过 GITEA_TOKEN 环境变量鉴权，公开仓库无需配置 - 更新 SKILL.md 对应执行步骤 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
admin
parent
d139ff3398
commit
c734958fa6
@ -1,6 +1,6 @@
|
||||
---
|
||||
name: weekly-report
|
||||
description: 生成项目过去7天的周报，统计每位协作者的 commit 和 PR，生成个人工作日志与项目整体报告，通过 SMTP 发送到配置邮箱
|
||||
description: 生成项目过去7天的周报，通过 Gitea API 统计每位协作者的 commit 和 PR，生成个人工作日志与项目整体报告，通过 SMTP 发送邮件
|
||||
argument-hint: ""
|
||||
---
|
||||
|
||||
@ -8,7 +8,7 @@ argument-hint: ""
|
||||
|
||||
## 步骤 1：读取配置
|
||||
|
||||
读取当前项目根目录的 `report-config.json`，格式如下：
|
||||
读取当前项目根目录的 `report-config.json`：
|
||||
|
||||
```json
|
||||
{
|
||||
@ -19,79 +19,67 @@ argument-hint: ""
|
||||
|
||||
如果文件不存在，**停止执行**，提示用户在项目根目录创建该文件后重试。
|
||||
|
||||
## 步骤 2：计算日期范围
|
||||
|
||||
用 Python 计算日期，避免平台差异：
|
||||
|
||||
```bash
|
||||
python -c "from datetime import date, timedelta; d=date.today()-timedelta(days=7); print(d)"
|
||||
```
|
||||
|
||||
## 步骤 3：拉取最新代码
|
||||
## 步骤 2：拉取最新代码
|
||||
|
||||
```bash
|
||||
git pull origin main
|
||||
```
|
||||
|
||||
如果当前不在 git 仓库内，停止并提示用户。
|
||||
如果当前目录不是 git 仓库，停止并提示用户。
|
||||
|
||||
## 步骤 4：获取过去 7 天的提交记录
|
||||
## 步骤 3：从 Gitea API 拉取数据
|
||||
|
||||
使用 Glob 工具找到本技能目录下 `fetch_gitea_data.py` 的绝对路径，然后执行：
|
||||
|
||||
```bash
|
||||
git log origin/main --since="7 days ago" --format="%an|||%ae|||%ad|||%s" --date=short --no-merges
|
||||
python <fetch_gitea_data.py的绝对路径> --days 7 --output /tmp/gitea_data.json
|
||||
```
|
||||
|
||||
将输出按 `|||` 分割，整理为结构化数据（作者姓名、邮箱、日期、提交说明）。
|
||||
脚本会自动从 `git remote get-url origin` 解析 Gitea 实例地址和仓库路径，无需额外配置。
|
||||
|
||||
## 步骤 5：获取过去 7 天合并的 PR
|
||||
**可选环境变量：**
|
||||
- `GITEA_TOKEN`：私有仓库访问令牌。公开仓库无需配置；私有仓库若不配置会收到 401 错误。
|
||||
|
||||
```bash
|
||||
gh pr list --state merged --base main --limit 100 --json number,title,author,mergedAt,additions,deletions
|
||||
```
|
||||
执行完成后，读取 `/tmp/gitea_data.json` 获取结构化数据，包含：
|
||||
- `meta`：仓库信息和统计时间范围
|
||||
- `summary`：总 commit 数、PR 数、贡献者人数
|
||||
- `by_author`：按作者分组的 commit 和 PR 列表
|
||||
|
||||
过滤 `mergedAt` 在过去 7 天内的 PR，按作者分组。
|
||||
## 步骤 4：生成报告内容
|
||||
|
||||
如果 `gh` 命令不可用，跳过此步并在报告中注明。
|
||||
基于 `/tmp/gitea_data.json` 的数据，生成以下内容：
|
||||
|
||||
## 步骤 6：生成报告内容
|
||||
### 个人周报（每位协作者一份，中文）
|
||||
|
||||
### 个人周报（每位协作者一份）
|
||||
|
||||
格式要求：
|
||||
- 姓名与提交统计（commit 数、PR 数、增删行数）
|
||||
每人包含：
|
||||
- 姓名、本周 commit 数、合并 PR 数
|
||||
- 合并 PR 列表（编号、标题）
|
||||
- 关键 commit 摘要（按日期列出，去除 merge commit）
|
||||
- 工作总结（2-3 句话，用中文概括本周主要工作）
|
||||
- 工作总结：2-3 句话概括本周主要工作内容和贡献
|
||||
|
||||
### 项目整体进度报告
|
||||
### 项目整体进度报告（中文）
|
||||
|
||||
格式要求：
|
||||
包含：
|
||||
- 统计摘要：参与人数、总 commit 数、合并 PR 总数
|
||||
- 主要完成内容（按功能或模块归类，中文）
|
||||
- 主要完成内容（按功能或模块归类）
|
||||
- 整体进度评估（根据提交内容客观描述）
|
||||
- 如有未合并的 open PR，列出数量作为下周关注点
|
||||
- 如有 open PR（未合并），列出数量作为下周关注点
|
||||
|
||||
### 拼装 HTML 报告
|
||||
### 拼装 HTML 邮件
|
||||
|
||||
将个人周报和整体报告拼装成一份 HTML 邮件，写入临时文件：
|
||||
将报告拼装成 HTML，写入临时文件：
|
||||
|
||||
```bash
|
||||
python -c "
|
||||
from datetime import date
|
||||
filename = f'/tmp/weekly_report_{date.today().strftime(\"%Y%m%d\")}.html'
|
||||
print(filename)
|
||||
"
|
||||
python -c "from datetime import date; print(f'/tmp/weekly_report_{date.today().strftime(\"%Y%m%d\")}.html')"
|
||||
```
|
||||
|
||||
HTML 结构建议：
|
||||
1. 标题：`【周报】<project_name> - <日期范围>`
|
||||
2. 项目整体报告（放在最前）
|
||||
HTML 结构：
|
||||
1. 页头：`【周报】<project_name> - <日期范围>`
|
||||
2. 项目整体报告
|
||||
3. 分割线
|
||||
4. 各协作者个人报告（每人一节）
|
||||
|
||||
## 步骤 7：检查 SMTP 环境变量
|
||||
|
||||
在发送前，检查以下环境变量是否全部存在：
|
||||
## 步骤 5：检查 SMTP 环境变量
|
||||
|
||||
```bash
|
||||
python -c "
|
||||
@ -99,14 +87,13 @@ import os, sys
|
||||
required = ['SMTP_HOST', 'SMTP_PORT', 'SMTP_USER', 'SMTP_PASSWORD']
|
||||
missing = [k for k in required if not os.environ.get(k)]
|
||||
if missing:
|
||||
print('缺少环境变量：' + ', '.join(missing))
|
||||
print('缺少：' + ', '.join(missing))
|
||||
sys.exit(1)
|
||||
else:
|
||||
print('环境变量检查通过')
|
||||
print('检查通过')
|
||||
"
|
||||
```
|
||||
|
||||
如果有缺失，**停止执行**，提示用户配置对应环境变量：
|
||||
如果有缺失，**停止执行**，提示用户配置以下环境变量（不要硬编码）：
|
||||
|
||||
| 变量 | 说明 | 示例 |
|
||||
|------|------|------|
|
||||
@ -115,17 +102,15 @@ else:
|
||||
| `SMTP_USER` | 发件人邮箱 | `noreply@company.com` |
|
||||
| `SMTP_PASSWORD` | SMTP 密码或授权码 | `your-auth-code` |
|
||||
|
||||
**绝对不要将这些值硬编码在任何文件中。**
|
||||
## 步骤 6：发送邮件
|
||||
|
||||
## 步骤 8：发送邮件
|
||||
|
||||
使用 Glob 工具找到与本 SKILL.md 同目录的 `send_email.py` 的绝对路径，然后执行：
|
||||
使用 Glob 找到本技能目录下 `send_email.py` 的绝对路径，然后执行：
|
||||
|
||||
```bash
|
||||
python <send_email.py的绝对路径> \
|
||||
--config report-config.json \
|
||||
--subject "【周报】<project_name> <YYYY-MM-DD>" \
|
||||
--body-file <步骤6生成的HTML文件路径>
|
||||
--body-file <步骤4生成的HTML文件路径>
|
||||
```
|
||||
|
||||
发送成功后，告知用户已发送至哪些邮箱，并清理临时文件。
|
||||
发送成功后，告知用户已发送至哪些邮箱，并删除 `/tmp/gitea_data.json` 和 HTML 临时文件。
|
||||
|
||||
218
skills/weekly-report/fetch_gitea_data.py
Normal file
218
skills/weekly-report/fetch_gitea_data.py
Normal file
@ -0,0 +1,218 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
从 Gitea API 拉取过去 N 天的 commits 和 merged PR 数据。
|
||||
自动从 git remote origin 解析 Gitea 实例地址和仓库路径。
|
||||
|
||||
可选环境变量：
|
||||
GITEA_TOKEN - 私有仓库访问令牌（公开仓库无需配置）
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
def get_remote_info() -> tuple[str, str, str]:
|
||||
"""从 git remote origin 解析 base_url、owner、repo"""
|
||||
try:
|
||||
url = subprocess.check_output(
|
||||
["git", "remote", "get-url", "origin"],
|
||||
stderr=subprocess.DEVNULL,
|
||||
text=True,
|
||||
).strip()
|
||||
except subprocess.CalledProcessError:
|
||||
print("错误：当前目录不是 git 仓库，或没有 origin remote", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# 去掉 .git 后缀
|
||||
if url.endswith(".git"):
|
||||
url = url[:-4]
|
||||
|
||||
parsed = urlparse(url)
|
||||
base_url = f"{parsed.scheme}://{parsed.netloc}"
|
||||
# path 格式: /owner/repo
|
||||
parts = parsed.path.strip("/").split("/")
|
||||
if len(parts) < 2:
|
||||
print(f"错误：无法从 remote URL 解析 owner/repo：{url}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
owner, repo = parts[0], parts[1]
|
||||
return base_url, owner, repo
|
||||
|
||||
|
||||
def gitea_get(base_url: str, path: str) -> object:
|
||||
"""调用 Gitea API，自动附加 token（如有）"""
|
||||
token = os.environ.get("GITEA_TOKEN", "")
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if token:
|
||||
headers["Authorization"] = f"token {token}"
|
||||
|
||||
url = f"{base_url}/api/v1{path}"
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||||
return json.loads(resp.read().decode())
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 401:
|
||||
print("错误：API 返回 401，私有仓库需要设置 GITEA_TOKEN 环境变量", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
print(f"错误：API 请求失败 {url} -> HTTP {e.code}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except urllib.error.URLError as e:
|
||||
print(f"错误：无法连接到 Gitea 实例 {base_url} - {e.reason}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def fetch_commits(base_url: str, owner: str, repo: str, since: datetime) -> list[dict]:
|
||||
"""拉取指定日期之后的所有 commits（分页）"""
|
||||
since_str = since.strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
commits = []
|
||||
page = 1
|
||||
while True:
|
||||
data = gitea_get(
|
||||
base_url,
|
||||
f"/repos/{owner}/{repo}/commits?sha=main&limit=50&page={page}&since={since_str}",
|
||||
)
|
||||
if not data:
|
||||
break
|
||||
commits.extend(data)
|
||||
if len(data) < 50:
|
||||
break
|
||||
page += 1
|
||||
return commits
|
||||
|
||||
|
||||
def fetch_merged_prs(base_url: str, owner: str, repo: str, since: datetime) -> list[dict]:
|
||||
"""拉取已关闭的 PR，过滤出 since 之后合并的"""
|
||||
prs = []
|
||||
page = 1
|
||||
while True:
|
||||
data = gitea_get(
|
||||
base_url,
|
||||
f"/repos/{owner}/{repo}/pulls?state=closed&limit=50&page={page}",
|
||||
)
|
||||
if not data:
|
||||
break
|
||||
for pr in data:
|
||||
merged_at = pr.get("merged_at") or pr.get("merged")
|
||||
if not merged_at:
|
||||
continue
|
||||
# 解析合并时间
|
||||
try:
|
||||
merged_dt = datetime.fromisoformat(merged_at.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
continue
|
||||
if merged_dt >= since:
|
||||
prs.append(pr)
|
||||
# 如果最后一条 PR 的合并时间早于 since，可以提前退出
|
||||
if data:
|
||||
last_merged = data[-1].get("merged_at") or data[-1].get("merged")
|
||||
if last_merged:
|
||||
try:
|
||||
last_dt = datetime.fromisoformat(last_merged.replace("Z", "+00:00"))
|
||||
if last_dt < since:
|
||||
break
|
||||
except ValueError:
|
||||
pass
|
||||
if len(data) < 50:
|
||||
break
|
||||
page += 1
|
||||
return prs
|
||||
|
||||
|
||||
def group_by_author(commits: list[dict], prs: list[dict]) -> dict:
|
||||
"""按作者分组整理数据"""
|
||||
authors = {}
|
||||
|
||||
for c in commits:
|
||||
commit_info = c.get("commit", {})
|
||||
author = commit_info.get("author", {})
|
||||
name = author.get("name", "unknown")
|
||||
email = author.get("email", "")
|
||||
key = name
|
||||
|
||||
if key not in authors:
|
||||
authors[key] = {"name": name, "email": email, "commits": [], "prs": []}
|
||||
authors[key]["commits"].append({
|
||||
"sha": c.get("sha", "")[:8],
|
||||
"message": commit_info.get("message", "").split("\n")[0],
|
||||
"date": author.get("date", "")[:10],
|
||||
"files": [f.get("filename") for f in c.get("files", [])],
|
||||
})
|
||||
|
||||
for pr in prs:
|
||||
author = pr.get("user", {}) or pr.get("poster", {})
|
||||
name = author.get("login", "unknown")
|
||||
key = name
|
||||
|
||||
if key not in authors:
|
||||
authors[key] = {"name": name, "email": "", "commits": [], "prs": []}
|
||||
authors[key]["prs"].append({
|
||||
"number": pr.get("number"),
|
||||
"title": pr.get("title", ""),
|
||||
"merged_at": (pr.get("merged_at") or "")[:10],
|
||||
"additions": pr.get("additions", 0),
|
||||
"deletions": pr.get("deletions", 0),
|
||||
})
|
||||
|
||||
return authors
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description="拉取 Gitea 仓库过去 N 天的数据")
|
||||
parser.add_argument("--days", type=int, default=7, help="统计天数（默认7天）")
|
||||
parser.add_argument("--output", default="-", help="输出 JSON 文件路径（默认输出到 stdout）")
|
||||
args = parser.parse_args()
|
||||
|
||||
base_url, owner, repo = get_remote_info()
|
||||
since = datetime.now(timezone.utc) - timedelta(days=args.days)
|
||||
|
||||
print(f"仓库：{base_url}/{owner}/{repo}", file=sys.stderr)
|
||||
print(f"统计范围：{since.strftime('%Y-%m-%d')} 至今", file=sys.stderr)
|
||||
|
||||
print("正在拉取 commits...", file=sys.stderr)
|
||||
commits = fetch_commits(base_url, owner, repo, since)
|
||||
print(f" 共 {len(commits)} 条", file=sys.stderr)
|
||||
|
||||
print("正在拉取已合并 PR...", file=sys.stderr)
|
||||
prs = fetch_merged_prs(base_url, owner, repo, since)
|
||||
print(f" 共 {len(prs)} 条", file=sys.stderr)
|
||||
|
||||
result = {
|
||||
"meta": {
|
||||
"base_url": base_url,
|
||||
"owner": owner,
|
||||
"repo": repo,
|
||||
"since": since.strftime("%Y-%m-%d"),
|
||||
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||||
},
|
||||
"summary": {
|
||||
"total_commits": len(commits),
|
||||
"total_prs": len(prs),
|
||||
"contributors": 0,
|
||||
},
|
||||
"by_author": group_by_author(commits, prs),
|
||||
"all_prs": prs,
|
||||
}
|
||||
result["summary"]["contributors"] = len(result["by_author"])
|
||||
|
||||
output = json.dumps(result, ensure_ascii=False, indent=2)
|
||||
|
||||
if args.output == "-":
|
||||
print(output)
|
||||
else:
|
||||
with open(args.output, "w", encoding="utf-8") as f:
|
||||
f.write(output)
|
||||
print(f"数据已写入 {args.output}", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
x
Reference in New Issue
Block a user