feat(gerp/system): GERP系统一级菜单(admin:super 专属)——状态/时序/secrets.ini/运行时 API + 菜单过滤支持 admin:super 专属

This commit is contained in:
TermiNexus committed 2026-08-13 04:20:30 +08:00
1 parent 8e428c1beb
commit d01a01bf11
7 files changed
+436 -18

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-CaVjQT8f.js"></script>
<script type="module" crossorigin src="/assets/index-QPnCOSVt.js"></script>
<link rel="modulepreload" crossorigin href="/assets/rolldown-runtime-hePW80VL.js">
<link rel="modulepreload" crossorigin href="/assets/element-plus-DtMR4CIi.js">
<link rel="modulepreload" crossorigin href="/assets/utils-C0vl14oO.js">
@@ -14,7 +14,7 @@
<link rel="modulepreload" crossorigin href="/assets/formkit-DL7Nej8X.js">
<link rel="modulepreload" crossorigin href="/assets/vue-vendor-DmO0PmrK.js">
<link rel="modulepreload" crossorigin href="/assets/serverScope-C6yhSV7-.js">
<link rel="stylesheet" crossorigin href="/assets/style-WjO5ZuLV.css">
<link rel="stylesheet" crossorigin href="/assets/style-C_CV4J6x.css">
</head>
<body>
<div id="app"></div>
+2 -2
View File
@@ -5,8 +5,8 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>银河锋锐软件开发工作室</title>
<link rel="icon" href="/assets/icon/ICON.png">
<script type="module" crossorigin src="/GVSite/index-CNA2BKwV.js"></script>
<link rel="stylesheet" crossorigin href="/GVSite/index-PmELqBcz.css">
<script type="module" crossorigin src="/GVSite/index-CBG06wf1.js"></script>
<link rel="stylesheet" crossorigin href="/GVSite/index-B89p3pTo.css">
</head>
<body>
<div id="app"></div>
+2 -2
View File
@@ -5,8 +5,8 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>银河锋锐软件开发工作室</title>
<link rel="icon" href="/assets/icon/ICON.png">
<script type="module" crossorigin src="/assets/index-D_44NdM4.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-juK_urpS.css">
<script type="module" crossorigin src="/GVSite/index-BlhJuZy4.js"></script>
<link rel="stylesheet" crossorigin href="/GVSite/index-BikM9NQE.css">
</head>
<body>
<div id="app"></div>
+7
View File
@@ -155,3 +155,10 @@ from .app import (
AppApproveView,
AppPendingView,
)
from .system_ops import (
GERPSystemStatusView,
GERPSystemSchedulesView,
GERPSystemScheduleTriggerView,
GERPSystemSecretsView,
GERPSystemRuntimeView,
)
+24 -12
View File
@@ -120,19 +120,31 @@ class MenuListCompatView(ELTAuthenticatedAPIView):
menu_map.pop(uuid_str, None)
# 权限过滤:父节点无权限则整个子树隐藏(避免子节点提升为根导致菜单结构混乱)
if not is_admin:
no_perm_uuids = set()
for uuid_str, item in menu_map.items():
perm_level = item['PermLevel']
perm_attr_str = item['PermAttr']
perm_attrs = [a.strip() for a in perm_attr_str.split(';') if a.strip()] if perm_attr_str else []
if perm_level > 0 and perm_attrs:
try:
if not PermissionVerifier(perm_level, perm_attrs, user, request):
no_perm_uuids.add(uuid_str)
except Exception:
logger.exception('MenuListCompatView 权限校验异常 uuid=%s', uuid_str)
# admin:super 专属菜单(如 GERP系统)对所有用户生效:即使 is_admin(admin:tenant)
# 也要按 IsSuperAdmin 校验,防止租户管理员看到系统级管理入口。
no_perm_uuids = set()
for uuid_str, item in menu_map.items():
perm_level = item['PermLevel']
perm_attr_str = item['PermAttr']
perm_attrs = [a.strip() for a in perm_attr_str.split(';') if a.strip()] if perm_attr_str else []
if not perm_attrs:
continue
if 'admin:super' in perm_attrs:
from gvsdsdk.auth.admin_service import IsSuperAdmin
try:
if not IsSuperAdmin(request=request, user=user, role_uuid=role_uuid_bytes):
no_perm_uuids.add(uuid_str)
except Exception:
logger.exception('MenuListCompatView admin:super 菜单校验异常 uuid=%s', uuid_str)
no_perm_uuids.add(uuid_str)
continue
if not is_admin and perm_level > 0:
try:
if not PermissionVerifier(perm_level, perm_attrs, user, request):
no_perm_uuids.add(uuid_str)
except Exception:
logger.exception('MenuListCompatView 权限校验异常 uuid=%s', uuid_str)
no_perm_uuids.add(uuid_str)
# 收集所有无权限节点的子孙节点
def _CollectDescendants(parent_uuid, acc):
for uid, item in menu_map.items():
+392
View File
@@ -0,0 +1,392 @@
"""模块:GERP系统 运维管理 API
系统管理员(admin:super)专属的系统级运维能力——只放"所有人现在都管不了的东西":
1. 系统状态:运行时间、进程、版本、Celery / Redis / 数据库健康
2. 系统时序:Celery beat 调度列表 + 手动触发
3. secrets.ini:配置查看(脱敏)与键值编辑
4. 运行时调整:日志级别热切换(持久化到 app.ini)
所有接口均要求当前角色拥有 admin:super 权限(跨租户管理员),
admin:tenant(租户内管理员)与普通用户一律拒绝。
所有命名遵循帕斯卡命名法。
"""
import logging
import platform
import time
from pathlib import Path
from rest_framework.response import Response
from . import ELTAuthenticatedAPIView
logger = logging.getLogger(__name__)
# secrets.ini / app.ini 位于项目根目录(与 gvsds/secrets.py 的定位一致)
_BASE_DIR = Path(__file__).resolve().parent.parent.parent.parent.parent
_SECRETS_FILE = _BASE_DIR / 'secrets.ini'
_APP_INI_FILE = _BASE_DIR / 'app.ini'
# 敏感键名判定:键名包含以下关键字时,读取值一律脱敏为占位符
_SENSITIVE_KEYWORDS = ('password', 'secret', 'key', 'token', 'private')
# 脱敏占位符:POST 提交该值时视为"未修改",跳过写入
_MASK = '******'
# 可手动触发的系统定时任务注册表(与 gvsds/settings.py CELERY_BEAT_SCHEDULE 对应)
_TRIGGERABLE_TASKS = {
'ResetDailyStats': 'gvsds.apps.stats.tasks.ResetDailyStats',
'ResetMonthlyStats': 'gvsds.apps.stats.tasks.ResetMonthlyStats',
'PurgeOldStats': 'gvsds.apps.stats.tasks.PurgeOldStats',
'BuildDailyRankings': 'gvsds.apps.stats.tasks.BuildDailyRankings',
}
def _IsSuperAdmin(request):
"""判断当前用户是否为系统管理员(仅 admin:super)"""
from gvsdsdk.auth.admin_service import IsSuperAdmin
return IsSuperAdmin(request)
def _Denied():
return Response({'code': 403, 'msg': '仅系统管理员(admin:super)可访问', 'data': None}, status=403)
def _IsSensitiveKey(key):
"""键名是否敏感(值需脱敏显示)"""
k = (key or '').lower()
return any(kw in k for kw in _SENSITIVE_KEYWORDS)
def _ReadIniStructure(path):
"""读取 ini 文件为 {section: {key: value}},保持文件内顺序"""
import configparser
cfg = configparser.ConfigParser()
if path.exists():
cfg.read(str(path), encoding='utf-8')
result = {}
for section in cfg.sections():
result[section] = dict(cfg.items(section))
return result
def _UpdateIniFile(path, updates):
"""行级原位更新 ini 文件(保留注释与格式),返回实际更新的键列表
updates: {section: {key: value}}
- 键已存在:替换该行
- 键不存在:追加到所属 section 末尾(不新建重复 section)
"""
if not path.exists():
return [], '配置文件不存在'
lines = path.read_text(encoding='utf-8').splitlines(keepends=True)
# 第一遍:扫描各 section 的行区间 [start, end)
sections_range = {} # section -> (start_idx, end_idx)
current = None
start = 0
for i, line in enumerate(lines):
s = line.strip()
if s.startswith('[') and s.endswith(']'):
if current is not None:
sections_range[current] = (start, i)
current = s[1:-1].strip()
start = i
if current is not None:
sections_range[current] = (start, len(lines))
# 待更新集合:排除脱敏占位符(未修改)
pending = {}
for section, kv in updates.items():
for key, value in kv.items():
if isinstance(value, str) and value.strip() == _MASK:
continue
if key.strip():
pending[(str(section).strip(), key.strip())] = value
updated_keys = []
result_lines = []
current_section = None
for line in lines:
s = line.strip()
if s.startswith('[') and s.endswith(']'):
current_section = s[1:-1].strip()
result_lines.append(line)
continue
# 键值行(跳过注释与空行)
if current_section and '=' in s and not s.startswith(('#', ';')):
key = s.split('=', 1)[0].strip()
if (current_section, key) in pending:
result_lines.append(f'{key} = {pending.pop((current_section, key))}\n')
updated_keys.append(f'{current_section}.{key}')
continue
result_lines.append(line)
# 剩余未匹配的键:追加到所属 section 末尾
if pending:
# 按 section 分组
by_section = {}
for (section, key), value in pending.items():
by_section.setdefault(section, []).append((key, value))
# 从后往前插入,保证顺序
for section, kv_list in by_section.items():
if section in sections_range:
_, end = sections_range[section]
insert_idx = end
# 跳过 section 末尾的注释/空行,插到最后一个内容行后
while insert_idx > 0 and (not result_lines[insert_idx - 1].strip()
or result_lines[insert_idx - 1].lstrip().startswith(('#', ';'))):
insert_idx -= 1
for key, value in kv_list:
result_lines.insert(insert_idx, f'{key} = {value}\n')
insert_idx += 1
updated_keys.append(f'{section}.{key}')
else:
# section 不存在:追加新 section 块
result_lines.append(f'\n[{section}]\n')
for key, value in kv_list:
result_lines.append(f'{key} = {value}\n')
updated_keys.append(f'{section}.{key}')
path.write_text(''.join(result_lines), encoding='utf-8')
return updated_keys, None
class GERPSystemStatusView(ELTAuthenticatedAPIView):
"""系统状态总览——运行时间、进程、版本、依赖服务健康
GET /API/GERPSystem/Status
"""
def get(self, request):
if not _IsSuperAdmin(request):
return _Denied()
import django
import psutil
from django.db import connection
proc = psutil.Process()
data = {
'Host': platform.node(),
'PID': proc.pid,
'StartTime': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(proc.create_time())),
'UptimeSeconds': int(time.time() - proc.create_time()),
'BootTime': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(psutil.boot_time())),
'Python': platform.python_version(),
'Django': django.get_version(),
'Gvsdsdk': '',
'MemoryMB': round(proc.memory_info().rss / 1024 / 1024, 1),
'CPUPercent': proc.cpu_percent(interval=0.1),
'Redis': 'unknown',
'Database': 'unknown',
'Celery': 'disabled',
'LogLevel': logging.getLogger('gvsds').getEffectiveLevel(),
}
try:
import gvsdsdk
data['Gvsdsdk'] = gvsdsdk.get_version_info().get('Version', '')
except Exception:
pass
# Redis 健康
try:
from django.conf import settings as _s
import redis as _redis
_r = _redis.Redis(
host=_s.REDIS_HOST, port=_s.REDIS_PORT, db=_s.REDIS_DB,
password=_s.REDIS_PASSWORD or None, socket_timeout=2,
)
data['Redis'] = 'ok' if _r.ping() else 'error'
except Exception:
data['Redis'] = 'error'
# 数据库健康
try:
with connection.cursor() as cur:
cur.execute('SELECT 1')
data['Database'] = 'ok'
except Exception:
data['Database'] = 'error'
# Celery 启用状态
try:
from django.conf import settings as _s
if getattr(_s, 'CELERY_BROKER_URL', None):
data['Celery'] = 'enabled'
except Exception:
pass
# 日志级别转为名称
data['LogLevel'] = logging.getLevelName(data['LogLevel'])
return Response({'code': 0, 'data': data})
class GERPSystemSchedulesView(ELTAuthenticatedAPIView):
"""系统时序——Celery beat 调度列表
GET /API/GERPSystem/Schedules
"""
def get(self, request):
if not _IsSuperAdmin(request):
return _Denied()
schedules = []
try:
from django.conf import settings as _s
beat = getattr(_s, 'CELERY_BEAT_SCHEDULE', None) or {}
for name, cfg in beat.items():
schedules.append({
'Name': name,
'Task': cfg.get('task', ''),
'Schedule': str(cfg.get('schedule', '')),
'Enabled': True,
})
except Exception:
logger.exception('GERPSystemSchedulesView 读取 beat schedule 失败')
triggerable = [
{'Name': n, 'Task': t} for n, t in _TRIGGERABLE_TASKS.items()
]
return Response({'code': 0, 'data': {'Schedules': schedules, 'Triggerable': triggerable}})
class GERPSystemScheduleTriggerView(ELTAuthenticatedAPIView):
"""手动触发系统定时任务
POST /API/GERPSystem/Schedules/Trigger Body: {Name: ResetDailyStats}
- 有 Celery:异步投递(delay)
- 无 Celery:同步执行(apply)
"""
def post(self, request):
if not _IsSuperAdmin(request):
return _Denied()
name = (request.data.get('Name') or '').strip()
task_path = _TRIGGERABLE_TASKS.get(name)
if not task_path:
return Response({'code': 1, 'msg': f'未知任务 {name}', 'data': None}, status=400)
try:
module_path, func_name = task_path.rsplit('.', 1)
import importlib
module = importlib.import_module(module_path)
task = getattr(module, func_name)
from django.conf import settings as _s
if getattr(_s, 'CELERY_BROKER_URL', None) and getattr(task, 'delay', None):
result = task.delay()
return Response({'code': 0, 'msg': f'任务 {name} 已投递(async)', 'data': {'Async': True}})
result = task.apply()
detail = result.result if result is not None else None
return Response({'code': 0, 'msg': f'任务 {name} 执行完成', 'data': {'Async': False, 'Result': detail}})
except Exception:
logger.exception('GERPSystemScheduleTriggerView 触发任务失败 name=%s', name)
return Response({'code': 1, 'msg': '任务执行异常,请查看服务器日志', 'data': None}, status=500)
class GERPSystemSecretsView(ELTAuthenticatedAPIView):
"""secrets.ini 配置管理(脱敏读取 / 键值编辑)
GET /API/GERPSystem/Secrets — 返回全部 section 与键,敏感键值脱敏
POST /API/GERPSystem/Secrets — Body: {Sections: {section: {key: value}}}
value 为 '******' 的键视为未修改,跳过写入
"""
def get(self, request):
if not _IsSuperAdmin(request):
return _Denied()
structure = _ReadIniStructure(_SECRETS_FILE)
sections = {}
for section, kv in structure.items():
sections[section] = {
key: {
'Value': _MASK if _IsSensitiveKey(key) else value,
'Masked': _IsSensitiveKey(key),
}
for key, value in kv.items()
}
return Response({
'code': 0,
'data': {
'FilePath': str(_SECRETS_FILE),
'Sections': sections,
},
})
def post(self, request):
if not _IsSuperAdmin(request):
return _Denied()
updates = request.data.get('Sections') or {}
if not isinstance(updates, dict) or not updates:
return Response({'code': 1, 'msg': '缺少 Sections 参数', 'data': None}, status=400)
updated_keys, err = _UpdateIniFile(_SECRETS_FILE, updates)
if err:
return Response({'code': 1, 'msg': err, 'data': None}, status=400)
# 热重载 secrets 配置缓存,使新增/修改立即可被后续读取
try:
from gvsds.secrets import reload as _reload_secrets
_reload_secrets()
except Exception:
logger.exception('secrets.ini 重载失败')
return Response({'code': 0, 'msg': '配置已保存', 'data': {'Updated': updated_keys}})
class GERPSystemRuntimeView(ELTAuthenticatedAPIView):
"""运行时调整——日志级别热切换(持久化到 app.ini [logging] LEVEL)
GET /API/GERPSystem/Runtime — 当前日志级别
POST /API/GERPSystem/Runtime — Body: {LogLevel: DEBUG|INFO|WARNING|ERROR}
"""
_LEVELS = ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL')
def get(self, request):
if not _IsSuperAdmin(request):
return _Denied()
current = logging.getLogger('gvsds').getEffectiveLevel()
# app.ini 持久化值(未配置时取运行值)
persisted = None
try:
from gvsds import app_config
persisted = app_config.get('logging', 'LEVEL', fallback=None)
except Exception:
pass
return Response({
'code': 0,
'data': {
'Current': logging.getLevelName(current),
'Persisted': persisted,
'Available': list(self._LEVELS),
},
})
def post(self, request):
if not _IsSuperAdmin(request):
return _Denied()
level = (request.data.get('LogLevel') or '').strip().upper()
if level not in self._LEVELS:
return Response({'code': 1, 'msg': f'日志级别无效,可选:{" / ".join(self._LEVELS)}', 'data': None}, status=400)
# 热切换:gvsds 命名空间下全部 logger 生效(root 级联)
logging.getLogger('gvsds').setLevel(level)
logging.getLogger().setLevel(level)
# 持久化到 app.ini [logging] LEVEL + 热重载
try:
from gvsds import app_config
updated, err = _UpdateIniFile(_APP_INI_FILE, {'logging': {'LEVEL': level}})
if err:
return Response({'code': 1, 'msg': f'运行时已切换,但持久化失败:{err}', 'data': None}, status=500)
app_config.reload()
except Exception:
logger.exception('日志级别持久化失败')
return Response({'code': 0, 'msg': f'日志级别已切换为 {level}', 'data': {'LogLevel': level}})
+7
View File
@@ -194,6 +194,13 @@ urlpatterns = [
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'),
# GERP系统 运维管理(仅系统管理员 admin:super)
path('API/GERPSystem/Status', gerp_views.GERPSystemStatusView.as_view(), name='gerp_system_status'),
path('API/GERPSystem/Schedules', gerp_views.GERPSystemSchedulesView.as_view(), name='gerp_system_schedules'),
path('API/GERPSystem/Schedules/Trigger', gerp_views.GERPSystemScheduleTriggerView.as_view(), name='gerp_system_trigger'),
path('API/GERPSystem/Secrets', gerp_views.GERPSystemSecretsView.as_view(), name='gerp_system_secrets'),
path('API/GERPSystem/Runtime', gerp_views.GERPSystemRuntimeView.as_view(), name='gerp_system_runtime'),
# Gitea 免密登录(OIDC Provider)
path('.well-known/openid-configuration', gerp_oidc.OIDCDiscoveryView.as_view(), name='oidc_discovery'),
path('oidc/authorize', gerp_oidc.OIDCAuthorizeView.as_view(), name='oidc_authorize'),