Initialize Gitea user cleanup tool with main script, README, environment configuration, and dependencies. Added .env.example and .gitignore files for environment variable management and to exclude sensitive files from version control.

.env.example

@@ -0,0 +1,3 @@
+GITEA_API_TOKEN=
+GITEA_URL=http://49.232.207.113:3000
+

.gitignore

@@ -0,0 +1,2 @@
+.env
+.venv

README.md

@@ -0,0 +1,100 @@
+# Gitea 用户清理工具
+
+一个用于清理 Gitea 实例中不在指定组织内的用户的 Python 脚本。
+
+## 功能特性
+
+- 获取 Gitea 实例中的所有用户
+- 获取指定组织的所有成员
+- 自动删除不在组织内的用户及其所有仓库
+- 使用进度条显示删除进度
+
+## 环境要求
+
+- Python 3.11
+- Gitea 实例访问权限
+- Gitea API Token（管理员权限）
+
+## 安装
+
+1. 克隆或下载此项目
+
+2. 安装依赖：
+
+```bash
+pip install -r requirements.txt
+```
+
+## 配置
+
+在项目根目录创建 `.env` 文件，配置以下环境变量：
+
+```env
+GITEA_URL=https://your-gitea-instance.com
+GITEA_API_TOKEN=your_api_token_here
+```
+
+### 获取 API Token
+
+1. 登录 Gitea 实例
+2. 进入 **设置** → **应用** → **生成新令牌**
+3. 选择权限范围（需要管理员权限）
+4. 复制生成的令牌到 `.env` 文件
+
+## 使用方法
+
+### 基本使用
+
+运行脚本将自动：
+1. 获取所有用户列表
+2. 获取指定组织的成员列表（默认为 "TianMa431"）
+3. 删除不在组织内的用户及其所有仓库
+
+```bash
+python main.py
+```
+
+### 自定义组织名称
+
+在 `main.py` 中修改 `get_org_user_names()` 函数的默认参数：
+
+```python
+org_user_names = get_org_user_names("YourOrgName")
+```
+
+### 单独删除用户
+
+取消注释并修改以下代码来删除特定用户：
+
+```python
+delete_user("username")
+```
+
+## 主要函数
+
+### `get_user_names()`
+获取 Gitea 实例中的所有用户名列表。
+
+### `get_org_user_names(org_name="TianMa431")`
+获取指定组织的所有成员用户名列表。
+
+### `delete_user(username)`
+删除指定用户及其所有仓库。**此操作不可逆！**
+
+## ⚠️ 警告
+
+- **此脚本会永久删除用户及其所有仓库数据，操作不可恢复！**
+- 使用前请确保已备份重要数据
+- 建议先在测试环境验证
+- 确保 API Token 具有足够的权限
+
+## 依赖包
+
+- `requests` - HTTP 请求库
+- `python-dotenv` - 环境变量管理
+- `tqdm` - 进度条显示
+
+## 许可证
+
+本项目仅供学习和内部使用。
+

main.py

@@ -0,0 +1,89 @@
+import asyncio
+import os
+from dotenv import load_dotenv
+from tqdm.asyncio import tqdm as atqdm
+import httpx
+
+load_dotenv()
+
+gitea_url = os.getenv("GITEA_URL")
+api_token = os.getenv("GITEA_API_TOKEN")
+headers = {"Authorization": f"token {api_token}"}
+
+# 强制保留用户
+PROTECTED_USER_NAMES = ['Ding-JJ', 'GAOJJJ', 'LAN206', 'Old-Watch', 'Yuayifa', 'bakaEC', 'cheny', 'gaoyuki', 'rs', 'shanshan', 'test2333', 'tianma_admin', 'zhangzd', 'zoujielin']
+
+# 最大并发数
+MAX_CONCURRENCY = 20
+
+
+# 删除用户及其仓库的异步函数
+async def delete_user(username, client: httpx.AsyncClient, semaphore: asyncio.Semaphore):
+    try:
+        async with semaphore:
+            # 获取用户的所有仓库
+            repos_url = f"{gitea_url}/api/v1/users/{username}/repos"
+            repos_resp = await client.get(repos_url)
+            if repos_resp.status_code == 200:
+                repos = repos_resp.json()
+                # 并行删除所有仓库
+                delete_tasks = []
+                for repo in repos:
+                    delete_repo_url = f"{gitea_url}/api/v1/repos/{username}/{repo['name']}"
+                    delete_tasks.append(client.delete(delete_repo_url))
+                if delete_tasks:
+                    await asyncio.gather(*delete_tasks, return_exceptions=True)
+            
+            # 删除用户
+            user_url = f"{gitea_url}/api/v1/admin/users/{username}"
+            await client.delete(user_url)
+    except (httpx.HTTPError, httpx.RequestError) as e:
+        print(f"删除用户 {username} 时出错: {e}")
+
+
+async def get_user_names(client: httpx.AsyncClient):
+    users_url = f"{gitea_url}/api/v1/admin/users"
+    resp = await client.get(users_url)
+    data = resp.json()
+    user_names = [user["username"] for user in data]
+    return user_names
+
+
+async def get_org_user_names(org_name="TianMa431", client: httpx.AsyncClient = None):
+    user_names = []
+    org_url = f"{gitea_url}/api/v1/orgs/{org_name}/members"
+    resp = await client.get(org_url)
+    data = resp.json()
+    for user in data:
+        user_names.append(user["username"])
+    return user_names
+
+
+async def delete_users_parallel(user_names):
+    async with httpx.AsyncClient(headers=headers, timeout=30.0) as client:
+        
+        semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
+        print(f"准备删除 {len(user_names)} 个用户...")
+        
+        # 创建所有删除任务
+        tasks = [delete_user(username, client, semaphore) for username in user_names if username not in PROTECTED_USER_NAMES]
+        
+        # 并行执行所有删除任务，并显示进度条
+        await atqdm.gather(*tasks, desc="删除用户中")
+
+
+async def main():
+    async with httpx.AsyncClient(headers=headers, timeout=30.0) as client:
+        # 并行获取用户列表和组织成员列表
+        user_names, org_user_names = await asyncio.gather(
+            get_user_names(client),
+            get_org_user_names(client=client)
+        )
+    
+    # 并行删除用户
+    users_to_delete = [name for name in user_names if name not in org_user_names]
+    await delete_users_parallel(users_to_delete)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

requirements.txt

@@ -0,0 +1,13 @@
+anyio==4.12.0
+certifi==2025.11.12
+charset-normalizer==3.4.4
+dotenv==0.9.9
+h11==0.16.0
+httpcore==1.0.9
+httpx==0.28.1
+idna==3.11
+python-dotenv==1.2.1
+requests==2.32.5
+tqdm==4.67.1
+typing_extensions==4.15.0
+urllib3==2.6.2