feat(gitea): 银河锋锐 Gitea 对接(申请/认领/改密/头像迁移 API + UserGiteaAccount)

This commit is contained in:
TermiNexus committed 2026-08-09 18:28:56 +08:00
1 parent d4558fbacb
commit 7dab803ef3
8 files changed
+653 -2

No files matched your search

+2 -2
View File
@@ -5,7 +5,7 @@
<link rel="icon" type="image/svg+xml" href="/favicon.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>GERP 管理系统</title>
<script type="module" crossorigin src="/assets/index-DI9-hmlZ.js"></script>
<script type="module" crossorigin src="/assets/index-BAY4q3dg.js"></script>
<link rel="modulepreload" crossorigin href="/assets/rolldown-runtime-hePW80VL.js">
<link rel="modulepreload" crossorigin href="/assets/element-plus-BsWRg887.js">
<link rel="modulepreload" crossorigin href="/assets/utils-C0vl14oO.js">
@@ -14,7 +14,7 @@
<link rel="modulepreload" crossorigin href="/assets/formkit-sTHkhPfk.js">
<link rel="modulepreload" crossorigin href="/assets/vue-vendor-r2EuHewM.js">
<link rel="modulepreload" crossorigin href="/assets/serverScope-C6yhSV7-.js">
<link rel="stylesheet" crossorigin href="/assets/style-4vmlf1rA.css">
<link rel="stylesheet" crossorigin href="/assets/style-BfILH_3P.css">
</head>
<body>
<div id="app"></div>
+216
View File
@@ -0,0 +1,216 @@
"""模块:Gitea 代码托管服务管理
通过 Gitea REST API(/api/v1)管理员工 Gitea 账号:
- 创建用户(admin API)、加入组织、加入团队(Owners)、查询、删除。
配置来自 secrets.ini 的 [gitea] 段(BASE_URL / API_TOKEN / ORG_NAME / OWNER_ROLE)。
已验证的 API 契约(Gitea 1.25.4):
- 当前用户:GET /api/v1/user
- 创建用户:POST /api/v1/admin/users
{"username":"<用户名>","email":"<邮箱>","password":"<密码>","full_name":"<姓名>","must_change_password":false}
- 查询用户:GET /api/v1/users/<login>(不存在返回 404)
- 删除用户:DELETE /api/v1/admin/users/<login>
- 加入组织:Gitea 无直接添加成员接口,须通过团队
(PUT /api/v1/teams/<team_id>/members/<login>,团队成员自动成为组织成员)
- 组织团队:GET /api/v1/orgs/<org>/teams → [{id,name,...}]
- 创建团队:POST /api/v1/orgs/<org>/teams → {"name":"Members","permission":"read"}
认证:Authorization: token <API_TOKEN>(需管理员权限);
头像上传使用用户自身的 Basic Auth(login + password),因为 /user/avatar 仅允许当前用户操作。
"""
import base64
import logging
import requests
from gvsds import secrets as app_secrets
logger = logging.getLogger(__name__)
class GiteaError(Exception):
"""Gitea 管理 API 调用失败。"""
def __init__(self, status_code, message):
super().__init__(message)
self.status_code = status_code
class GiteaApiClient:
"""Gitea REST API 管理服务层。"""
@staticmethod
def _BaseUrl():
return app_secrets.get('gitea', 'BASE_URL', 'https://git.gvsds.com').rstrip('/')
@staticmethod
def _Token():
return app_secrets.get('gitea', 'API_TOKEN', '')
@staticmethod
def _OrgName():
return app_secrets.get('gitea', 'ORG_NAME', 'GVSDS')
@staticmethod
def _OwnerRole():
return app_secrets.get('gitea', 'OWNER_ROLE', '银河锋锐负责人')
@classmethod
def _AuthHeader(cls):
return {'Authorization': f'token {cls._Token()}'}
@classmethod
def _Request(cls, method, path, **kwargs):
"""调用 Gitea API,非 2xx 一律抛 GiteaError。"""
url = f'{cls._BaseUrl()}/api/v1{path}'
headers = cls._AuthHeader()
headers.setdefault('Content-Type', 'application/json')
try:
resp = requests.request(method, url, headers=headers, timeout=15, **kwargs)
except requests.RequestException as e:
logger.error('[Gitea] %s %s 请求失败: %s', method, path, e)
raise GiteaError(0, f'连接 Gitea 失败: {e}') from e
try:
payload = resp.json()
except ValueError:
payload = {}
if resp.status_code >= 400:
msg = ''
if isinstance(payload, dict):
msg = payload.get('message') or payload.get('error') or ''
logger.error('[Gitea] %s %s -> %s %s', method, path, resp.status_code, msg or resp.reason)
raise GiteaError(resp.status_code, msg or f'HTTP {resp.status_code}')
return payload
@classmethod
def GetUser(cls, login):
"""查询用户,返回 dict 或 None(不存在返回 404 → None)。"""
try:
return cls._Request('GET', f'/users/{login}')
except GiteaError as e:
if e.status_code == 404:
return None
raise
@classmethod
def SearchUserByEmail(cls, email):
"""按邮箱搜索用户,返回 email 精确匹配的用户 dict 或 None。
Gitea GET /users/search 的 q 参数同时匹配用户名与邮箱,此处过滤精确邮箱。
"""
if not email:
return None
try:
payload = cls._Request('GET', '/users/search', params={'q': email, 'limit': 20})
except GiteaError as e:
logger.error('[Gitea] 按邮箱搜索用户失败: %s', e)
return None
items = payload.get('data', []) if isinstance(payload, dict) else []
email_lower = email.lower()
for it in items:
if (it.get('email') or '').lower() == email_lower:
return it
return None
@classmethod
def CreateUser(cls, login, email, password, full_name=''):
"""创建 Gitea 用户(管理员 API)。
注意:POST /admin/users 的必填字段名为 username(非 login)。
"""
return cls._Request('POST', '/admin/users', json={
'username': login,
'email': email,
'password': password,
'full_name': full_name,
'must_change_password': False,
})
@classmethod
def DeleteUser(cls, login):
"""删除 Gitea 用户(管理员 API)。"""
return cls._Request('DELETE', f'/admin/users/{login}')
@classmethod
def ChangePassword(cls, login, new_password):
"""修改用户密码(管理员 API)。"""
return cls._Request('PATCH', f'/admin/users/{login}', json={'password': new_password})
@classmethod
def UploadAvatar(cls, login, password, avatar_bytes):
"""上传用户头像(使用用户自身 Basic Auth)。
POST /user/avatar 仅允许当前用户操作,故用用户账号密码做 Basic 认证;
body 为 JSON {"image": "<base64 图片>"}(Gitea UpdateUserAvatarOption 契约)。
"""
url = f'{cls._BaseUrl()}/api/v1/user/avatar'
basic = base64.b64encode(f'{login}:{password}'.encode('utf-8')).decode('ascii')
headers = {'Authorization': f'Basic {basic}', 'Content-Type': 'application/json'}
payload = {'image': base64.b64encode(avatar_bytes).decode('ascii')}
try:
resp = requests.post(url, headers=headers, json=payload, timeout=15)
except requests.RequestException as e:
logger.error('[Gitea] 上传头像请求失败: %s', e)
raise GiteaError(0, f'上传头像失败: {e}') from e
if resp.status_code >= 400:
try:
payload = resp.json()
except ValueError:
payload = {}
msg = payload.get('message') or payload.get('error') or f'HTTP {resp.status_code}'
logger.error('[Gitea] 上传头像 -> %s %s', resp.status_code, msg)
raise GiteaError(resp.status_code, msg)
return resp.json() if resp.content else {}
@classmethod
def ListTeams(cls, org):
"""列出组织团队。"""
return cls._Request('GET', f'/orgs/{org}/teams')
@classmethod
def AddUserToTeam(cls, team_id, login):
"""把用户加入团队(团队成员自动成为组织成员——Gitea 无直接添加组织成员接口)。"""
return cls._Request('PUT', f'/teams/{team_id}/members/{login}')
@classmethod
def GetOrCreateMemberTeam(cls, org, team_name='Members'):
"""获取组织普通成员团队,不存在则创建。
Gitea 通过团队加入组织:优先返回现有的非 Owners 团队,
否则创建名为 team_name 的只读团队。
"""
try:
teams = cls.ListTeams(org)
except GiteaError as e:
logger.error('[Gitea] 查询组织团队失败: %s', e)
raise
for t in teams:
name = t.get('name') or ''
if name == team_name or (name and name != 'Owners'):
return t.get('id')
created = cls._Request('POST', f'/orgs/{org}/teams', json={
'name': team_name,
'permission': 'read',
})
return created.get('id')
@classmethod
def OwnersTeamId(cls, org):
"""返回组织 Owners 团队 id(不存在返回 None)。"""
try:
teams = cls.ListTeams(org)
except GiteaError as e:
logger.error('[Gitea] 查询组织团队失败: %s', e)
return None
for t in teams:
if t.get('name') == 'Owners':
return t.get('id')
return None
@classmethod
def SessionConfig(cls):
"""返回前端展示所需的连接信息。"""
return {
'BaseUrl': cls._BaseUrl(),
'Org': cls._OrgName(),
'OwnerRole': cls._OwnerRole(),
}
@@ -0,0 +1,35 @@
"""Add UserGiteaAccount
用户 Gitea 账号表——记录员工与 Gitea 账号的映射
(申请时在 git.gvsds.com 创建用户并加入组织,本表记录归属与状态)。
"""
import datetime
from django.db import migrations, models
import gvsds.apps.core.models
class Migration(migrations.Migration):
dependencies = [
('gerp', '0020_mail_contact'),
]
operations = [
migrations.CreateModel(
name='UserGiteaAccount',
fields=[
('user_gitea_account_uuid', gvsds.apps.core.models.UUIDField(db_column='UserGiteaAccountUUID', primary_key=True, serialize=False)),
('user_uuid', models.ForeignKey(db_column='UserUUID', db_index=True, on_delete=models.CASCADE, to='core.user')),
('gitea_login', models.CharField(db_column='GiteaLogin', max_length=256, unique=True)),
('gitea_user_id', models.IntegerField(blank=True, db_column='GiteaUserId', null=True)),
('status', models.SmallIntegerField(db_column='Status', default=1)),
('create_time', models.DateTimeField(db_column='CreateTime', default=datetime.datetime.utcnow)),
],
options={
'verbose_name': '用户Gitea账号',
'verbose_name_plural': '用户Gitea账号',
'db_table': 'UserGiteaAccount',
'ordering': ['-CreateTime'],
},
),
]
@@ -0,0 +1,20 @@
"""Add Email to UserGiteaAccount
记录 Gitea 账号绑定的工作邮箱("我的邮箱"的 xxx@gvsds.com)。
"""
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('gerp', '0021_user_gitea_account'),
]
operations = [
migrations.AddField(
model_name='usergiteaaccount',
name='email',
field=models.CharField(blank=True, db_column='Email', default='', max_length=256),
),
]
+26
View File
@@ -746,6 +746,32 @@ class MailContact(QModel):
return self.ContactEmail
class UserGiteaAccount(QModel):
"""用户 Gitea 账号表——记录员工与 Gitea 账号的映射
申请 Gitea 账号时在 git.gvsds.com 创建用户并加入组织(GVSDS),
本表记录归属关系;密码不落库(员工自设,仅存于 Gitea)。
"""
UserGiteaAccountUUID = UUIDField(primary_key=True, verbose_name='Gitea记录UUID')
UserUUID = models.ForeignKey(User, on_delete=models.CASCADE, db_column='UserUUID', db_index=True, verbose_name='用户UUID')
GiteaLogin = models.CharField(max_length=256, unique=True, verbose_name='Gitea用户名')
GiteaUserId = models.IntegerField(null=True, blank=True, verbose_name='Gitea用户ID')
Email = models.CharField(max_length=256, default='', blank=True, db_column='Email', verbose_name='工作邮箱')
Status = models.SmallIntegerField(default=1, verbose_name='状态(1启用/0禁用)')
CreateTime = models.DateTimeField(default=datetime.datetime.utcnow, verbose_name='创建时间')
class Meta:
db_table = 'UserGiteaAccount'
app_label = 'gerp'
ordering = ['-CreateTime']
verbose_name = '用户Gitea账号'
verbose_name_plural = '用户Gitea账号'
def __str__(self):
return self.GiteaLogin
# ==================== 从 gvsdsdk 导入重复模型 ====================
# 以下模型原在 gerp/models.py 定义,现统一依赖 gvsdsdk.models(映射 SAAS 主库)
# 注意:gvsdsdk 模型用 UUIDField(非 ForeignKey),不支持 select_related 和关联属性访问
+9
View File
@@ -135,3 +135,12 @@ from .mail import (
MailAdminListView,
MailDisableView,
)
from .gitea import (
GiteaSessionView,
GiteaApplyView,
GiteaMyAccountView,
GiteaChangePasswordView,
GiteaAvatarView,
GiteaAdminListView,
GiteaDisableView,
)
+336
View File
@@ -0,0 +1,336 @@
"""模块:GERP Gitea 视图(代码托管对接)
提供员工 Gitea 账号的自助申请、我的账号、管理员的列表/禁用。
与"我的邮箱"同模式:API 请求全部走后端(管理 Token 存服务端 secrets.ini),
员工申请(自选用户名 + 自设密码)→ 创建 Gitea 用户并加入组织 → 负责人自动成为 Owner。
URL 路径最后一段作为 action 名称,无需前端额外传 Action 字段。
支持 action:
Session - 连接配置(BaseUrl / Org / OwnerRole)
Apply - 员工申请 Gitea 账号
MyAccount - 我的账号信息
AdminList - 管理员:全部账号列表
Disable - 管理员:禁用账号(删除 Gitea 用户)
"""
import logging
import re
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from gvsds import secrets as app_secrets
from gvsdsdk.model_utils import UUIDObj
from gvsdsdk.auth.permissions import IsAdminUser
from . import ELTAuthenticatedAPIView
from ..models import UserGiteaAccount, UserMailbox
from ..gitea import GiteaApiClient, GiteaError
logger = logging.getLogger(__name__)
# Gitea 用户名:字母数字开头,含字母数字/点/下划线/连字符,总长 3~39(Gitea 用户名上限)
_GITEA_LOGIN_RE = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9._-]{2,38}$')
# 密码强度:8~64 位,至少含一个字母和一个数字(与邮箱申请一致)
_PASSWORD_RE = re.compile(r'^(?=.*[A-Za-z])(?=.*\d).{8,64}$')
def _GetUserBytes(request):
"""取当前用户的 UserUUID 原始字节。"""
user = request.user
if user is None:
return None
val = getattr(user, 'UserUUID', None)
if val is None:
return None
if isinstance(val, bytes):
return val
try:
return UUIDObj(val).bytes
except Exception:
return None
def _UserCompanyNames(user_bytes):
"""返回用户所属公司名称列表(经 UserRole 关联)。"""
from gvsdsdk.models import UserRole as SdkUserRole, Company as SdkCompany
roles = SdkUserRole.objects.filter(UserUUID=user_bytes).values_list('CompanyUUID', flat=True)
comp_uuids = [c for c in roles if c]
if not comp_uuids:
return []
return list(SdkCompany.objects.filter(CompanyUUID__in=comp_uuids).values_list('CompanyName', flat=True))
def _CheckCompanyAllowed(user_bytes):
"""校验用户所属公司是否允许申请 Gitea 账号。
secrets.ini [gitea] ALLOW_COMPANY 为空时仅要求用户属于任一公司;
配置了公司名(逗号分隔)时要求匹配其中之一。
"""
names = _UserCompanyNames(user_bytes)
if not names:
return False, '当前用户未归属任何公司,无法申请 Gitea 账号'
allowed = app_secrets.get('gitea', 'ALLOW_COMPANY', '').strip()
if not allowed:
return True, ''
allowed_list = [n.strip() for n in allowed.split(',') if n.strip()]
if any(n in allowed_list for n in names):
return True, ''
return False, f'仅限 {allowed} 公司的员工申请 Gitea 账号'
def _IsOwnerRole(user_bytes):
"""用户是否拥有 OWNER_ROLE 配置的角色(负责人自动成为组织 Owner)。"""
from gvsdsdk.auth.managers import RoleManager
owner_role = GiteaApiClient._OwnerRole()
if not owner_role:
return False
try:
user_roles = RoleManager.GetUserRoleDetails(user_bytes)
except Exception:
return False
for ur in user_roles:
try:
role = RoleManager.GetRoleByUUID(ur.RoleUUID)
except Exception:
role = None
name = (ur.RoleName or '').strip()
if not name and role is not None:
name = (role.RoleName or '').strip()
if name and name == owner_role:
return True
return False
class GiteaSessionView(ELTAuthenticatedAPIView):
"""返回 Gitea 连接配置(BaseUrl / Org / OwnerRole)。"""
permission_classes = [IsAuthenticated]
def get(self, request):
return Response({'code': 0, 'data': GiteaApiClient.SessionConfig()})
class GiteaApplyView(ELTAuthenticatedAPIView):
"""员工自助申请 Gitea 账号(自选用户名 + 自设密码)。"""
permission_classes = [IsAuthenticated]
def post(self, request):
user_bytes = _GetUserBytes(request)
if not user_bytes:
return Response({'code': 1, 'msg': '无法识别当前用户'}, status=401)
# 每个用户最多一个 Gitea 账号
if UserGiteaAccount.objects.filter(UserUUID=request.user).exists():
return Response({'code': 1, 'msg': '您已拥有 Gitea 账号,一个用户只能申请一个'}, status=400)
login = (request.data.get('GiteaLogin') or '').strip()
password = request.data.get('Password') or ''
if not _GITEA_LOGIN_RE.match(login):
return Response({'code': 1, 'msg': '用户名需为字母/数字/点/下划线/连字符,长度 3~39'}, status=400)
if not _PASSWORD_RE.match(password):
return Response({'code': 1, 'msg': '密码需 8~64 位且同时包含字母和数字'}, status=400)
# 公司归属校验
ok, err = _CheckCompanyAllowed(user_bytes)
if not ok:
return Response({'code': 1, 'msg': err}, status=403)
# 唯一性校验(本表 + Gitea 侧)
if UserGiteaAccount.objects.filter(GiteaLogin=login).exists():
return Response({'code': 1, 'msg': '该用户名已被申请'}, status=400)
# 使用工作邮箱("我的邮箱"的 xxx@gvsds.com)注册 Gitea;
# 未申请企业邮箱则引导先申请
mailbox = UserMailbox.objects.filter(UserUUID=request.user, Status=1).first()
if mailbox is None:
return Response({'code': 1, 'msg': '请先申请企业邮箱,再申请 Gitea 账号'}, status=400)
email = mailbox.MailboxName
full_name = getattr(request.user, 'UserName', '') or ''
claimed = False
try:
# 用户名查重:已存在且邮箱相同(同一人)→ 认领已有账号;否则拒绝
existing = GiteaApiClient.GetUser(login)
if existing is not None:
if (existing.get('email') or '').lower() == email.lower():
created = existing
claimed = True
else:
return Response({'code': 1, 'msg': '该用户名在 Gitea 已存在'}, status=400)
else:
# 用户名未占用,但邮箱已被其他账号使用 → 引导认领
by_email = GiteaApiClient.SearchUserByEmail(email)
if by_email and (by_email.get('login') or '') != login:
return Response({
'code': 1,
'msg': f'该邮箱已被 Gitea 账号 {by_email.get("login")} 使用,请申请该用户名',
}, status=400)
# 创建 Gitea 用户
created = GiteaApiClient.CreateUser(login, email, password, full_name=full_name)
except GiteaError as e:
logger.error('[Gitea] 创建/认领用户失败: %s', e)
return Response({'code': 1, 'msg': f'创建 Gitea 用户失败: {e}'}, status=502)
gitea_user_id = created.get('id') if isinstance(created, dict) else None
org = GiteaApiClient._OrgName()
is_owner = _IsOwnerRole(user_bytes)
# 加入组织(Gitea 无直接添加成员接口,通过团队加入):
# 负责人加入 Owners 团队,普通员工加入 Members 团队(自动创建)
try:
team_id = GiteaApiClient.OwnersTeamId(org) if is_owner else GiteaApiClient.GetOrCreateMemberTeam(org)
if team_id:
GiteaApiClient.AddUserToTeam(team_id, login)
except GiteaError as e:
logger.error('[Gitea] 加入组织失败: %s', e)
return Response({'code': 1, 'msg': f'加入组织失败: {e}'}, status=502)
UserGiteaAccount.objects.create(
UserUUID=request.user,
GiteaLogin=login,
GiteaUserId=gitea_user_id,
Email=email,
Status=1,
)
return Response({
'code': 0,
'msg': 'Gitea 账号申请成功(已认领已有账号)' if claimed else 'Gitea 账号申请成功',
'data': {'GiteaLogin': login, 'Org': org, 'IsOwner': is_owner, 'Claimed': claimed, 'Email': email},
}, status=201)
class GiteaMyAccountView(ELTAuthenticatedAPIView):
"""我的 Gitea 账号信息 + 连接配置。"""
permission_classes = [IsAuthenticated]
def get(self, request):
acct = UserGiteaAccount.objects.filter(UserUUID=request.user).first()
session = GiteaApiClient.SessionConfig()
if acct is None:
return Response({'code': 0, 'data': {'Account': None, 'Session': session}})
return Response({
'code': 0,
'data': {
'Account': {
'GiteaLogin': acct.GiteaLogin,
'Email': acct.Email,
'Status': acct.Status,
'HomeUrl': f"{session['BaseUrl']}/{acct.GiteaLogin}",
'Org': session['Org'],
'CreateTime': acct.CreateTime.isoformat() if acct.CreateTime else None,
},
'Session': session,
},
})
class GiteaChangePasswordView(ELTAuthenticatedAPIView):
"""修改 Gitea 账号密码(管理员 API,无需旧密码)。"""
permission_classes = [IsAuthenticated]
def post(self, request):
acct = UserGiteaAccount.objects.filter(UserUUID=request.user).first()
if acct is None:
return Response({'code': 1, 'msg': '尚未申请 Gitea 账号'}, status=400)
new_password = request.data.get('NewPassword') or ''
if not _PASSWORD_RE.match(new_password):
return Response({'code': 1, 'msg': '密码需 8~64 位且同时包含字母和数字'}, status=400)
try:
GiteaApiClient.ChangePassword(acct.GiteaLogin, new_password)
except GiteaError as e:
logger.error('[Gitea] 修改密码失败: %s', e)
return Response({'code': 1, 'msg': f'修改密码失败: {e}'}, status=502)
return Response({'code': 0, 'msg': '密码修改成功'})
class GiteaAvatarView(ELTAuthenticatedAPIView):
"""一键迁移 GERP 头像到 Gitea。
Gitea 上传头像接口(POST /user/avatar)仅允许当前用户操作,
故用用户的 Gitea 密码做 Basic Auth;密码仅此一次传输,不落库。
"""
permission_classes = [IsAuthenticated]
def post(self, request):
acct = UserGiteaAccount.objects.filter(UserUUID=request.user).first()
if acct is None:
return Response({'code': 1, 'msg': '尚未申请 Gitea 账号'}, status=400)
password = request.data.get('Password') or ''
if not password:
return Response({'code': 1, 'msg': '请输入 Gitea 密码'}, status=400)
from gvsds.apps.core.models import UserAvatar
# UserAvatar.UserUUID 为 UUIDField(非 FK),须用 UserUUID 字节查询
user_bytes = _GetUserBytes(request)
avatar = UserAvatar.objects.filter(UserUUID=user_bytes).first() if user_bytes else None
if avatar is None:
return Response({'code': 1, 'msg': '您尚未设置 GERP 头像,请先到个人资料上传头像'}, status=400)
try:
GiteaApiClient.UploadAvatar(acct.GiteaLogin, password, bytes(avatar.UserAvatar))
except GiteaError as e:
logger.error('[Gitea] 上传头像失败: %s', e)
return Response({'code': 1, 'msg': f'上传头像失败: {e}'}, status=502)
return Response({'code': 0, 'msg': '头像迁移成功,请在 Gitea 刷新查看'})
class GiteaAdminListView(ELTAuthenticatedAPIView):
"""管理员:全部员工 Gitea 账号列表(含用户名)。"""
permission_classes = [IsAuthenticated, IsAdminUser]
def get(self, request):
from gvsds.apps.core.models import User as CoreUser
rows = list(UserGiteaAccount.objects.all().order_by('-CreateTime'))
user_map = {}
for acct in rows:
key = acct.UserUUID_id if hasattr(acct, 'UserUUID_id') else acct.UserUUID
user_map.setdefault(key, None)
if user_map:
uuids = [k for k in user_map.keys() if k]
for u in CoreUser.objects.filter(UserUUID__in=uuids):
try:
user_map[u.UserUUID.bytes] = u.UserName
except Exception:
user_map[str(u.UserUUID)] = u.UserName
data = []
for acct in rows:
key = acct.UserUUID_id if hasattr(acct, 'UserUUID_id') else acct.UserUUID
data.append({
'UserGiteaAccountUUID': UUIDObj(acct.UserGiteaAccountUUID).str,
'UserName': user_map.get(key, '') or '',
'GiteaLogin': acct.GiteaLogin,
'Email': acct.Email or '',
'Status': acct.Status,
'CreateTime': acct.CreateTime.isoformat() if acct.CreateTime else None,
})
return Response({'code': 0, 'data': {'list': data, 'total': len(data)}})
class GiteaDisableView(ELTAuthenticatedAPIView):
"""管理员:禁用账号(删除 Gitea 用户并置状态为 0)。"""
permission_classes = [IsAuthenticated, IsAdminUser]
def post(self, request):
acct_uuid = request.data.get('UserGiteaAccountUUID')
acct = None
if acct_uuid:
try:
acct = UserGiteaAccount.objects.filter(UserGiteaAccountUUID=UUIDObj(acct_uuid).bytes).first()
except (ValueError, AttributeError):
acct = None
if acct is None:
return Response({'code': 1, 'msg': '记录不存在'}, status=404)
if acct.Status == 0:
return Response({'code': 1, 'msg': '该账号已禁用'}, status=400)
try:
GiteaApiClient.DeleteUser(acct.GiteaLogin)
except GiteaError as e:
logger.error('[Gitea] 删除用户失败: %s', e)
return Response({'code': 1, 'msg': f'删除 Gitea 用户失败: {e}'}, status=502)
acct.Status = 0
acct.save(update_fields=['Status'])
return Response({'code': 0, 'msg': '已禁用'})
+9
View File
@@ -173,6 +173,15 @@ urlpatterns = [
path('API/Mail/AdminList', gerp_views.MailAdminListView.as_view(), name='mail_admin_list'),
path('API/Mail/Disable', gerp_views.MailDisableView.as_view(), name='mail_disable'),
# Gitea 代码托管对接(员工申请 + 管理员管理)
path('API/Gitea/Session', gerp_views.GiteaSessionView.as_view(), name='gitea_session'),
path('API/Gitea/Apply', gerp_views.GiteaApplyView.as_view(), name='gitea_apply'),
path('API/Gitea/MyAccount', gerp_views.GiteaMyAccountView.as_view(), name='gitea_my_account'),
path('API/Gitea/ChangePassword', gerp_views.GiteaChangePasswordView.as_view(), name='gitea_change_password'),
path('API/Gitea/Avatar', gerp_views.GiteaAvatarView.as_view(), name='gitea_avatar'),
path('API/Gitea/AdminList', gerp_views.GiteaAdminListView.as_view(), name='gitea_admin_list'),
path('API/Gitea/Disable', gerp_views.GiteaDisableView.as_view(), name='gitea_disable'),
path('API/doc/doc/GetDocs', gerp_views.DocView.as_view(), name='doc_list'),
path('API/doc/doc/GetDoc', gerp_views.DocView.as_view(), name='doc_get'),
path('API/doc/doc/AddDoc', gerp_views.DocView.as_view(), name='doc_add'),