From 7dab803ef3a7156438d8978a839d06333a4575db Mon Sep 17 00:00:00 2001 From: TermiNexus Date: Sun, 9 Aug 2026 18:28:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(gitea):=20=E9=93=B6=E6=B2=B3=E9=94=8B?= =?UTF-8?q?=E9=94=90=20Gitea=20=E5=AF=B9=E6=8E=A5=EF=BC=88=E7=94=B3?= =?UTF-8?q?=E8=AF=B7/=E8=AE=A4=E9=A2=86/=E6=94=B9=E5=AF=86/=E5=A4=B4?= =?UTF-8?q?=E5=83=8F=E8=BF=81=E7=A7=BB=20API=20+=20UserGiteaAccount?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- TempLate/GERP/index.html | 4 +- gvsds/apps/gerp/gitea.py | 216 +++++++++++ .../migrations/0021_user_gitea_account.py | 35 ++ .../gerp/migrations/0022_user_gitea_email.py | 20 ++ gvsds/apps/gerp/models.py | 26 ++ gvsds/apps/gerp/views/__init__.py | 9 + gvsds/apps/gerp/views/gitea.py | 336 ++++++++++++++++++ gvsds/urls.py | 9 + 8 files changed, 653 insertions(+), 2 deletions(-) create mode 100644 gvsds/apps/gerp/gitea.py create mode 100644 gvsds/apps/gerp/migrations/0021_user_gitea_account.py create mode 100644 gvsds/apps/gerp/migrations/0022_user_gitea_email.py create mode 100644 gvsds/apps/gerp/views/gitea.py diff --git a/TempLate/GERP/index.html b/TempLate/GERP/index.html index e76bf349..db286d60 100644 --- a/TempLate/GERP/index.html +++ b/TempLate/GERP/index.html @@ -5,7 +5,7 @@ GERP 管理系统 - + @@ -14,7 +14,7 @@ - +
diff --git a/gvsds/apps/gerp/gitea.py b/gvsds/apps/gerp/gitea.py new file mode 100644 index 00000000..5f55e91b --- /dev/null +++ b/gvsds/apps/gerp/gitea.py @@ -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/（不存在返回 404） +- 删除用户：DELETE /api/v1/admin/users/ +- 加入组织：Gitea 无直接添加成员接口，须通过团队 + （PUT /api/v1/teams//members/，团队成员自动成为组织成员） +- 组织团队：GET /api/v1/orgs//teams → [{id,name,...}] +- 创建团队：POST /api/v1/orgs//teams → {"name":"Members","permission":"read"} +认证：Authorization: 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": ""}（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(), + } diff --git a/gvsds/apps/gerp/migrations/0021_user_gitea_account.py b/gvsds/apps/gerp/migrations/0021_user_gitea_account.py new file mode 100644 index 00000000..f93cc649 --- /dev/null +++ b/gvsds/apps/gerp/migrations/0021_user_gitea_account.py @@ -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'], + }, + ), + ] diff --git a/gvsds/apps/gerp/migrations/0022_user_gitea_email.py b/gvsds/apps/gerp/migrations/0022_user_gitea_email.py new file mode 100644 index 00000000..e90ce3d5 --- /dev/null +++ b/gvsds/apps/gerp/migrations/0022_user_gitea_email.py @@ -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), + ), + ] diff --git a/gvsds/apps/gerp/models.py b/gvsds/apps/gerp/models.py index 8d9421e1..e8cae6d9 100644 --- a/gvsds/apps/gerp/models.py +++ b/gvsds/apps/gerp/models.py @@ -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 和关联属性访问 diff --git a/gvsds/apps/gerp/views/__init__.py b/gvsds/apps/gerp/views/__init__.py index a9c9ef68..6a6b7c82 100644 --- a/gvsds/apps/gerp/views/__init__.py +++ b/gvsds/apps/gerp/views/__init__.py @@ -135,3 +135,12 @@ from .mail import ( MailAdminListView, MailDisableView, ) +from .gitea import ( + GiteaSessionView, + GiteaApplyView, + GiteaMyAccountView, + GiteaChangePasswordView, + GiteaAvatarView, + GiteaAdminListView, + GiteaDisableView, +) diff --git a/gvsds/apps/gerp/views/gitea.py b/gvsds/apps/gerp/views/gitea.py new file mode 100644 index 00000000..c9befad6 --- /dev/null +++ b/gvsds/apps/gerp/views/gitea.py @@ -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': '已禁用'}) diff --git a/gvsds/urls.py b/gvsds/urls.py index 7dbae48e..fe006276 100644 --- a/gvsds/urls.py +++ b/gvsds/urls.py @@ -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'),