325 lines
13 KiB
Python
325 lines
13 KiB
Python
import json
|
|
import shutil
|
|
import os
|
|
import threading
|
|
from app.views.ViewsBase import *
|
|
from django.shortcuts import render
|
|
from app.utils.OSSystem import OSSystem
|
|
from app.utils.GlobalUtils import g_filepath_settings_json, GlobalUtils
|
|
from framework.settings import PROJECT_VERSION, PROJECT_FLAG, PROJECT_UA, PROJECT_BUILT, PROJECT_ADMIN_START_TIMESTAMP
|
|
|
|
def f_readSettings(lang):
|
|
try:
|
|
for encoding in ["utf-8", "gbk"]:
|
|
try:
|
|
with open(g_filepath_settings_json, 'r', encoding=encoding) as f:
|
|
data = json.load(f)
|
|
# 从 languages 字典中直接获取对应语言的 oem 配置
|
|
languages = data.get("languages", {})
|
|
return languages.get(lang, {}).get("oem", {})
|
|
except Exception as e:
|
|
g_logger.error("f_readSettings() error: %s" % str(e))
|
|
return {}
|
|
except:
|
|
return {}
|
|
|
|
def f_writeSettings(lang, settings_data):
|
|
for encoding in ["utf-8", "gbk"]:
|
|
try:
|
|
with open(g_filepath_settings_json, 'r', encoding=encoding) as f:
|
|
data = json.load(f)
|
|
|
|
# 更新 languages 字典中对应语言的 oem 配置
|
|
languages = data.get("languages", {})
|
|
if lang in languages:
|
|
languages[lang]["oem"] = settings_data
|
|
|
|
with open(g_filepath_settings_json, 'w', encoding=encoding) as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
return True
|
|
except Exception as e:
|
|
g_logger.error("f_writeSettings() error: %s"%str(e))
|
|
continue
|
|
|
|
def config(request):
|
|
lang = f_parseRequestLang(request)
|
|
branding = f_readSettings(lang)
|
|
cfg = g_config.to_dict()
|
|
context = {
|
|
"config": cfg,
|
|
"branding": branding
|
|
}
|
|
return render(request, 'app/system/config.html', context)
|
|
|
|
|
|
def _config_api_payload():
|
|
data = g_config.to_dict()
|
|
# 前端习惯 0/1 开关
|
|
for k in (
|
|
"autoAddStreamProxy", "isEnableLoginCaptcha", "logDebug", "isEnableUpdatePopup",
|
|
"telemetryEnabled", "updateCheckEnabled",
|
|
"isEnableMediaProxyRtmp", "autoStartMedia", "analysisSharedInference",
|
|
"recordingEnabled",
|
|
):
|
|
if k in data:
|
|
data[k] = 1 if _bool_web(data.get(k)) else 0
|
|
sip = data.get("sipServer") or {}
|
|
if "autoInviteAfterRecCateLog" in sip:
|
|
sip["autoInviteAfterRecCateLog"] = 1 if _bool_web(sip.get("autoInviteAfterRecCateLog")) else 0
|
|
data["sipServer"] = sip
|
|
return data
|
|
|
|
|
|
def _bool_web(v):
|
|
return v is True or v == 1 or str(v).strip().lower() in ("1", "true", "yes")
|
|
|
|
|
|
def api_openConfig(request):
|
|
ret = False
|
|
msg = LANG_VIEWS_T(request, "msg_unknown_error")
|
|
|
|
if request.method == 'GET':
|
|
# 加载配置：同时返回基础配置 + OEM 品牌信息
|
|
__check_ret, __check_msg = f_checkRequestSafe(request)
|
|
if __check_ret:
|
|
try:
|
|
lang = f_parseRequestLang(request)
|
|
branding = f_readSettings(lang)
|
|
data = _config_api_payload()
|
|
ret = True
|
|
msg = LANG_VIEWS_T(request, "msg_success")
|
|
res = {
|
|
"code": 1000,
|
|
"msg": msg,
|
|
"data": data,
|
|
"config": data,
|
|
"branding": branding
|
|
}
|
|
g_logger.info("SystemView.openConfig(GET) ok")
|
|
return f_responseJson(res)
|
|
except Exception as e:
|
|
msg = str(e)
|
|
else:
|
|
msg = __check_msg
|
|
|
|
elif request.method == 'POST':
|
|
__check_ret, __check_msg = f_checkRequestSafe(request)
|
|
if __check_ret:
|
|
params = f_parsePostParams(request)
|
|
g_logger.info("SystemView.openConfig() requested")
|
|
|
|
try:
|
|
before = g_config.to_dict(include_secrets=True)
|
|
g_config.save_from_web(params)
|
|
notes = GlobalUtils.apply_runtime_config(before)
|
|
|
|
ret = True
|
|
msg = LANG_VIEWS_T(request, "syscfg_save_success")
|
|
hint_parts = []
|
|
if "adminPort" in notes:
|
|
hint_parts.append(LANG_VIEWS_T(request, "syscfg_hint_admin_port"))
|
|
if "logDebug" in notes:
|
|
hint_parts.append(LANG_VIEWS_T(request, "syscfg_hint_log_debug"))
|
|
if "zlm" in notes:
|
|
hint_parts.append(LANG_VIEWS_T(request, "syscfg_hint_zlm"))
|
|
if hint_parts:
|
|
msg = msg + " " + " ".join(hint_parts)
|
|
|
|
except Exception as e:
|
|
msg = str(e)
|
|
else:
|
|
msg = __check_msg
|
|
else:
|
|
msg = LANG_VIEWS_T(request, "msg_method_not_supported")
|
|
|
|
res = {
|
|
"code": 1000 if ret else 0,
|
|
"msg": msg
|
|
}
|
|
g_logger.info("SystemView.openConfig() res: %s" % str(res))
|
|
return f_responseJson(res)
|
|
|
|
def api_openSaveSettings(request):
|
|
ret = False
|
|
msg = LANG_VIEWS_T(request, "msg_unknown_error")
|
|
|
|
if request.method == 'POST':
|
|
__check_ret, __check_msg = f_checkRequestSafe(request)
|
|
if __check_ret:
|
|
params = f_parsePostParams(request)
|
|
g_logger.info("SystemView.openSaveSettings() params: %s" % str(params))
|
|
|
|
try:
|
|
lang = f_parseRequestLang(request)
|
|
|
|
branding_name = str(params.get("name", "")).strip()
|
|
branding_welcome = str(params.get("welcome", "")).strip()
|
|
branding_logo_url = str(params.get("logo_url", "")).strip()
|
|
branding_bottom_name = str(params.get("bottom_name", "")).strip()
|
|
branding_author = str(params.get("author", "")).strip()
|
|
branding_author_link = str(params.get("author_link", "")).strip()
|
|
branding_check_version_download_url = str(params.get("check_version_download_url", "")).strip()
|
|
branding_is_show_author = params.get("is_show_author", False)
|
|
if isinstance(branding_is_show_author, str):
|
|
branding_is_show_author = branding_is_show_author.lower() in ['true', '1', 'yes']
|
|
|
|
settings_data = {
|
|
"name": branding_name,
|
|
"welcome": branding_welcome,
|
|
"logo_url": branding_logo_url,
|
|
"bottom_name": branding_bottom_name,
|
|
"is_show_author": branding_is_show_author,
|
|
"author": branding_author,
|
|
"author_link": branding_author_link,
|
|
"check_version_download_url": branding_check_version_download_url
|
|
}
|
|
|
|
f_writeSettings(lang, settings_data)
|
|
ret = True
|
|
msg = LANG_VIEWS_T(request, "syscfg_save_oem_success")
|
|
|
|
except Exception as e:
|
|
msg = str(e)
|
|
else:
|
|
msg = __check_msg
|
|
else:
|
|
msg = LANG_VIEWS_T(request, "msg_method_not_supported")
|
|
|
|
res = {
|
|
"code": 1000 if ret else 0,
|
|
"msg": msg
|
|
}
|
|
g_logger.info("SystemView.openSaveSettings() res: %s" % str(res))
|
|
return f_responseJson(res)
|
|
|
|
def api_openExportLogs(request):
|
|
# 导出日志
|
|
ret = False
|
|
msg = LANG_VIEWS_T(request, "msg_unknown_error")
|
|
info = {
|
|
|
|
}
|
|
request_ip = f_parseRequestIp(request)
|
|
lang = f_parseRequestLang(request)
|
|
|
|
export_dir = None
|
|
if request.method == 'POST':
|
|
__check_ret, __check_msg = f_checkRequestSafe(request)
|
|
if __check_ret:
|
|
try:
|
|
params = f_parsePostParams(request)
|
|
g_logger.info("SystemView.openExportLogs() params:%s" % str(params))
|
|
|
|
g_gb28181SipServer.log_status()
|
|
run_errors = []
|
|
|
|
# 导出外层文件夹
|
|
export_dirname = "logs%s-%s-%s" % (PROJECT_VERSION, PROJECT_FLAG, datetime.now().strftime("%Y%m%d%H%M%S"))
|
|
export_dir = os.path.join(g_config.storageTempDir, export_dirname)
|
|
export_filename = "%s.xclogs" % export_dirname
|
|
export_filepath = os.path.join(g_config.storageTempDir, export_filename)
|
|
|
|
if not os.path.exists(export_dir):
|
|
os.makedirs(export_dir)
|
|
|
|
osSystem = OSSystem()
|
|
|
|
|
|
try:
|
|
# 压缩log文件夹->log.tar
|
|
log_dir = os.path.join(BASE_DIR, "log")
|
|
if os.path.exists(log_dir):
|
|
log_tar_filepath = os.path.join(export_dir, "log")
|
|
shutil.make_archive(log_tar_filepath, 'tar', log_dir)
|
|
except Exception as e:
|
|
run_errors.append("export log error:%s" % str(e))
|
|
|
|
# 写入config.json, settings.json
|
|
if os.path.exists(g_filepath_config_json):
|
|
dst = os.path.join(export_dir, "config.json")
|
|
shutil.copyfile(g_filepath_config_json, dst)
|
|
|
|
if os.path.exists(g_filepath_settings_json):
|
|
dst = os.path.join(export_dir, "settings.json")
|
|
shutil.copyfile(g_filepath_settings_json, dst)
|
|
|
|
|
|
if os.path.exists(g_config.mediaStartConfigPath):
|
|
dst = os.path.join(export_dir, "config.ini")
|
|
shutil.copyfile(g_config.mediaStartConfigPath, dst)
|
|
|
|
allowed_hosts_src = os.path.join(BASE_DIR, ".allowed_hosts")
|
|
if os.path.exists(allowed_hosts_src):
|
|
allowed_hosts_dst = os.path.join(export_dir, ".allowed_hosts")
|
|
shutil.copyfile(allowed_hosts_src, allowed_hosts_dst)
|
|
|
|
|
|
# 写入env.txt
|
|
export_filepath_env = os.path.join(export_dir, "env.txt")
|
|
env_f = open(export_filepath_env, 'w', encoding="utf-8")
|
|
env_f.write("name=%s\n" % PROJECT_UA)
|
|
env_f.write("built=%s\n" % PROJECT_BUILT)
|
|
env_f.write("version=%s\n" % PROJECT_VERSION)
|
|
env_f.write("flag=%s\n" % PROJECT_FLAG)
|
|
env_f.write("log_filename=%s\n" % export_filename)
|
|
env_f.write("app_start_date=%s\n" % datetime.fromtimestamp(PROJECT_ADMIN_START_TIMESTAMP).strftime('%Y-%m-%d %H:%M'))
|
|
env_f.write("current_date=%s\n" % datetime.now().strftime('%Y-%m-%d %H:%M'))
|
|
env_f.write("system=%s\n" % osSystem.getSystemName())
|
|
env_f.write("machine=%s\n" % osSystem.getMachineNode())
|
|
env_f.write("uname_a=%s\n" % osSystem.getMachineUnameA())
|
|
env_f.write("zlm.threadsLoad=%s\n" % str(g_zlm.getThreadsLoad()))
|
|
env_f.write("cpu=%s\n" % osSystem.getMachineCpu())
|
|
env_f.write("nvidia=%s\n" % osSystem.getMachineNvidia())
|
|
env_f.write("ascend=%s\n" % osSystem.getMachineAscend())
|
|
env_f.write("rknpu=%s\n" % osSystem.getMachineRknpu())
|
|
env_f.write("os=%s\n" % str(osSystem.getOSInfo()))
|
|
env_f.write("os_release=%s\n" % osSystem.getMachineOsRelease())
|
|
env_f.write("lscpu=%s\n" % osSystem.getMachineLsCpu())
|
|
env_f.close()
|
|
|
|
# 写入online.txt
|
|
export_filepath_online = os.path.join(export_dir, "online.txt")
|
|
online_f = open(export_filepath_online, 'w', encoding="utf-8")
|
|
run_info = {}
|
|
for k, v in run_info.items():
|
|
online_f.write("%s=%s\n" % (str(k), str(v)))
|
|
|
|
online_f.write("av_log=%s\n" % str(g_database.select("select * from av_log order by id desc limit 100")))
|
|
online_f.write("run_errors=%s\n" % str(run_errors))
|
|
online_f.close()
|
|
|
|
# 压缩导出文件夹
|
|
shutil.make_archive(export_filepath.replace(".xclogs", ""), 'tar', export_dir)
|
|
# 重命名为 .xclogs
|
|
tar_filepath = export_filepath.replace(".xclogs", ".tar")
|
|
if os.path.exists(tar_filepath):
|
|
shutil.move(tar_filepath, export_filepath)
|
|
|
|
from app.views.StorageView import issue_download_file_id
|
|
info["download_file_id"] = issue_download_file_id(export_filename)
|
|
info["download_filename"] = export_filename
|
|
ret = True
|
|
msg = LANG_VIEWS_T(request, "syscfg_export_log_success")
|
|
except Exception as e:
|
|
msg = str(e)
|
|
else:
|
|
msg = __check_msg
|
|
else:
|
|
msg = LANG_VIEWS_T(request, "msg_method_not_supported")
|
|
|
|
g_logger.info("export_dir=%s" % str(export_dir))
|
|
if export_dir:
|
|
try:
|
|
if os.path.exists(export_dir):
|
|
shutil.rmtree(export_dir)
|
|
except Exception as e:
|
|
g_logger.error("e=%s" % str(e))
|
|
|
|
res = {
|
|
"code": 1000 if ret else 0,
|
|
"msg": msg,
|
|
"info": info
|
|
}
|
|
g_logger.info("SystemView.openExportLogs() res:%s" % str(res))
|
|
return f_responseJson(res)
|