多功能实现的 SDK / Multifunctional SDK 
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: ForceUpdateInfo.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x15\x46orceUpdateInfo.proto\"+\n\x0f\x46orceUpdateInfo\x12\x18\n\x10\x66orce_update_url\x18\x01 \x01(\tB\x16\n\x14org.sorapointa.protob\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'ForceUpdateInfo_pb2', _globals)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  DESCRIPTOR._serialized_options = b'\n\024org.sorapointa.proto'
  _globals['_FORCEUPDATEINFO']._serialized_start=25
  _globals['_FORCEUPDATEINFO']._serialized_end=68
# @@protoc_insertion_point(module_scope)

data/proto/ForceUpdateInfo_pb2.py
CokeSR/Hk4e-SDKSever-Py
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: QueryCurrRegionHttpRsp.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()


import RegionInfo_pb2 as RegionInfo__pb2
import ForceUpdateInfo_pb2 as ForceUpdateInfo__pb2
import StopServerInfo_pb2 as StopServerInfo__pb2


DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1cQueryCurrRegionHttpRsp.proto\x1a\x10RegionInfo.proto\x1a\x15\x46orceUpdateInfo.proto\x1a\x14StopServerInfo.proto\"\xa6\x02\n\x16QueryCurrRegionHttpRsp\x12\x0f\n\x07retcode\x18\x01 \x01(\x05\x12\x0b\n\x03msg\x18\x02 \x01(\t\x12 \n\x0bregion_info\x18\x03 \x01(\x0b\x32\x0b.RegionInfo\x12\x19\n\x11\x63lient_secret_key\x18\x0b \x01(\x0c\x12&\n\x1eregion_custom_config_encrypted\x18\x0c \x01(\x0c\x12-\n%client_region_custom_config_encrypted\x18\r \x01(\x0c\x12(\n\x0c\x66orce_udpate\x18\x04 \x01(\x0b\x32\x10.ForceUpdateInfoH\x00\x12&\n\x0bstop_server\x18\x05 \x01(\x0b\x32\x0f.StopServerInfoH\x00\x42\x08\n\x06\x64\x65tailB\x16\n\x14org.sorapointa.protob\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'QueryCurrRegionHttpRsp_pb2', _globals)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  DESCRIPTOR._serialized_options = b'\n\024org.sorapointa.proto'
  _globals['_QUERYCURRREGIONHTTPRSP']._serialized_start=96
  _globals['_QUERYCURRREGIONHTTPRSP']._serialized_end=390
# @@protoc_insertion_point(module_scope)

data/proto/QueryCurrRegionHttpRsp_pb2.py
CokeSR/Hk4e-SDKSever-Py
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: QueryRegionListHttpRsp.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()


import data.proto.RegionSimpleInfo_pb2 as RegionSimpleInfo__pb2


DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1cQueryRegionListHttpRsp.proto\x1a\x16RegionSimpleInfo.proto\"\xad\x01\n\x16QueryRegionListHttpRsp\x12\x0f\n\x07retcode\x18\x01 \x01(\x05\x12&\n\x0bregion_list\x18\x02 \x03(\x0b\x32\x11.RegionSimpleInfo\x12\x19\n\x11\x63lient_secret_key\x18\x05 \x01(\x0c\x12&\n\x1e\x63lient_custom_config_encrypted\x18\x06 \x01(\x0c\x12\x17\n\x0f\x65nable_login_pc\x18\x07 \x01(\x08\x42\x16\n\x14org.sorapointa.protob\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'QueryRegionListHttpRsp_pb2', _globals)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  DESCRIPTOR._serialized_options = b'\n\024org.sorapointa.proto'
  _globals['_QUERYREGIONLISTHTTPRSP']._serialized_start=57
  _globals['_QUERYREGIONLISTHTTPRSP']._serialized_end=230
# @@protoc_insertion_point(module_scope)

data/proto/QueryRegionListHttpRsp_pb2.py
CokeSR/Hk4e-SDKSever-Py
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: RegionInfo.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()


import ResVersionConfig_pb2 as ResVersionConfig__pb2


DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x10RegionInfo.proto\x1a\x16ResVersionConfig.proto\"\xaf\x06\n\nRegionInfo\x12\x15\n\rgateserver_ip\x18\x01 \x01(\t\x12\x17\n\x0fgateserver_port\x18\x02 \x01(\r\x12\x18\n\x10pay_callback_url\x18\x03 \x01(\t\x12\x11\n\tarea_type\x18\x07 \x01(\t\x12\x14\n\x0cresource_url\x18\x08 \x01(\t\x12\x10\n\x08\x64\x61ta_url\x18\t \x01(\t\x12\x14\n\x0c\x66\x65\x65\x64\x62\x61\x63k_url\x18\n \x01(\t\x12\x14\n\x0c\x62ulletin_url\x18\x0b \x01(\t\x12\x18\n\x10resource_url_bak\x18\x0c \x01(\t\x12\x14\n\x0c\x64\x61ta_url_bak\x18\r \x01(\t\x12\x1b\n\x13\x63lient_data_version\x18\x0e \x01(\r\x12\x14\n\x0chandbook_url\x18\x10 \x01(\t\x12#\n\x1b\x63lient_silence_data_version\x18\x12 \x01(\r\x12\x17\n\x0f\x63lient_data_md5\x18\x13 \x01(\t\x12\x1f\n\x17\x63lient_silence_data_md5\x18\x14 \x01(\t\x12-\n\x12res_version_config\x18\x16 \x01(\x0b\x32\x11.ResVersionConfig\x12\x12\n\nsecret_key\x18\x17 \x01(\x0c\x12\x1e\n\x16official_community_url\x18\x18 \x01(\t\x12\x1d\n\x15\x63lient_version_suffix\x18\x1a \x01(\t\x12%\n\x1d\x63lient_silence_version_suffix\x18\x1b \x01(\t\x12\"\n\x1ause_gateserver_domain_name\x18\x1c \x01(\x08\x12\x1e\n\x16gateserver_domain_name\x18\x1d \x01(\t\x12\x17\n\x0fuser_center_url\x18\x1e \x01(\t\x12\x18\n\x10\x61\x63\x63ount_bind_url\x18\x1f \x01(\t\x12\x11\n\tcdkey_url\x18  \x01(\t\x12\x1a\n\x12privacy_policy_url\x18! \x01(\t\x12\x19\n\x11next_resource_url\x18\" \x01(\t\x12\x32\n\x17next_res_version_config\x18# \x01(\x0b\x32\x11.ResVersionConfig\x12\x10\n\x08game_biz\x18$ \x01(\tB\x16\n\x14org.sorapointa.protob\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'RegionInfo_pb2', _globals)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  DESCRIPTOR._serialized_options = b'\n\024org.sorapointa.proto'
  _globals['_REGIONINFO']._serialized_start=45
  _globals['_REGIONINFO']._serialized_end=860
# @@protoc_insertion_point(module_scope)

data/proto/RegionInfo_pb2.py
CokeSR/Hk4e-SDKSever-Py
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: RegionSimpleInfo.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x16RegionSimpleInfo.proto\"S\n\x10RegionSimpleInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\r\n\x05title\x18\x02 \x01(\t\x12\x0c\n\x04type\x18\x03 \x01(\t\x12\x14\n\x0c\x64ispatch_url\x18\x04 \x01(\tB\x16\n\x14org.sorapointa.protob\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'RegionSimpleInfo_pb2', _globals)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  DESCRIPTOR._serialized_options = b'\n\024org.sorapointa.proto'
  _globals['_REGIONSIMPLEINFO']._serialized_start=26
  _globals['_REGIONSIMPLEINFO']._serialized_end=109
# @@protoc_insertion_point(module_scope)

data/proto/RegionSimpleInfo_pb2.py
CokeSR/Hk4e-SDKSever-Py
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: ResVersionConfig.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x16ResVersionConfig.proto\"\xa2\x01\n\x10ResVersionConfig\x12\x0f\n\x07version\x18\x01 \x01(\r\x12\x0f\n\x07relogin\x18\x02 \x01(\x08\x12\x0b\n\x03md5\x18\x03 \x01(\t\x12\x1a\n\x12release_total_size\x18\x04 \x01(\t\x12\x16\n\x0eversion_suffix\x18\x05 \x01(\t\x12\x0e\n\x06\x62ranch\x18\x06 \x01(\t\x12\x1b\n\x13next_script_version\x18\x07 \x01(\tB\x16\n\x14org.sorapointa.protob\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'ResVersionConfig_pb2', _globals)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  DESCRIPTOR._serialized_options = b'\n\024org.sorapointa.proto'
  _globals['_RESVERSIONCONFIG']._serialized_start=27
  _globals['_RESVERSIONCONFIG']._serialized_end=189
# @@protoc_insertion_point(module_scope)

data/proto/ResVersionConfig_pb2.py
CokeSR/Hk4e-SDKSever-Py
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: StopServerInfo.proto
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x14StopServerInfo.proto\"b\n\x0eStopServerInfo\x12\x17\n\x0fstop_begin_time\x18\x01 \x01(\r\x12\x15\n\rstop_end_time\x18\x02 \x01(\r\x12\x0b\n\x03url\x18\x03 \x01(\t\x12\x13\n\x0b\x63ontent_msg\x18\x04 \x01(\tB\x16\n\x14org.sorapointa.protob\x06proto3')

_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'StopServerInfo_pb2', _globals)
if _descriptor._USE_C_DESCRIPTORS == False:
  DESCRIPTOR._options = None
  DESCRIPTOR._serialized_options = b'\n\024org.sorapointa.proto'
  _globals['_STOPSERVERINFO']._serialized_start=24
  _globals['_STOPSERVERINFO']._serialized_end=122
# @@protoc_insertion_point(module_scope)

data/proto/StopServerInfo_pb2.py
CokeSR/Hk4e-SDKSever-Py
# WEBGM功能 先占位
function/NotImplemented/gamemanagecontrol.py
CokeSR/Hk4e-SDKSever-Py
# 解析https功能(openssl) 先占位
function/NotImplemented/handlehttps.py
CokeSR/Hk4e-SDKSever-Py
# 热更新服务(本地dispatch) 先占位
function/NotImplemented/hotpatchservice.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import re
import random
import string
import settings.repositories as repositories

from flask_caching import Cache
from flask_mail import Message
from settings.database import get_db
from settings.library import password_hash
from settings.loadconfig import get_config
from settings.response import json_rsp_with_msg
from flask import request, render_template, flash, current_app

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

#=====================找回密码=====================#
# 找回密码(功能不可用)
@app.route('/account/recover', methods=['GET', 'POST'])
def account_recover():
    cursor = get_db().cursor()
    cached_data = cache.get(request.form.get('email'))
    if request.method == 'POST':
        email = request.form.get('email')
        verifycode = request.form.get('verifycode')
        password = request.form.get('password')
        passwordv2 = request.form.get('passwordv2')
        if not re.fullmatch(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', email):
            flash('邮箱格式不正确', 'error')
        else:
            cursor.execute("SELECT * FROM `t_accounts` WHERE `email` = %s", (email,))
            email_exists = cursor.fetchone()
            if not email_exists:
                flash('该邮箱不存在', 'error')
            elif verifycode != cached_data and get_config()['Mail']['ENABLE']:
                flash('验证码错误', 'error')
            elif password != passwordv2:
                flash('两次输入的密码不一致', 'error')
            elif len(password) < get_config()["Security"]["min_password_len"]:
                flash(f"密码长度不能小于 {get_config()['Security']['min_password_len']} 字节", 'error')
            else:
                new_password = password_hash(password)
                cursor.execute("UPDATE `t_accounts` SET `password` = %s WHERE `email` = %s",
                               (new_password, email))
                flash('密码重置成功，请返回登录', 'success')
                cache.delete(email)
    return render_template("account/recover.tmpl")

# 邮件验证码 用于找回密码
@app.route('/account/recover_code', methods=['POST'])
def recover_code():
    email = request.form.get('email')
    email_pattern = '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$'
    if not re.match(email_pattern, email):
        return json_rsp_with_msg(repositories.RES_FAIL, "邮箱格式不正确", {})
    cursor = get_db().cursor()
    user_query = "SELECT * FROM `t_accounts` WHERE `email` = %s"
    cursor.execute(user_query, (email,))
    user = cursor.fetchone()
    if not user:
        return json_rsp_with_msg(repositories.RES_FAIL, "该邮箱不存在", {})
    reset_code = ''.join(random.choices(string.digits, k=4))
    mail = current_app.extensions['mail']
    msg = Message(f"重置密码请求", recipients=[email])
    msg.body = f"您的重置密码验证码是：{reset_code}，验证码5分钟内有效"
    try:
        mail.send(msg)
    except:
        return json_rsp_with_msg(repositories.RES_FAIL, "未知异常，请联系管理员", {})
    cache.set(email, reset_code, timeout=60 * 5)
    return json_rsp_with_msg(repositories.RES_SUCCESS, "验证码发送成功，请查收邮箱", {})
function/accountrecover.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import re
import random
import string
import settings.repositories as repositories

from time import time as epoch
from flask_mail import Message
from flask_caching import Cache
from settings.database import get_db
from settings.library import password_hash
from settings.loadconfig import get_config
from settings.response import json_rsp_with_msg
from flask import request, render_template, flash, current_app

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

#=====================注册模块=====================#
# 游戏账号注册
@app.route('/account/register', methods=['GET', 'POST'])
@app.route('/mihoyo/common/accountSystemSandboxFE/index.html', methods=['GET', 'POST'])         # 国内沙箱 注册和找回URL是同一个
def account_register():
    cursor = get_db().cursor()
    cached_data = cache.get(request.form.get('email'))
    if request.method == 'POST':
        username = request.form.get('username')
        mobile = request.form.get('mobile')
        email = request.form.get('email')
        verifycode = request.form.get('verifycode')
        password = request.form.get('password')
        passwordv2 = request.form.get('passwordv2')
        cursor.execute("SELECT * FROM `t_accounts` WHERE `name` = %s", (username,))
        user = cursor.fetchone()
        if user:
            flash('您准备注册的用户名已被使用', 'error')
        cursor.execute("SELECT * FROM `t_accounts` WHERE `mobile` = %s", (mobile,))
        phone_number = cursor.fetchone()
        if phone_number:
            flash('电话号码已经被注册了', 'error')
        cursor.execute("SELECT * FROM `t_accounts` WHERE `email` = %s", (email,))
        email_exists = cursor.fetchone()
        if email_exists:
            flash('邮箱已经被注册了', 'error')
        elif not re.fullmatch(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', email):
            flash('邮箱格式不正确', 'error')
        elif not re.fullmatch(r'^\d{11}$', mobile):
            flash('手机号码格式不正确', 'error')
        elif verifycode != cached_data and get_config()['Mail']['ENABLE']:
            flash('验证码错误', 'error')
        elif password != passwordv2:
            flash('两次输入的密码不一致', 'error')
        elif len(password) < get_config()["Security"]["min_password_len"]:
            flash(f"密码长度不能小于 {get_config()['Security']['min_password_len']} 字节", 'error')
        else:
            cursor.execute("INSERT INTO `t_accounts` (`name`, `mobile`, `email`, `password`, `type`, `epoch_created`) "
                           "VALUES (%s, %s, %s, %s, %s, %s)",
                           (username, mobile, email, password_hash(password), repositories.ACCOUNT_TYPE_NORMAL, int(epoch())))
            flash('游戏账号注册成功，请返回登录', 'success')
            cache.delete(email)
    return render_template("account/register.tmpl")

# 邮件验证码 用于注册
@app.route('/account/register_code', methods=['POST'])
def register_code():
    email = request.form.get('email')
    email_pattern = '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$'
    if not re.match(email_pattern, email):
        return json_rsp_with_msg(repositories.RES_FAIL, "邮箱格式不正确", {})
    cursor = get_db().cursor()
    user_query = "SELECT * FROM `t_accounts` WHERE `email` = %s"
    cursor.execute(user_query, (email,))
    user = cursor.fetchone()
    if user:
        return json_rsp_with_msg(repositories.RES_FAIL, "邮箱已经被注册了", {})
    verification_code = ''.join(random.choices(string.digits, k=4))
    mail = current_app.extensions['mail']
    msg = Message(f"注册验证", recipients=[email])
    msg.body = f"你的注册验证码是：{verification_code}，验证码5分钟内有效"
    try:
        mail.send(msg)
    except:
        return json_rsp_with_msg(repositories.RES_FAIL, "未知异常，请联系管理员", {})
    cache.set(email, verification_code, timeout=60*5)
    return json_rsp_with_msg(repositories.RES_SUCCESS, "验证码发送成功，请查收邮箱", {})

function/accountregister.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import json
import random
import settings.repositories as repositories

from flask import request
from time import time as epoch
from flask_caching import Cache
from settings.database import get_db
from settings.loadconfig import get_config
from settings.response import json_rsp, json_rsp_with_msg
from settings.library import request_ip, get_country_for_ip, mask_string, mask_email

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

#=====================校验模块=====================#
# 账号校验(t_sdk_config)
@app.route('/inner/account/verify', methods=['POST'])
def inner_account_verify():
    try:
        data = json.loads(request.data)
        cursor = get_db().cursor()
        token_query = "SELECT * FROM `t_combo_tokens` WHERE `token` = %s AND `uid` = %s"
        cursor.execute(token_query, (data["combo_token"], data["open_id"]))
        token = cursor.fetchone()
        if not token:
            return json_rsp(repositories.RES_SDK_VERIFY_FAIL, {})
        user_query = "SELECT * FROM `t_accounts` WHERE `uid` = %s"
        cursor.execute(user_query, (token["uid"],))
        user = cursor.fetchone()
        if not user:
            print(f"Found valid combo_token={token['token']} for uid={token['uid']}, but no such account exists")
            return json_rsp(repositories.RES_SDK_VERIFY_FAIL, {})
        return json_rsp(repositories.RES_SDK_VERIFY_SUCC, {
            "data": {
                "guest": True if user["type"] == repositories.ACCOUNT_TYPE_GUEST else False,
                "account_type": user["type"],
                "account_uid": str(token["uid"]),
                "ip_info": {
                    "country_code": get_country_for_ip(token["ip"]) or "CN"
                }
            }
        })
    except Exception as err:
        print(f"处理账号校验事件时出现意外错误{err=}, {type(err)=}")
        return json_rsp(repositories.RES_SDK_VERIFY_FAIL, {})

# 账号风险验证
@app.route('/account/risky/api/check', methods=['POST'])
def account_risky_api_check():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "id": "none",
            "action": repositories.RISKY_ACTION_NONE,
            "geetest": None
        }
    })

# 验证account_id和combo_token
@app.route('/hk4e_cn/combo/granter/login/beforeVerify', methods=['POST'])
@app.route('/hk4e_global/combo/granter/login/beforeVerify', methods=['POST'])
def combo_granter_login_verify():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "is_guardian_required": get_config()["Player"]["guardian_required"],      # 未满14周岁阻止登录
            "is_heartbeat_required": get_config()["Player"]["heartbeat_required"],
            "is_realname_required": get_config()["Player"]["realname_required"]       # 实名认证请求
        }
    })

# 二次登录校验
@app.route('/hk4e_cn/combo/granter/login/v2/login', methods=['POST'])
@app.route('/hk4e_global/combo/granter/login/v2/login', methods=['POST'])
def combo_granter_login_v2_login():
    try:
        cursor = get_db().cursor()
        data = json.loads(request.json["data"])
        if data["guest"]:
            guest_query = "SELECT * FROM `t_accounts_guests` WHERE `device` = %s AND `uid` = %s"
            cursor.execute(guest_query, (request.json["device"], data["uid"]))
            guest = cursor.fetchone()
            if not guest:
                return json_rsp_with_msg(repositories.RES_LOGIN_FAILED, "游戏账号信息缓存错误", {})
            user_query = "SELECT * FROM `t_accounts` WHERE `uid` = %s AND `type` = %s"
            cursor.execute(user_query, (data["uid"], repositories.ACCOUNT_TYPE_GUEST))
            user = cursor.fetchone()
            if not user:
                print(f"Found valid account_guest={guest['uid']} for device={guest['device']}, but no such account exists")
                return json_rsp_with_msg(repositories.RES_LOGIN_ERROR, "系统错误，请稍后再试", {})
        else:
            token_query = "SELECT * FROM `t_accounts_tokens` WHERE `token` = %s AND `uid` = %s"
            cursor.execute(token_query, (data["token"], data["uid"]))
            token = cursor.fetchone()
            if not token:
                return json_rsp_with_msg(repositories.RES_LOGIN_FAILED, "游戏账号信息缓存错误", {})
            user_query = "SELECT * FROM `t_accounts` WHERE `uid` = %s AND `type` = %s"
            cursor.execute(user_query, (token["uid"], repositories.ACCOUNT_TYPE_NORMAL))
            user = cursor.fetchone()
            if not user:
                print(f"Found valid account_token={token['token']} for uid={token['uid']}, but no such account exists")
                return json_rsp_with_msg(repositories.RES_LOGIN_ERROR, "系统错误，请稍后再试", {})
        combo_token = ''.join(random.choices('0123456789abcdef', k=get_config()["Security"]["token_length"]))
        device = request.json["device"]
        ip = request_ip(request)
        epoch_generated = int(epoch())
        combo_token_query = "INSERT INTO `t_combo_tokens` (`uid`, `token`, `device`, `ip`, `epoch_generated`) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE `token` = VALUES(`token`), `device` = VALUES(`device`), `ip` = VALUES(`ip`), `epoch_generated` = VALUES(`epoch_generated`)"
        cursor.execute(combo_token_query, (user["uid"], combo_token, device, ip, epoch_generated))
        return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
            "data": {
                "combo_id": 0,
                "account_type": user["type"],
                "data": json.dumps({"guest": True if data["guest"] else False}, separators=(',', ':')),
                "fatigue_remind": {
                    "nickname": "旅行者",
                    "reset_point": 14400,
                    "durations": [180, 240, 300]
                },
                "heartbeat": get_config()["Player"]["heartbeat_required"],
                "open_id": str(data["uid"]),
                "combo_token": combo_token
            }
        })
    except Exception as err:
        print(f"处理 combo 登录事件时出现意外错误 {err=}，{type(err)=}")
        return json_rsp_with_msg(repositories.RES_FAIL, "系统错误，请稍后再试", {})

# 游戏账号信息缓存校验
@app.route('/hk4e_cn/mdk/shield/api/verify', methods=['POST'])
@app.route('/hk4e_global/mdk/shield/api/verify', methods=['POST'])
def mdk_shield_api_verify():
    try:
        cursor = get_db().cursor()
        token_query = "SELECT * FROM `t_accounts_tokens` WHERE `token` = %s AND `uid` = %s"
        cursor.execute(token_query, (request.json["token"], request.json["uid"]))
        token = cursor.fetchone()
        if not token:
            return json_rsp_with_msg(repositories.RES_LOGIN_FAILED, "游戏账号信息缓存错误", {})
        if token["device"] != request.headers.get('x-rpc-device_id'):
            return json_rsp_with_msg(repositories.RES_LOGIN_FAILED, "登录态失效，请重新登录", {})
        user_query = "SELECT * FROM `t_accounts` WHERE `uid` = %s AND `type` = %s"
        cursor.execute(user_query, (token["uid"], repositories.ACCOUNT_TYPE_NORMAL))
        user = cursor.fetchone()
        if not user:
            print(f"Found valid account_token={token['token']} for uid={token['uid']}, but no such account exists")
            return json_rsp_with_msg(repositories.RES_LOGIN_ERROR, "系统错误，请稍后再试", {})
        return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
            "data": {
                "account": {
                    "uid": str(user["uid"]),
                    "name": mask_string(user["name"]),
                    "email": mask_email(user["email"]),
                    "is_email_verify": get_config()["Login"]["email_verify"],
                    "token": token["token"],
                    "country": get_country_for_ip(request_ip(request)) or "CN",
                    "area_code": None
                }
            }
        })
    except Exception as err:
        print(f"处理 MDK Shield API 验证时出现意外错误 {err=}，{type(err)=}")
        return json_rsp_with_msg(repositories.RES_FAIL, "系统错误，请稍后再试", {})
function/accountverify.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import os
import settings.repositories as repositories

from flask_caching import Cache
from settings.loadconfig import get_config
from settings.response import json_rsp_with_msg
from flask import render_template, request, send_file

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

#=====================公告模块=====================#
# 公告功能
@app.route('/common/hk4e_cn/announcement/api/getAlertAnn',methods = ['GET'])
@app.route('/common/hk4e_global/announcement/api/getAlertAnn',methods = ['GET'])
def get_alertann():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "alert": get_config()["Announce"]["alert"],
            "alert_id": 0,
            "remind": get_config()["Announce"]["remind"],
            "extra_remind": get_config()["Announce"]["extra_remind"]
        }
    })

# 蓝贴
@app.route("/admin/swpreload/hk4e_cn/zh-cn.json",methods=['GET'])
def blue_post():
    file_path = repositories.ANNOUNCE_BLUE_PATH
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"


# 获取公告(进门前获取content和list，进门后获取ann pic content 用是否有level字段来区分)
@app.route('/common/hk4e_cn/announcement/api/getAlertPic', methods=['GET'])
@app.route('/common/hk4e_global/announcement/api/getAlertPic', methods=['GET'])
def get_pic():
    file_path = repositories.ANNOUNCE_PATH_INGAME
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/common/hk4e_cn/announcement/api/getAnnList', methods=['GET'])
@app.route('/common/hk4e_global/announcement/api/getAnnList', methods=['GET'])
def get_list():
    if 'level' in request.args:
        return send_file(repositories.ANNOUNCE_PATH_INGAME, mimetype='application/json')
    else:
        return send_file(repositories.ANNOUNCE_PATH_INGATE, mimetype='application/json')
# 游戏外level=undefined 要注意一下
@app.route('/common/hk4e_cn/announcement/api/getAnnContent', methods=['GET'])
@app.route('/common/hk4e_global/announcement/api/getAnnContent', methods=['GET'])
def get_content():
    level = request.args.get('level')
    if level == 'undefined':
        return send_file(repositories.ANNOUNCE_CONTENT_PATH_INGATE, mimetype='application/json')
    elif level is not None:                   # level参数是无或者是其他的默认返回游戏内公告内容
        return send_file(repositories.ANNOUNCE_CONTENT_PATH_INGAME, mimetype='application/json')
    else:
        return send_file(repositories.ANNOUNCE_CONTENT_PATH_INGATE, mimetype='application/json')

# 公告模块
@app.route('/hk4e/announcement/index.html', methods=['GET'])
def handle_announcement():
    return render_template("announce/announcement.tmpl")

# 获取字体
@app.route('/hk4e_cn/combo/granter/api/getFont', methods = ['GET'])
@app.route('/hk4e_global/combo/granter/api/getFont', methods = ['GET'])
def get_font():
    file_path = repositories.ANNOUNCE_FONT_PATH
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"

# 资源文件
@app.route('/hk4e/announcement/2_2e4d2779ad3d19e6406f.js',methods=['GET'])
def get_js():
    file_path = repositories.ANNOUNCE_JS_PATH1
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/hk4e/announcement/vendors_3230378b06826ebc94d3.js',methods=['GET'])
def get_vendors_js():
    file_path = repositories.ANNOUNCE_JS_PATH2
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/hk4e/announcement/bundle_9f9d2fd05b63fd8decfc.js',methods=['GET'])
def get_bundle_js():
    file_path = repositories.ANNOUNCE_JS_PATH3
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/hk4e/announcement/2_cb04d2d72d7555e2ab83.css',methods=['GET'])
def get_css():
    file_path = repositories.ANNOUNCE_CSS_PATH1
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/hk4e/announcement/bundle_dad917ca6970b9fa0ec0.css',methods=['GET'])
def get_bundel():
    file_path = repositories.ANNOUNCE_CSS_PATH2
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/favicon.ico',methods=['GET'])
def get_favicon():
    file_path = repositories.ANNOUNCE_FAVICON_PATH
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/dora/lib/vue/2.6.11/vue.min.js',methods=['GET'])
def get_vue_min():
    file_path = repositories.ANNOUNCE_VUEMIN_PATH
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/dora/biz/mihoyo-analysis/v2/main.js',methods=['GET'])
def get_mainjs():
    file_path = repositories.ANNOUNCE_MAINJS_PATH
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/dora/biz/mihoyo-h5log/v1.0/main.js',methods=['GET'])
def get_mainh5js():
    file_path = repositories.ANNOUNCE_MAINH5JS_PATH
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"
@app.route('/dora/lib/firebase-performance/8.2.7/firebase-performance-standalone-cn.js',methods=['GET'])
@app.route('/dora/lib/firebase-performance/8.2.7/firebase-performance-standalone.js',methods=['GET'])
def get_fprjs():
    file_path = repositories.ANNOUNCE_FPTJS_PATH
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"

function/announcement.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import settings.repositories as repositories

from flask import request
from flask_caching import Cache
from settings.loadconfig import get_config
from settings.response import json_rsp, json_rsp_with_msg

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

def json_rsp(code, message, data=None):
    response = {
        "code": code,
        "message": message
    }
    if data is not None:
        response["data"] = data
    return response

#=====================Api功能=====================#
# Api-Config(https://sandbox-sdk-os.hoyoverse.com)
@app.route('/hk4e_cn/combo/granter/api/getConfig', methods=['GET'])
@app.route('/hk4e_global/combo/granter/api/getConfig', methods=['GET'])
def combo_granter_api_config():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "log_level": "INFO",
            "announce_url": "https://sdk.hoyoverse.com/hk4e/announcement/index.html?sdk_presentation_style=fullscreen\u0026announcement_version=1.40\u0026sdk_screen_transparent=true\u0026game_biz=hk4e_global\u0026auth_appid=announcement\u0026game=hk4e#/",
            "disable_ysdk_guard": get_config()["Player"]["disable_ysdk_guard"],  
            "enable_announce_pic_popup": get_config()["Player"]["enable_announce_pic_popup"],  
            "protocol": get_config()["Player"]["protocol"],  
            "qr_enabled": get_config()["Player"]["qr_enabled"],
            "app_name": "原神",
    		"qr_enabled_apps": {
			"bbs": get_config()['Player']['qr_bbs'],
			"cloud": get_config()['Player']['qr_cloud']
            },
            "qr_app_icons": {
			"app": "",
			"bbs": "",
			"cloud": ""
		    },
            "qr_cloud_display_name": "云·原神",
		    "enable_user_center": get_config()['Player']['enable_user_center'],
		    "functional_switch_configs": {}
        }
    })

# 登录相关
@app.route('/hk4e_cn/mdk/shield/api/loadConfig', methods=['GET', 'POST'])
@app.route('/hk4e_global/mdk/shield/api/loadConfig', methods=['GET', 'POST'])
def mdk_shield_api_loadConfig():
    client = request.args.get('client', '')  # 提供默认值为空字符串
    if client.isdigit():
        client = repositories.PLATFORM_TYPE[int(client)]
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "id": 6,
            "game_key": request.args.get('game_key'),
            "client": client,
            "identity": "I_IDENTITY",
            "guest": get_config()["Auth"]["enable_guest"],
            "ignore_versions": "",
            "scene": repositories.SCENE_ACCOUNT,
            "name": "原神",
            "disable_regist": get_config()["Login"]["disable_regist"],                     
            "enable_email_captcha": get_config()["Login"]["enable_email_captcha"],          
            "thirdparty": ["tp","fb","tw"], # taptap facebook twitter
            "disable_mmt": get_config()["Login"]["disable_mmt"],                            
            "server_guest": get_config()["Auth"]["enable_guest"],
            "thirdparty_ignore": {},
            "enable_ps_bind_account": get_config()["Login"]["enable_ps_bind_account"],
    		"thirdparty_login_configs": {
			"fb": {
				"token_type": "TK_GAME_TOKEN",
				"game_token_expires_in": 2592000
			},
			"tw": {
				"token_type": "TK_GAME_TOKEN",
				"game_token_expires_in": 2592000
			}},
		    "initialize_firebase": get_config()["Login"]["initialize_firebase"],
		    "bbs_auth_login": get_config()["Login"]["bbs_auth_login"],
		    "bbs_auth_login_ignore": [],
		    "fetch_instance_id": get_config()["Login"]["fetch_instance_id"],
		    "enable_flash_login": get_config()["Login"]["enable_flash_login"]
        }
    })

# 获取协议信息
@app.route('/hk4e_cn/mdk/agreement/api/getAgreementInfos', methods=['GET'])
@app.route('/hk4e_global/mdk/agreement/api/getAgreementInfos', methods=['GET'])
def mdk_agreement_api_get():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "marketing_agreements": []
        }
    })

# 协议比较(https://sandbox-sdk-os.hoyoverse.com)
# 
@app.route('/hk4e_cn/combo/granter/api/compareProtocolVersion', methods=['POST','GET'])
@app.route('/hk4e_global/combo/granter/api/compareProtocolVersion', methods=['POST','GET'])
def combo_granter_api_protocol():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "modified": get_config()["Other"]["modified"],
            "protocol": {
			"id": 0,
			"app_id": 4,
			"language": "zh-cn",
			"user_proto": "",
			"priv_proto": "",
			"major": 12,
			"minimum": 0,
			"create_time": "0",
			"teenager_proto": "",
			"third_proto": "",
			"full_priv_proto": ""
            }
        }
    })

# 获取SDKCombo配置信息
@app.route('/combo/box/api/config/sdk/combo', methods=['GET'])
def combo_box_api_config_sdk_combo():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "vals": {
                "telemetry_config": "{\n\"dataupload_enable\": 1\n}",
                "new_register_page_enable": get_config()['Other']['new_register_page_enable'],
                "disable_email_bind_skip": get_config()["Login"]["disable_email_bind_skip"],
                "kcp_enable": get_config()['Other']['kcp_enable'],
                "enable_web_dpi": get_config()['Other']['enable_web_dpi'],
                "network_report_config": "{ \"enable\": 1, \"status_codes\": [206], \"url_paths\": [\"dataUpload\"] }",
                "email_bind_remind_interval": "7",  # 不知道
                "kibana_pc_config": "{ \"enable\": 1, \"level\": \"Info\", \"modules\": [\"download\"] }",
                "pay_payco_centered_host": "bill.payco.com",
                "list_price_tierv2_enable": get_config()['Other']['list_price_tierv2_enable'],
                "email_bind_remind": get_config()["Login"]["email_bind_remind"],
                "payment_cn_config": "{ \"h5_cashier_enable\": 0, \"h5_cashier_timeout\": 3}"
            }
        }
    })

# 预加载
@app.route('/combo/box/api/config/sw/precache', methods=['GET'])
def combo_box_api_config_sw_precache():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": {
            "vals": {
                "url":"https://webstatic.mihoyo.com/sw.html",
                "enable": get_config()["Other"]["serviceworker"]         # 是否加载ServiceWorker进行分析
            }
        }
    })

# 问卷调查
@app.route('/device-fp/api/getExtList', methods=['GET'])
def device_fp_get_ext_list():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
	"data": {
		"code": 200,
		"msg": "ok",
		"ext_list": ["userAgent", "browserScreenSize", "maxTouchPoints", "isTouchSupported", "browserLanguage", "browserPlat", "browserTimeZone", "webGlRender", "webGlVendor", "numOfPlugins", "listOfPlugins", "screenRatio", "deviceMemory", "hardwareConcurrency", "cpuClass", "ifNotTrack", "ifAdBlock", "hasLiedLanguage", "hasLiedResolution", "hasLiedOs", "hasLiedBrowser"],
		"pkg_list": [],
		"pkg_str": ""
	}
    })

# 设备上报
@app.route('/device-fp/api/getFp',methods=['POST'])
def device_report():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
    "data":{
        "device_fp":"38d7ecd67b187", # 假的
        "code":200,
        "msg":"ok"
    }
    })

# 抓出来的我也不知道是什么 似乎是玩家登录信息   
@app.route('/hk4e_cn/combo/guard/api/ping', methods=['POST'])
@app.route('/hk4e_cn/combo/guard/api/ping2', methods=['POST'])
@app.route('/hk4e_global/combo/guard/api/ping',methods=['POST'])
@app.route('/hk4e_global/combo/guard/api/ping2',methods=['POST'])
def pingResponse():
   return json_rsp(repositories.RES_SUCCESS, {})

# 消费提醒
@app.route('/common/hk4e_cn/announcement/api/consumeRemind',methods=['GET'])
@app.route('/common/hk4e_global/announcement/api/consumeRemind',methods=['GET'])
def consume_remind():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": []
    })
function/apiservice.py
CokeSR/Hk4e-SDKSever-Py
# Url示例：http://192.168.1.8:21000/common/api/exchangecdk?sign_type=2&auth_appid=apicdkey&authkey_ver=1
# 孩子随便写写玩的
# Gameserver和Multiserver邮件产出设置
# <ItemOutputLimit>
#   <Item item_id="201" reward_limit="1000" drop_limit="1000" mail_limit="1000"/>
#   <Item item_id="223" reward_limit="100" drop_limit="100" mail_limit="100" />
#   <Item item_id="224" reward_limit="100" drop_limit="100" mail_limit="100" />
# </ItemOutputLimit>
from __main__ import app
from flask import request
from settings.database import get_db_cdk
from settings.library import authkey, send
from settings.checkstatus import get_config
from settings.response import json_rsp_common

import time
import json
import datetime as dt
import settings.repositories as repositories

# 时间计算 用于提供邮件过期时间
server_time = dt.datetime.now()
current_time = int(time.time())
delta_30_days = dt.timedelta(days=30)
after_30_days = server_time + delta_30_days
unixt_30_days = str(time.mktime(after_30_days.timetuple())).split('.')[0]

#===========================游戏内部兑换CDK功能===========================#
@app.route('/common/api/exchangecdk',methods=['GET'])
def cdk_verify():
    # Auth解密来获取兑换者数据,计入数据库
    cdkey = request.args.get('cdkey')
    auth_key = request.args.get("authkey")
    authkey_ver = request.args.get("authkey_ver")
    
    keys = authkey(auth_key, authkey_ver)
    message = json.loads(keys)
    uid = message.get("uid")
    game = message.get("game")
    region = message.get("region")
    account_uid = message.get("account_uid")
    account_type = message.get("ext").get("account_type")
    platform_type = message.get("ext").get("platform_type")

    # 检查请求的CDK内容合法性
    cursor = get_db_cdk().cursor()
    cursor.execute("SELECT * FROM `t_cdk_redeem` WHERE `cdk_name` = %s", cdkey)
    cdk_status = cursor.fetchone()
    if cdk_status is None:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "无效的CDK")

    # 检查CDK状态
    times = cdk_status.get("times")
    enabled = cdk_status.get("enabled")
    cdk_name = cdk_status.get("cdk_name")
    open_time = cdk_status.get("open_time")
    expire_time = cdk_status.get("expire_time")
    template_id = cdk_status.get("template_id")
    
    if open_time > current_time:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "CDK尚未启用")
    elif expire_time < current_time:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "CDK已过期")
    elif enabled == 0:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "CDK尚未启用")
    elif times <= 0:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "CDK已被使用")
    
    # 检查对应的CDK中是否有兑换者的信息
    cursor.execute("SELECT * FROM `t_cdk_record` WHERE `cdk_name` = %s AND `uid` = %s", (cdkey, uid))
    record = cursor.fetchone()
    if record is not None:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "你已经兑换过这个CDK了")

    # 查询与 CDK 相关的邮件模板 有则拼接邮件内容发出去
    cursor.execute("SELECT * FROM `t_cdk_template` WHERE `cdk_template_id` = %s", template_id)
    templates = cursor.fetchone()
    if templates is None:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "邮件模板不存在！CDK发送失败")
    
    expire_time = unixt_30_days
    title = templates.get("title")
    sender = templates.get("sender")
    content = templates.get("content")
    item_list = templates.get("item_list")
    importance = templates.get("importance")
    is_collectible = templates.get("is_collectible")
    region = get_config()['Muipserver']['region']
    content = f"title={title}&sender={sender}&content={content}&expire_time={expire_time}&importance={importance}&is_collectible={is_collectible}&item_list={item_list}&region={region}"
    mail = send(uid, content)
    
    # 如果邮件功能寄掉了
    if mail is None:
        return json_rsp_common(repositories.RES_CDK_EXCHANGE_FAIL, "邮件功能错误！CDK发送失败")
    
    # 将兑换者的信息计入数据库
    cursor.execute("INSERT INTO `t_cdk_record` (`cdk_name`, `uid`, `account_type`, `account_uid`,`region`, `game`, `platform`, `used_time`) VALUES (%s, %s, %s, %s, %s, %s, %s, %s);", (cdk_name, uid, account_type, account_uid, region, game, platform_type, current_time))
    
    # 刷新该CDK的使用次数
    times -= 1
    cursor.execute("UPDATE t_cdk_redeem SET times = %s WHERE cdk_name = %s", (times, cdk_name))
    return json_rsp_common(repositories.RES_CDK_EXCHANGE_SUCC,'SEND MAIL SUCC')
function/cdkservice.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import sys
import settings.repositories as repositories
import data.proto.QueryRegionListHttpRsp_pb2 as RegionList

from base64 import b64encode
from flask_caching import Cache
from settings.library import forward_request
from flask import Response, abort, request
from settings.library import check_config_exists

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = check_config_exists()
    return {'config': config}

with open(repositories.DISPATCH_KEY, 'rb') as f:
    dispatch_key = f.read()
with open(repositories.DISPATCH_SEED, 'rb') as f:
    dispatch_seed = f.read()

#=====================Dispatch配置=====================#
# 实验性分区 Dispatch
# version=CNRELWin4.0.1 lang=2 platform=3 binary=1 time=291 channel_id=1 sub_channel_id=1
# version=OSRELWin2.8.0 lang=2 platform=3 binary=1 time=454 channel_id=1 sub_channel_id=1
@app.route('/query_region_list', methods=['GET'])
def query_dispatch():
    response = RegionList.QueryRegionListHttpRsp()
    response.retcode = repositories.RES_SUCCESS
    for entry in check_config_exists()['Gateserver']:
        if ('name' not in entry or not entry['name'] or
            'title' not in entry or not entry['title'] or
            'dispatchUrl' not in entry or not entry['dispatchUrl']):
            print("#=====================Region读取失败！请检查[Config]配置=====================#")       # 客户端请求的时候再检查一下 如果不符合就原地抛锚
            sys.exit(1)
        region_info = response.region_list.add()
        region_info.name = entry['name']
        region_info.title = entry['title']
        region_info.type = "PRODUTC"
        region_info.dispatch_url = entry['dispatchUrl']
        version = request.args.get('version')
    if version.startswith('CNREL'):
        custom_config = '{"sdkenv":"0","checkdevice":"false","loadPatch":"false","showexception":"false","regionConfig":"pm|fk|add","downloadMode":"0"}'
    elif version == 'CNCB':
        custom_config = '{"sdkenv":"6","checkdevice":"false","loadPatch":"false","showexception":"false","regionConfig":"pm|fk|add","downloadMode":"0"}'
    elif version == 'OSCB':
        custom_config = '{"sdkenv":"7","checkdevice":"false","loadPatch":"false","showexception":"false","regionConfig":"pm|fk|add","downloadMode":"0"}'
    else:
        custom_config = '{"sdkenv":"2","checkdevice":"false","loadPatch":"false","showexception":"false","regionConfig":"pm|fk|add","downloadMode":"0"}'
    encrypted_config = bytearray()
    for coke in range(len(custom_config)):
        encrypted_config.append(ord(custom_config[coke]) ^ dispatch_key[coke % len(dispatch_key)])
    updated_region_list = RegionList.QueryRegionListHttpRsp()
    updated_region_list.region_list.extend(response.region_list)
    updated_region_list.client_secret_key = dispatch_seed
    updated_region_list.client_custom_config_encrypted = bytes(encrypted_config)
    updated_region_list.enable_login_pc = True      # 试过了 无论是否都没用 但是必须要存在 :)
    # 序列化
    serialized_data = updated_region_list.SerializeToString()
    base64_str = b64encode(serialized_data).decode()
    return Response(base64_str, content_type='text/plain')

# 实验性解析QueryCurRegion(live版本服务端dispatchurl必须要填写sdk的地址，通过sdk来传递区服信息)
# version=OSRELWin2.8.0 lang=2 platform=3 binary=1 time=348 channel_id=1 sub_channel_id=1 account_type=1 dispatchSeed=caffbdd6d7460dff key_id=3
@app.route('/query_region/<name>', methods=['GET'])
def query_cur_region(name):
    try:
        return forward_request(request, f"{check_config_exists()['Dispatch']['list'][name]}/query_cur_region?{request.query_string.decode()}")
    except KeyError:
        print(f"未知的Region={name}")
        abort(404)
    except Exception as err:
        print(f"处理请求事件发生以外错误{err=}, {type(err)=}")
        abort(500)
function/dispatch.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import json
import settings.repositories as repositories

from flask_caching import Cache
from settings.loadconfig import get_config
from flask import request,abort, render_template

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

#=====================抽卡模块=====================#
# 祈愿规则
@app.route('/gacha/info/<int:id>', methods=['GET'])
def gacha_info(id):

    schedule = {}
    textmap = {
        'title_map': {},
        'item_map': {}
    }
    language = request.args.get('lang') or 'en'
    try:
        f = open(f"{repositories.GACHA_SCHEDULE_PATH}/{id}.json")                  # 加载当前卡池祈愿规则
        schedule = json.load(f)
        f.close()
    except Exception as err:
        abort(
            404, description=f"Unexpected {err=}, {type(err)=} while loading gacha schedule data for {id=}")
    try:
        f = open(f"{repositories.GACHA_TEXTMAP_PATH}/{language}.json")            # 加载适配的语言
        textmap = json.load(f)
        f.close()
    except Exception as err:
        print(
            f"Unexpected {err=}, {type(err)=} while loading textmap for {language=}")
    return render_template("gacha/details.tmpl", schedule=schedule, textmap=textmap, id=id)

# 祈愿记录
@app.route('/gacha/record/<int:type>', methods=['GET'])
def gacha_log(type):
    return render_template("gacha/history.tmpl")
function/gachaservice.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import random
import re
import string
import settings.repositories as repositories

from flask import request
from time import time as epoch
from flask_caching import Cache
from settings.database import get_db
from settings.loadconfig import get_config
from settings.response import json_rsp_with_msg
from settings.library import request_ip, get_country_for_ip, password_verify, mask_string, mask_email,decrypt_rsa_password

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

#=====================登录模块=====================#
def validate_user_format(user):
    phone_pattern = r'^\d{11}$'
    email_pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    return re.match(phone_pattern, user) is not None or re.match(email_pattern, user) is not None

@app.route('/hk4e_cn/mdk/shield/api/login', methods=['POST'])
@app.route('/hk4e_global/mdk/shield/api/login', methods=['POST'])
def mdk_shield_api_login():
    try:
        cursor = get_db().cursor()
        if "account" not in request.json:
            return json_rsp_with_msg(repositories.RES_FAIL, "缺少登录凭据", {})
        account = request.json["account"]
        email_name_query = "SELECT * FROM `t_accounts` WHERE (`email` = %s OR `mobile` = %s ) AND `type` = %s"
        cursor.execute(email_name_query, (account, account, repositories.ACCOUNT_TYPE_NORMAL))
        user = cursor.fetchone()
        if not validate_user_format(account):
            return json_rsp_with_msg(repositories.RES_FAIL, "错误的登录格式", {})
        if not user:
            return json_rsp_with_msg(repositories.RES_LOGIN_FAILED, "该账号未注册", {})
        if get_config()["Auth"]["enable_password_verify"]:
            if request.json["is_crypto"]:
                password = decrypt_rsa_password(request.json["password"])
            else:
                password = request.json["password"]
            if not password_verify(password, user["password"]):
                return json_rsp_with_msg(repositories.RES_LOGIN_FAILED, "用户名或密码不正确", {})
        token = ''.join(random.choices(string.ascii_letters, k=get_config()["Security"]["token_length"]))
        device_id = request.headers.get('x-rpc-device_id')
        ip = request_ip(request)
        epoch_generated = int(epoch())
        insert_token_query = "INSERT INTO `t_accounts_tokens` (`uid`, `token`, `device`, `ip`, `epoch_generated`) VALUES (%s, %s, %s, %s, %s)"
        cursor.execute(insert_token_query, (user["uid"], token, device_id, ip, epoch_generated))
        return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
            "data": {
                "account": {
                    "uid": str(user["uid"]),
                    "name": mask_string(user["name"]),
                    "email": mask_email(user["email"]),
                    "is_email_verify": get_config()["Login"]["email_verify"],
                    "token": token,
                    "country": get_country_for_ip(ip) or "CN",
                    "area_code": None
                },
                "device_grant_required": get_config()["Login"]["device_grant_required"],
                "realname_operation": None,
                "realperson_required": get_config()["Login"]["realperson_required"],
                "safe_mobile_required": get_config()["Login"]["safe_mobile_required"]
            }
        })
    except Exception as err:
        print(f"处理登录事件时出现意外错误 {err=}，{type(err)=}")
        return json_rsp_with_msg(repositories.RES_FAIL, "系统错误，请稍后再试", {})

# 快速游戏
@app.route('/hk4e_cn/mdk/guest/guest/login', methods=['POST'])
@app.route('/hk4e_cn/mdk/guest/guest/v2/login', methods=['POST'])
@app.route('/hk4e_global/mdk/guest/guest/login', methods=['POST'])
@app.route('/hk4e_global/mdk/guest/guest/v2/login', methods=['POST'])
def mdk_guest_login():
    if not get_config()["Auth"]["enable_guest"]:
        return json_rsp_with_msg(repositories.RES_LOGIN_CANCEL, "游客模式已关闭", {})
    try:
        cursor = get_db().cursor()
        guest_query = "SELECT * FROM `t_accounts_guests` WHERE `device` = %s"
        cursor.execute(guest_query, (request.json["device"],))
        guest = cursor.fetchone()
        if not guest:
            insert_accounts_query = "INSERT INTO `t_accounts` (`type`, `epoch_created`) VALUES (%s, %s)"
            cursor.execute(insert_accounts_query, (repositories.ACCOUNT_TYPE_GUEST, int(epoch())))
            user = {"uid": cursor.lastrowid, "type": repositories.ACCOUNT_TYPE_GUEST}
            insert_guests_query = "INSERT INTO `t_accounts_guests` (`uid`, `device`) VALUES (%s, %s)"
            cursor.execute(insert_guests_query, (user["uid"], request.json["device"]))
        else:
            user_query = "SELECT * FROM `t_accounts` WHERE `uid` = %s AND `type` = %s"
            cursor.execute(user_query, (guest["uid"], repositories.ACCOUNT_TYPE_GUEST))
            user = cursor.fetchone()
            if not user:
                print(f"Found valid account_guest={guest['uid']} for device={guest['device']}, but no such account exists")
                return json_rsp_with_msg(repositories.RES_LOGIN_ERROR, "系统错误，请稍后再试", {})
        return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
            "data": {
                "account_type": user["type"],
                "guest_id": str(user["uid"])
            }
        })
    except Exception as err:
        print(f"处理游客登录事件时出现意外错误 {err=}, {type(err)=}")
        return json_rsp_with_msg(repositories.RES_FAIL, "系统错误，请稍后再试", {})

function/loginservice.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import yaml
import settings.repositories as repositories

from flask_caching import Cache
from settings.loadconfig import get_config
from flask import send_from_directory, render_template
from settings.response import json_rsp, json_rsp_with_msg

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

def json_rsp(code, message, data=None):
    response = {
        "code": code,
        "message": message
    }
    if data is not None:
        response["data"] = data
    return response
#====================SDKServer====================#
# 首页
@app.route('/')
@app.route('/sandbox/index.html', methods=['GET'])
def account_index():
    return render_template("account/index.tmpl")

# 检查SDK配置(https://testing-abtest-api-data-sg.mihoyo.com) 不知道什么用 不写config
@app.route('/data_abtest_api/config/experiment/list', methods=['GET', 'POST'])
def abtest_config_experiment_list():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
        "data": [{
		"code": 1000,
		"type": 2,
		"config_id": "246",
		"period_id": "5147_328",
		"version": "2",
		"configs": {
			"cardType": "native",
			"cashierId": "34b2b997-b1c7-412b-b5f2-eeef09a03d92"
		}}, {
		"code": 1000,
		"type": 2,
		"config_id": "245",
		"period_id": "5145_536",
		"version": "1",
		"configs": {
			"foldOther": "true"             # 简化界面？
		}}, {
		"code": 1000,
		"type": 2,
		"config_id": "244",
		"period_id": "5144_535",
		"version": "1",
		"configs": {
			"expandMixedQRcode": "true"     # 二维码
		}}, {
		"code": 1010,
		"type": 2,
		"config_id": "243",
		"period_id": "",
		"version": "",
		"configs": {
			"disableMarket": "false"        # 市场？
		}}]
    })

#=====================状态收集=====================#
# log收集
@app.route('/log', methods=['POST'])
@app.route('/h5/upload',methods=['POST'])
@app.route('/log/sdk/upload', methods=['POST'])
@app.route('/crash/dataUpload', methods=['POST'])
@app.route('/client/event/dataUpload', methods = ['POST'])
@app.route('/sdk/dataUpload', methods=['POST'])
@app.route('/common/h5log/log/batch', methods=['POST'])
def sdk_log():
    return json_rsp(repositories.RES_SUCCESS, {})

# 红点配置 一般infos为空 特别写的
@app.route('/hk4e_cn/combo/red_dot/list', methods=['POST'])
@app.route('/hk4e_global/combo/red_dot/list', methods=['POST'])
def red_dot():
    return json_rsp(repositories.RES_SUCCESS, "ok", {
        "infos": []
    })
"""
def red_dot():
    return json_rsp(define.RES_SUCCESS, "ok", {
        "infos": [
            {
                "red_point_type": 2201,
                "content_id": 184,
                "display": get_config()["Reddot"]["display"]
            }
        ]
    })
"""
#======================mi18n======================#
@app.route('/admin/mi18n/plat_cn/m2020030410/m2020030410-version.json', methods=['GET'])
@app.route('/admin/mi18n/plat_oversea/m202003049/m202003049-version.json',methods=['GET'])
@app.route('/admin/mi18n/plat_oversea/m2020030410/m2020030410-version.json', methods=['GET'])
def mi18n_version():
    return json_rsp(repositories.RES_SUCCESS, {"version": 79})
@app.route('/admin/mi18n/plat_os/m09291531181441/m09291531181441-version.json',methods=['GET'])
def min18_os_version():
    return json_rsp(repositories.RES_SUCCESS, {"version": 16})
@app.route('/admin/mi18n/plat_cn/m2020030410/m2020030410-<language>.json', methods=['GET'])
@app.route('/admin/mi18n/plat_oversea/m2020030410/m2020030410-<language>.json', methods=['GET'])
def mi18n_serve(language):
    return send_from_directory(repositories.MI18N_PATH, f"{language}.json")

#================cokeserver-config================#
@app.route('/hk4e_cn/developers/config.yaml',methods=['GET'])
@app.route('/hk4e_global/developers/config.yaml',methods=['GET'])
def view_config():
    config_path = repositories.CONFIG_FILE_PATH
    try:
        with open(config_path, 'r', encoding='utf-8') as file:
            config_data = yaml.safe_load(file)
        return config_data
    except FileNotFoundError:
        return "Config file not found"
    except Exception as e:
        return f"Error reading config file: {str(e)}"

@app.route('/hk4e_cn/developers/keys/authverify.pem',methods=['GET'])
@app.route('/hk4e_global/developers/keys/authverify.pem',methods=['GET'])
def view_authverify_key():
    config_path = repositories.AUTHVERIFY_KEY_PATH
    try:
        with open(config_path, 'r', encoding='utf-8') as file:
            config_data = yaml.safe_load(file)
        return config_data
    except FileNotFoundError:
        return "Config file not found"
    except Exception as e:
        return f"Error reading config file: {str(e)}"

@app.route('/hk4e_cn/developers/keys/password.pem',methods=['GET'])
@app.route('/hk4e_global/developers/keys/password.pem',methods=['GET'])
def view_password_key():
    config_path = repositories.PASSWDWORD_KEY_PATH
    try:
        with open(config_path, 'r', encoding='utf-8') as file:
            config_data = yaml.safe_load(file)
        return config_data
    except FileNotFoundError:
        return "Config file not found"
    except Exception as e:
        return f"Error reading config file: {str(e)}"
function/otherservice.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import os
import settings.repositories as repositories

from flask import send_file
from flask_caching import Cache
from settings.loadconfig import get_config
from settings.response import json_rsp_with_msg

cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.context_processor
def inject_config():
    config = get_config()
    return {'config': config}

#=====================支付模块=====================#
# 支付窗口-RMB
@app.route('/hk4e_cn/mdk/shopwindow/shopwindow/listPriceTier', methods = ['POST'])
@app.route('/hk4e_cn/mdk/shopwindow/shopwindow/listPriceTierV2', methods = ['POST'])
def price_tier_serve_cn():
    file_path = repositories.SHOPWINDOW_TIERS_PATH_CN
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"

@app.route('/hk4e_global/mdk/shopwindow/shopwindow/getCurrencyAndCountryByIp',methods=['POST'])
def get_cur_country():
    return json_rsp_with_msg(repositories.RES_SUCCESS)

# 支付窗口-美元
@app.route('/hk4e_global/mdk/shopwindow/shopwindow/listPriceTier', methods = ['POST'])
@app.route('/hk4e_global/mdk/shopwindow/shopwindow/listPriceTierV2', methods = ['POST'])
def price_tier_serve_os():
    file_path = repositories.SHOPWINDOW_TIERS_PATH_OS
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"

# 支付平台展示1
@app.route('/plutus/api/v2/listPayPlat', methods = ['GET'])
@app.route('/hk4e_cn/mdk/tally/tally/listPayPlat', methods = ['POST'])
def price_pay_types_serve_2():
    file_path = repositories.SHOPWINDOW_PAY_TYPES_PATH_CN
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"

# 支付平台展示2
@app.route('/hk4e_global/mdk/tally/tally/listPayPlat', methods = ['POST'])
def price_pay_types_serve_1():
    file_path = repositories.SHOPWINDOW_PAY_TYPES_PATH_OS
    if os.path.exists(file_path):
        return send_file(file_path)
    else:
        return "Not found"

# 支付确认
@app.route('/plutus/api/v2/check',methods=['GET'])
def charge_check():
    return json_rsp_with_msg(repositories.RES_SUCCESS, "OK", {
    "data":{
        "status":"CheckStatusInit"
    }
    })
function/rechargeservice.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import json
import settings.repositories as repositories

from time import time as epoch
from flask import abort,request
from settings.database import get_db
from settings.response import json_rsp

#=====================GameServer请求处理=====================#
# 玩家登入
@app.route('/bat/game/gameLoginNotify', methods = ['POST'])
@app.route('/inner/bat/game/gameLoginNotify', methods = ['POST'])
# @ip_whitelist(['192.168.1.8'])
def player_login():
   cursor = get_db().cursor()
   player_info = json.loads(request.data.decode())
   uid = player_info['uid']
   account_type = player_info['account_type']
   account = player_info['account']
   platform = player_info['platform']
   region = player_info['region']
   biz_game = player_info['biz_game']
   cursor.execute('INSERT INTO `t_accounts_events` (`uid`, `method`, `account_type`, `account_id`, `platform`, `region`, `biz_game`, `epoch_created`) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)', (uid, 'LOGIN', account_type, account, platform, region, biz_game, int(epoch())))
   return json_rsp(repositories.RES_SUCCESS, {})

# 玩家登出
@app.route('/bat/game/gameLogoutNotify', methods = ['POST'])
@app.route('/inner/bat/game/gameLogoutNotify', methods = ['POST'])
def player_logout():
   cursor = get_db().cursor()
   player_info = json.loads(request.data.decode())
   uid = player_info['uid']
   account_type = player_info['account_type']
   account = player_info['account']
   platform = player_info['platform']
   region = player_info['region']
   biz_game = player_info['biz_game']
   cursor.execute('INSERT INTO `t_accounts_events` (`uid`, `method`, `account_type`, `account_id`, `platform`, `region`, `biz_game`, `epoch_created`) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)', (uid, 'LOGOUT', account_type, account, platform, region, biz_game, int(epoch())))
   return json_rsp(repositories.RES_SUCCESS, {})

# 心跳包
@app.route('/bat/game/gameHeartBeatNotify', methods = ['POST'])
@app.route('/inner/bat/game/gameHeartBeatNotify', methods = ['POST'])
def player_heartbeat():
   print(request.data)
   try:
      return json_rsp(repositories.RES_SUCCESS, {})
   except Exception as err:
      print(f"处理心跳包时出现意外错误{err=}, {type(err)=}")
      abort(500)

# 配置验证
@app.route('/config/verify',methods = ['GET'])
@app.route('/perf/config/verify',methods = ['GET'])
def config_verify():
   return json_rsp(repositories.RES_SUCCESS, {"code": 0})

# Data上传
@app.route('/dataUpload',methods = ['POST'])
@app.route('/perf/dataUpload',methods = ['POST'])
def data_upload():
   return json_rsp(repositories.RES_SUCCESS, {"code": 0})
function/safeservice.py
CokeSR/Hk4e-SDKSever-Py
from flask import Flask
app = Flask(__name__)
import sys
import codecs
import settings.logoutput
import function.dispatch
import function.apiservice
import function.safeservice
import function.gachaservice
import function.otherservice
import function.announcement
import function.loginservice
import function.accountverify
import function.accountrecover
import function.accountregister
import function.rechargeservice
import function.cdkservice

from flask_mail import Mail
from werkzeug.serving import run_simple
from settings.loadconfig import load_config
from settings.database import initialize_database
from werkzeug.middleware.proxy_fix import ProxyFix
from settings.checkstatus import check_region , check_dispatch , check_config , check_mysql_connection , check_database_exists,check_muipserver
sys.stdout = codecs.getwriter("utf-8")(sys.stdout.detach())

#======================Falsk(main)======================#
def start_flask_server(config):
    app.secret_key = config["Setting"]["secret_key"]
    app.debug = config["Setting"]["debug"]
    app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)
    mail_config = config.get('Mail', {})
    enable_mail = mail_config.get('ENABLE', True)
    if enable_mail:
        app.config.update(mail_config)
        mail = Mail(app)
        app.extensions['Mail'] = mail
    run_simple(
        config["Setting"]["listen"],
        config["Setting"]["port"],
        app,
        # use_reloader= config["Setting"]["reload"],    # 热重载1 按照config配置文件来设置
        # use_debugger= config["Setting"]["debug"],
        # threaded= config["Setting"]["threaded"]
        use_reloader=True,                              # 热重载2 快捷设置 方便debug
        use_debugger=False,
        threaded=True                                   # 多线程模式 默认开启
    )

if __name__ == '__main__':
    if len(sys.argv) < 2:
        raise Exception("请提供命令行参数: serve 或 initdb")
    config = load_config()
    command = sys.argv[1]
    if command == "serve": 
        if not check_config():
            print("#=====================[Config]文件校验失败！请检查服务配置=====================#")
            sys.exit(1)
        if not check_mysql_connection():
            print("#======================[Mysql]连接失败！请检查服务配置======================#")
            sys.exit(1)
        if not check_database_exists():
            print("#======================[Mysql]库查询失败！请检查服务配置======================#")
            sys.exit(1)
        if not check_region():
            print("#=====================[Region]读取失败！请检查[Config]配置=====================#")
            sys.exit(1)
        if not check_dispatch():
            print("#=====================[Dispatch]读取失败！请检查[Config]配置=====================#")
            sys.exit(1)
        if not check_muipserver():
            print("#=====================[Muipserver]读取失败！请检查[Config]配置=====================#")
            sys.exit(1)
        start_flask_server(config)
    elif command == "initdb":
        if not check_config():
            print("#=====================[Config]文件校验失败！请检查服务配置=====================#")
            sys.exit(1)
        if not check_mysql_connection():
            print("#======================[Mysql]连接失败！请检查服务配置======================#")
            sys.exit(1)
        initialize_database()
    else:
        raise Exception("未知的操作！必须是以下命令: serve 或 initdb")
main.py
CokeSR/Hk4e-SDKSever-Py
import os
import subprocess

# 绝对路径
protoc_path = "C:/Users/15008/Desktop/GitHub/Hk4e-SDKSever-Py/protobuild/srv/bin/protoc.exe"
proto_folder_path = "C:/Users/15008/Desktop/GitHub/Hk4e-SDKSever-Py/protobuild/proto"
output_folder_path = "C:/Users/15008/Desktop/GitHub/Hk4e-SDKSever-Py/data/proto"

def convert_proto_to_python(proto_folder, output_folder, protoc_path):
    for root, dirs, files in os.walk(proto_folder):
        for file in files:
            if file.endswith(".proto"):
                proto_path = os.path.join(root, file)
                output_file = file.replace(".proto", "_pb2.py")
                output_path = os.path.join(output_folder, output_file)
                command = f"{protoc_path} --python_out={output_folder} --proto_path={proto_folder} {proto_path}"
                subprocess.call(command, shell=True)
convert_proto_to_python(proto_folder_path, output_folder_path, protoc_path)
protobuild/BuildProto_Python.py
CokeSR/Hk4e-SDKSever-Py
import yaml
import pymysql
import settings.repositories as repositories

def get_config():
    with open(repositories.CONFIG_FILE_PATH, encoding='utf-8') as file:
        config = yaml.safe_load(file)
    return config

#======================mysql检查=====================#
# 检查连接
def check_mysql_connection():
    config = get_config()['Database']
    try:
        conn = pymysql.connect(
            host=config['host'],
            user=config['user'],
            port=config['port'],
            password=config['password'],
            charset='utf8mb4'
        )
        conn.close()
        return True
    except pymysql.Error:
        return False

# 检查连接后是否存在库
def check_database_exists():
    config = get_config()['Database']
    try:
        conn = pymysql.connect(
            host=config['host'],
            user=config['user'],
            port=config['port'],
            password=config['password'],
            charset='utf8mb4'
        )
        cursor = conn.cursor()
        cursor.execute("SHOW DATABASES")
        databases = cursor.fetchall()
        cursor.close()
        conn.close()
        found_account_library = False
        found_exchcdk_library = False
        for db in databases:
            if db[0] == config['account_library_name']:
                found_account_library = True
            if db[0] == config['exchcdk_library_name']:
                found_exchcdk_library = True
        if found_account_library and found_exchcdk_library:
            return True
        elif not found_account_library:
            print(">> [Error] 未找到账号管理库")
        elif not found_exchcdk_library:
            print(">> [Error] 未找到CDK管理库")
        return False
    except pymysql.Error:
        return False

#=====================Config检查完整性=====================#
def check_config():
    try:
        with open(repositories.CONFIG_FILE_PATH, 'r',encoding='utf-8') as file:
            config = yaml.safe_load(file)
        required_settings = {
            'Setting': ['listen', 'port', 'reload', 'debug', 'threaded', 'high_frequency_logs', 'cdkexchange','secret_key'],
            'Database': ['host', 'user', 'port', 'autocreate','account_library_name','exchcdk_library_name','password'],
            'Login': ['disable_mmt', 'disable_regist', 'disable_email_bind_skip', 'enable_email_captcha', 'enable_ps_bind_account', 'email_bind_remind', 'email_verify', 'realperson_required', 'safe_mobile_required', 'device_grant_required', 'initialize_firebase', 'bbs_auth_login', 'fetch_instance_id', 'enable_flash_login'],
            'Player': ['disable_ysdk_guard', 'enable_announce_pic_popup', 'protocol', 'qr_enabled', 'qr_bbs', 'qr_cloud', 'enable_user_center', 'guardian_required', 'realname_required', 'heartbeat_required'],
            'Announce': ['remind', 'alert', 'extra_remind'],
            'Security': ['token_length', 'min_password_len'],
            'Auth': ['enable_password_verify', 'enable_guest'],
            'Other': ['modified', 'serviceworker', 'new_register_page_enable', 'kcp_enable', 'enable_web_dpi', 'list_price_tierv2_enable'],
            'Muipserver': ['address','region','port','sign'],
            'Dispatch': ['list'],
            'Mail': ['ENABLE', 'MAIL_SERVER', 'MAIL_PORT', 'MAIL_USE_TLS', 'MAIL_USE_SSL', 'MAIL_USERNAME', 'MAIL_PASSWORD', 'MAIL_DEFAULT_SENDER']
        }
        for section, settings in required_settings.items():
            if section not in config:
                return False
            for setting in settings:
                if setting not in config[section]:
                    return False
        return True
    except FileNotFoundError:
        return False
    except yaml.YAMLError:
        return False

# 单独拎出来 检查region对不对
def check_region():
    for entry in get_config()['Gateserver']:
        if ('name' not in entry or not entry['name'] or
            'title' not in entry or not entry['title'] or
            'dispatchUrl' not in entry or not entry['dispatchUrl']):
            print(">> [Error]-[Gateserver]配置表中有项为空或不完全")
            return False
    return True

# 检查dispatch_list 每个字段是不是空的 是空的你玩鸡毛
def check_dispatch():
    config = get_config()['Dispatch']
    if ('list' not in config or not isinstance(config['list'], dict)):
        print(">> [Error]-[Dispatch]配置项损坏")
        return False
    for name, url in config['list'].items():
        if not isinstance(name, str) or not isinstance(url, str) or not url.startswith('http' or 'https'):
            print(">> [Error]-[Disaptch]配置表中有项为空")
            return False
    return True

# 检查Muipserver每个字段是不是空的 是空的你玩鸡毛
def check_muipserver():
    config = get_config()['Muipserver']
    if ('address' not in config or 'region' not in config or 'port' not in config or 'sign' not in config):
        print(">> [Error]-[Muipserver]配置项损坏")
        return False
    if (not isinstance(config['address'], str) or
        not isinstance(config['port'], int)):
        print(">> [Error]-[Muipserver]配置表中所设置的格式不正确(address:str|port:int)")
        return False
    if (not config['address'] or not config['region'] or not config['port']):
        print(">> [Error]-[Muipserver]配置表中有项为空")
        return False
    return True
settings/checkstatus.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import sys
import pymysql
import settings.database as database

from flask import g
from settings.library import check_config_exists
from settings.checkstatus import check_mysql_connection

#=====================数据库创建=====================#
# 在原有的基础上直接cv 懒得思考了
# 重置数据库的时候账号管理连着CDK配置一起扬了
def dict_factory(cursor, row):
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d

def get_db():
    db = getattr(g, '_database', None)
    if db is None:
        config = check_config_exists()['Database']
        db = g._database = pymysql.connect(
            host=config['host'],
            user=config['user'],
            port=config['port'],
            password=config['password'],
            database=config['account_library_name'],
            cursorclass=pymysql.cursors.DictCursor
        )
    return db

def get_db_cdk():
    db = getattr(g, '_database', None)
    if db is None:
        config = check_config_exists()['Database']
        db = g._database = pymysql.connect(
            host=config['host'],
            user=config['user'],
            port=config['port'],
            password=config['password'],
            database=config['exchcdk_library_name'],
            cursorclass=pymysql.cursors.DictCursor
        )
    return db

# 账号管理库
def init_db(auto_create = check_config_exists()['Database']['autocreate']):
    config = check_config_exists()['Database']
    conn = pymysql.connect(
        host=config['host'],
        user=config['user'],
        port=config['port'],
        password=config['password'],
        charset='utf8mb4'
    )
    cursor = conn.cursor()
    if auto_create:
        cursor.execute("CREATE DATABASE IF NOT EXISTS `{}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci".format(config['account_library_name']))
    cursor.execute("USE `{}`".format(config['account_library_name']))
    cursor.execute("DROP TABLE IF EXISTS `t_accounts`")
    cursor.execute("DROP TABLE IF EXISTS `t_accounts_tokens`")
    cursor.execute("DROP TABLE IF EXISTS `t_accounts_guests`")
    cursor.execute("DROP TABLE IF EXISTS `t_accounts_events`")
    cursor.execute("DROP TABLE IF EXISTS `t_accounts_thirdparty`")
    cursor.execute("DROP TABLE IF EXISTS `t_thirdparty_tokens`")
    cursor.execute("DROP TABLE IF EXISTS `t_combo_tokens`")
    
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_accounts` (
                     `uid` INT AUTO_INCREMENT PRIMARY KEY COMMENT '玩家UID',
                     `name` VARCHAR(255) UNIQUE COMMENT '用户名',
                     `mobile` VARCHAR(255) UNIQUE COMMENT '手机号',
                     `email` VARCHAR(255) UNIQUE COMMENT '电子邮件',
                     `password` VARCHAR(255) COMMENT '哈希密码',
                     `type` INT NOT NULL COMMENT '类型',
                     `epoch_created` INT NOT NULL COMMENT '时间戳'
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT='玩家账号信息表'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_accounts_tokens` (
                     `uid` INT NOT NULL COMMENT '玩家UID',
                     `token` VARCHAR(255) NOT NULL COMMENT '登录Token',
                     `device` VARCHAR(255) NOT NULL COMMENT '设备ID',
                     `ip` VARCHAR(255) NOT NULL COMMENT '登录IP',
                     `epoch_generated` INT NOT NULL COMMENT '时间戳',
                     PRIMARY KEY(`uid`,`token`)
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT='账号登录token'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_accounts_guests` (
                     `uid` INT NOT NULL COMMENT '玩家UID',
                     `device` VARCHAR(255) NOT NULL COMMENT '设备ID',
                     PRIMARY KEY(`uid`,`device`)
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT='游客登录信息表'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_accounts_events` (
                     `uid` INT NOT NULL COMMENT '玩家UID',
                     `method` VARCHAR(255) NOT NULL COMMENT '登录方式',
                     `account_type` INT NOT NULL COMMENT '账号类型',
                     `account_id` INT NOT NULL COMMENT '账号ID',
                     `platform` INT NOT NULL COMMENT '平台',
                     `region` VARCHAR(255) NOT NULL COMMENT '区服信息',
                     `biz_game` VARCHAR(255) NOT NULL,
                     `epoch_created` INT NOT NULL COMMENT '时间戳',
                     PRIMARY KEY(`epoch_created`)
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT='账号活动记录表 由GameServer控制'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_accounts_thirdparty` (
                     `uid` INT NOT NULL COMMENT '玩家UID',
                     `type` INT NOT NULL COMMENT '类型',
                     `external_name` VARCHAR(255) NOT NULL COMMENT '标识名称',
                     `external_id` INT NOT NULL COMMENT '标识ID',
                     PRIMARY KEY(`uid`,`type`)
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT='第三方账号登录信息表'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_thirdparty_tokens` (
                     `uid` INT NOT NULL COMMENT '玩家UID',
                     `type` INT NOT NULL COMMENT '类型',
                     `token` VARCHAR(255) NOT NULL COMMENT '登录Token'
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT='第三方账号登录token'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_combo_tokens` (
                     `uid` INT NOT NULL COMMENT '玩家UID',
                     `token` VARCHAR(255) NOT NULL COMMENT '登录Token',
                     `device` VARCHAR(255) NOT NULL COMMENT '设备ID',
                     `ip` VARCHAR(255) NOT NULL COMMENT '登录IP',
                     `epoch_generated` INT NOT NULL COMMENT '时间戳',
                     PRIMARY KEY(`uid`)
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT='设备信息token'
    """)
    conn.commit()
    conn.close()

# CDK管理库
def init_db_cdk(auto_create = check_config_exists()['Database']['autocreate']):
    config = check_config_exists()['Database']
    conn = pymysql.connect(
        host=config['host'],
        user=config['user'],
        port=config['port'],
        password=config['password'],
        charset='utf8mb4'
    )
    cursor = conn.cursor()
    if auto_create:
        cursor.execute("CREATE DATABASE IF NOT EXISTS `{}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci".format(config['exchcdk_library_name']))
    cursor.execute("USE `{}`".format(config['exchcdk_library_name']))
    cursor.execute("DROP TABLE IF EXISTS `t_cdk_record`")
    cursor.execute("DROP TABLE IF EXISTS `t_cdk_redeem`")
    cursor.execute("DROP TABLE IF EXISTS `t_cdk_template`")

    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_cdk_record` (
                    `cdk_name` varchar(255) NOT NULL COMMENT '使用的CDK',
                    `uid` INT NOT NULL COMMENT '玩家UID',
                    `account_type` varchar(255) NOT NULL COMMENT '账号类型',
                    `account_uid` INT NOT NULL COMMENT '账号ID',
                    `region` varchar(255) NOT NULL COMMENT '所在区服',
                    `game` varchar(255) NOT NULL COMMENT 'cn/global',
                    `platform` varchar(255) NOT NULL COMMENT '客户端平台',
                    `used_time` INT NOT NULL COMMENT '使用时间'
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT '玩家CDK兑换记录'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_cdk_redeem` (
                    `cdk_name` varchar(255) NOT NULL COMMENT 'CDK配置',
                    `open_time` INT NOT NULL COMMENT '启用时间',
                    `expire_time` INT NOT NULL COMMENT '过期时间',
                    `enabled` INT NOT NULL COMMENT '1启用0不启用',
                    `template_id` INT NOT NULL COMMENT '与CDK邮件配置相对应',
                    `times` INT NOT NULL COMMENT '使用次数',
                    PRIMARY KEY (`cdk_name`)
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT 'CDK配置'
    """)
    cursor.execute("""CREATE TABLE IF NOT EXISTS `t_cdk_template` (
                    `cdk_template_id` int NOT NULL COMMENT '与CDK配置相对应',
                    `title` varchar(255) NOT NULL COMMENT '邮件标头',
                    `sender` varchar(255) NOT NULL COMMENT '署名',
                    `content` varchar(255) NOT NULL COMMENT '邮件内容',
                    `importance` INT NOT NULL COMMENT '是否是星标邮件(0/1)',
                    `is_collectible` varchar(255) NOT NULL COMMENT '是否纳入收藏夹(true/false)',
                    `item_list` varchar(255) NOT NULL COMMENT '物品id:数量 逗号分隔',
                    PRIMARY KEY (`cdk_template_id`)
                  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                  COMMENT 'CDK邮件配置'
    """)
    conn.commit()
    conn.close()
@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, '_database', None)
    if db is not None:
        db.commit()
        db.close()

# 重置数据库
def initialize_database():
    print(">> [Waring] 正在初始化数据库结构(清空数据)...")
    database.init_db()
    database.init_db_cdk()
    print(">> [Successful] 初始化数据库完成")
settings/database.py
CokeSR/Hk4e-SDKSever-Py
import sys
import time
import rsa
import yaml
import base64
import bcrypt
import hashlib
import datetime
import requests
import urllib.parse
import geoip2.database
import settings.repositories as repositories

from functools import wraps
from flask import abort, request
from rsa import PublicKey, transform, core
from settings.restoreconfig import recover_config
from settings.loadconfig import get_config, get_json_config

#=====================函数库=====================#
# 检查[Config]文件是否存在并告知是否创建
# database dispatch 读取的是这里
def check_config_exists():
   config = None
   try:
      with open(repositories.CONFIG_FILE_PATH, encoding='utf-8') as file:
         config = yaml.safe_load(file)
   except FileNotFoundError:
      print("#=====================未检测到[Config]文件！运行失败=====================#")
      select = input(">> 是否创建新的[Config]文件？(y/n):")
      if select == 'y' or select == 'Y':
         recover_config()
         print(">> [Successful]-[Config]文件创建成功")
         sys.exit(1)
      elif select == 'n' or select == 'N':
         print(">> [Waring] 取消创建[Config]文件，停止运行...")
         sys.exit(1)
      else:
         print(">> [Error] 非法输入！停止运行...")
         sys.exit(1)
   return config

# 通过IP来检查是哪个国家
def get_country_for_ip(ip):
   with geoip2.database.Reader(repositories.GEOIP2_DB_PATH) as reader:
      try:
         return reader.country(ip).country.iso_code
      except geoip2.errors.AddressNotFoundError:
         pass
      except geoip2.errors.GeoIP2Error as err:
         print(f"Unexpected {err=} while resolving country code for {ip=}")
         pass
   return None

# 白名单准入
def ip_whitelist(allowed_ips):
   def decorator(func):
      @wraps(func)
      def wrapper(*args, **kwargs):
         if request.remote_addr not in allowed_ips:
            abort(403)
         return func(*args, **kwargs)
      return wrapper
   return decorator

# 信息处理
def forward_request(request, url):
   return requests.get(url, headers={"miHoYoCloudClientIP": request_ip(request)}).content
def forward_request_database(request, url, data):
   return requests.post(url, headers={"miHoYoCloudClientIP": request_ip(request)}, data=data)  
def request_ip(request):
   return request.remote_addr
def chunked(size, source):
   for i in range(0, len(source), size):
      yield source[i:i+size]

# 密码保存
def password_hash(password):
   h = hashlib.new('sha256')
   h.update(password.encode())
   return bcrypt.hashpw(password.encode(), bcrypt.gensalt())

# 密码验证
def password_verify(password, hashed):
   h = hashlib.new('sha256')
   h.update(password.encode())
   return bcrypt.checkpw(h.hexdigest().encode(), hashed)

# 信息安全(脱敏)
def mask_string(text):
   if len(text) < 4:
      return "*" * len(text)                          # 如果源小于4个字符，则将其全部屏蔽
   else:
      start_pos = 2 if len(text) >= 10 else 1         # 根据总长度，显示1或2个第一个字符
      end_post = 2 if len(text) > 5 else 1            # 显示最后2个字符，但前提是总长度大于5个字符
      return f"{text[0:start_pos]}****{text[len(text)-end_post:]}"
def mask_email(email):
   text = email.split('@')
   return f"{mask_string(text[0])}@{text[1]}"

#=====================加密解密(暂时无用)=====================#
# Auth_key解密
def decrypt_sdk_authkey(version, message):
   return rsa.decrypt(base64.b64decode(message), rsa.PublicKey.load_pkcs1(get_config()["Key"]["auth_key"][version])).decode()

# 密码rsa私钥解密
def decrypt_rsa_password(message):
   return rsa.decrypt(base64.b64decode(message), rsa.PrivateKey.load_pkcs1(get_config()["Crypto"]["rsa"]["password"])).decode()

#=====================AuthKey解密返回信息=====================#
def decrypt(cipher, PUBLIC_KEY):
   public_key = PublicKey.load_pkcs1(PUBLIC_KEY)
   encrypted = transform.bytes2int(cipher)
   decrypted = core.decrypt_int(encrypted, public_key.e, public_key.n)
   text = transform.int2bytes(decrypted)

   if len(text) > 0 and text[0] == 1:
      pos = text.find(b'\x00')
      if pos > 0:
         return text[pos+1:]
      else:
         return b""

def authkey(auth_key, auth_key_version):
   public_key = get_json_config()["crypto"]["rsa"]["authkey"][auth_key_version]
   result = b""
   for chunk in chunked(256, base64.b64decode(auth_key)):
      result += decrypt(chunk, public_key)
   return result.strip()

#=====================Muip签名计算与配置=====================#
def send(uid, content):
   def query_sha256_sign(query, sign):
      querys = query.split("&")
      sort_querys = [q for q in querys if q.split("=")[1] != ""]
      sort_querys.sort()
      ready_sign_query = "&".join(sort_querys)
      http_sign = sha256_encode(ready_sign_query + sign)
      return http_sign

   def query_escape(query):
      querys = query.split("&")
      new_querys = ["=".join([key_and_value[0], urllib.parse.quote(
         key_and_value[1])]) for key_and_value in (q.split("=") for q in querys)]
      return "&".join(new_querys)

   def sha256_encode(data):
      sha256 = hashlib.sha256()
      sha256.update(data.encode('utf-8'))
      return sha256.hexdigest()

   command = f"cmd=1005&uid={uid}&{content}" + "&ticket=" + "COKESERVER@" + str(time.mktime(datetime.datetime.now().timetuple())).split('.')[0]
   query = query_escape(command)
   http_sign = query_sha256_sign(command, check_config_exists()["Muipserver"]["sign"])
   request = "http://" + check_config_exists()["Muipserver"]["address"] + ":" + str(check_config_exists()["Muipserver"]["port"]) + "/api?" + query + "&sign=" + http_sign
   response = requests.get(request)
   return response.text.strip()
settings/library.py
CokeSR/Hk4e-SDKSever-Py
import yaml
import json
import settings.repositories as repositories

from flask import g

#=====================读取Config=====================#
def load_config():
    with open(repositories.CONFIG_FILE_PATH, encoding='utf-8') as file:
        return yaml.safe_load(file)

def get_config():
    config = getattr(g, '_config', None)
    if config is None:
        with open(repositories.CONFIG_FILE_PATH, encoding='utf-8'):
            config = g._config = load_config()
    return config

def load_json_config():
   with open(repositories.CONFIG_FILE_JSON_PATH) as file:
     return json.load(file)

def get_json_config():
   config = getattr(g, '_config', None)
   if config is None:
      with open(repositories.CONFIG_FILE_JSON_PATH):
        config = g._config = load_json_config()
   return config
settings/loadconfig.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import os
import yaml
import time
import shutil
import atexit
import logging
import settings.repositories as repositories

from flask import request
from datetime import datetime

#======================log设置======================#
log_dir = 'logs'
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, 'sdkserver-running.log')
logging.basicConfig(filename=log_file, level=logging.INFO,
                    format='%(asctime)s|%(levelname)s|%(message)s')

console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

logger = logging.getLogger()
logger.addHandler(console_handler)

def rename_log_file():
    logger.removeHandler(console_handler)
    logging.shutdown()
    console_handler.close()
    time.sleep(1)
    now = datetime.now()
    new_filename = now.strftime("sdkserver-%Y-%m-%d %H-%M-%S.0000")
    new_log_file = os.path.join(log_dir, new_filename)
    shutil.move(log_file, new_log_file)
atexit.register(rename_log_file)

# 加载配置文件
def load_config():
    with open(repositories.CONFIG_FILE_PATH, 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)
    return config

# 获取请求日志记录配置项
def get_request_logging_config():
    config = load_config()
    return config.get('Setting', {}).get('high_frequency_logs', True)

@app.before_request
def log_request_content():
    enable_request_logging = get_request_logging_config()
    if enable_request_logging:
        content = request.get_data(as_text=True)
        encoded_content = content.encode('utf-8')
        logging.info(f"[信息上报]: {encoded_content}")
settings/logoutput.py
CokeSR/Hk4e-SDKSever-Py
# 文件目录
MI18N_PATH = './data/mi18n'
CONFIG_FILE_PATH = "./config.yaml"
CONFIG_FILE_JSON_PATH = "./data/config.json"
GEOIP2_DB_PATH = "./data/GeoLite2-Country.mmdb"
AUTHVERIFY_KEY_PATH = "./data/key/pems/authverify.pem"
PASSWDWORD_KEY_PATH = "./data/key/pems/password.pem"

DISPATCH_KEY = "./data/key/bins/dispatchkey.bin"
DISPATCH_SEED = "./data/key/bins/dispatchseed.bin"

GACHA_TEXTMAP_PATH = './data/gacha/textmap'
GACHA_SCHEDULE_PATH = './data/gacha/schedule'

SHOPWINDOW_TIERS_PATH_CN = "./data/shopwindow/cn/tiers_cn.json"
SHOPWINDOW_PAY_TYPES_PATH_CN = "./data/shopwindow/cn/pay_types_cn.json"
SHOPWINDOW_TIERS_PATH_OS = "./data/shopwindow/os/tiers_os.json"
SHOPWINDOW_PAY_TYPES_PATH_OS = "./data/shopwindow/os/pay_types_os.json"

ANNOUNCE_PATH_INGAME = "./data/announce/game/list.json"
ANNOUNCE_PATH_INGATE = "./data/announce/gate/list.json"
ANNOUNCE_CONTENT_PATH_INGAME = "./data/announce/game/content.json"
ANNOUNCE_CONTENT_PATH_INGATE = "./data/announce/gate/content.json"
ANNOUNCE_BLUE_PATH = "./data/announce/zh-cn.json"
ANNOUNCE_FONT_PATH = "./data/announce/font.json"
ANNOUNCE_JS_PATH1 = "./static/js/announce/2_2e4d2779ad3d19e6406f.js"
ANNOUNCE_JS_PATH2 = "./static/js/announce/vendors_3230378b06826ebc94d3.js"
ANNOUNCE_JS_PATH3 = "./static/js/announce/bundle_9f9d2fd05b63fd8decfc.js"
ANNOUNCE_CSS_PATH1 = "./static/css/announce/2_cb04d2d72d7555e2ab83.css"
ANNOUNCE_CSS_PATH2 = "./static/css/announce/bundle_dad917ca6970b9fa0ec0.css"
ANNOUNCE_FAVICON_PATH = "./static/favicon.ico"
ANNOUNCE_MAINJS_PATH = "./static/js/announce/main.js"
ANNOUNCE_VUEMIN_PATH = "./static/js/announce/vue.min.js"
ANNOUNCE_MAINH5JS_PATH = "./static/js/announce/main-h5log.js"
ANNOUNCE_FPTJS_PATH = "./static/js/announce/firebase-performance-standalone.js"

# 账号类型
ACCOUNT_TYPE_GUEST = 0
ACCOUNT_TYPE_NORMAL = 1

# 官服 B服
CHANNEL_ID_MIHOYO = 1
CHANNEL_ID_BILIBILI = 14

# 登录场景
SCENE_NORMAL = "S_NORMAL"        # 手机号+用户名 默认手机号
SCENE_ACCOUNT = "S_ACCOUNT"      # 手机号+用户名 默认用户名
SCENE_USER = "S_USER"            # 仅账号
SCENE_TEMPLE = "S_TEMPLE"        # 仅账号 无需注册

# 客户端平台
PLATFORM_TYPE = {
   0: "EDITOR",
   1: "IOS",
   2: "ANDROID",
   3: "PC",
   4: "PS4",
   5: "SERVER",
   6: "CLOUD_ANDROID",
   7: "CLOUD_IOS",
   8: "PS5",
   9: "CLOUD_WEB",
   10: "CLOUD_TV",
   11: "CLOUD_MAC",
   12: "CLOUD_PC",
   13: "CLOUD_THIRD_PARTY_MOBILE",
   14: "CLOUD_THIRD_PARTY_PC"
}

# 返回的状态码
RES_SUCCESS = 0
RES_FAIL = -1
RES_CANCEL = -2
RES_NO_SUCH_METHOD = -10
RES_LOGIN_BASE = -100
RES_LOGIN_FAILED = -101
RES_LOGIN_CANCEL = -102
RES_LOGIN_ERROR = -103
RES_LOGOUT_FAILED = -104
RES_LOGOUT_CANCEL = -105
RES_LOGOUT_ERROR = 106
RES_PAY_FAILED = -107
RES_PAY_CANCEL = -108
RES_PAY_ERROR = -109
RES_PAY_LAUNCH = -120
RES_PAY_UNKNOWN = -116
RES_EXIT_FAILED = -110
RES_EXIT_NO_DIALOG = -111
RES_EXIT_CANCEL = -112
RES_EXIT_ERROR = -113
RES_INIT_FAILED = -114
RES_INIT_ERROR = -115
RES_LOGIN_FORBIDDED = -117
RES_NEED_REALNAME = -118
RES_NEED_GUARDIAN = -119
RES_EOS_DLL_ERROR = -1001
RES_EOS_TOKEN_ERROR = -1002
RES_GOOGLE_PC_TOKEN_ERROR = -1003
RES_CDK_EXCHANGE_SUCC = 0
RES_CDK_EXCHANGE_FAIL = -2003
RES_SDK_VERIFY_SUCC = 0
RES_SDK_VERIFY_FAIL = 1

# risky
RISKY_ACTION_NONE = "ACTION_NONE"

settings/repositories.py
CokeSR/Hk4e-SDKSever-Py
from __main__ import app
import json

from flask import Response, render_template

#=====================创建回应=====================#
# 错误处理
@app.errorhandler(404)
def page_not_found(e):
   print(f"ErrorHandler: {e=}, {e.description}")
   return render_template('404.tmpl'), 404


# 自定义json响应
def json_rsp(code, data):
   return Response(json.dumps({"retcode": code} | data, separators=(',', ':')), mimetype='application/json')

def json_rsp_with_msg(code, msg, data):
   return Response(json.dumps({"retcode": code, "message": msg} | data, separators=(',', ':')), mimetype='application/json')

def json_rsp_common(code, msg):
   return Response(json.dumps({"retcode": code, "message": msg}), mimetype='application/json')
settings/response.py
CokeSR/Hk4e-SDKSever-Py
import yaml
import settings.repositories as repositories

#=====================[Config]恢复=====================#
# 对于在根目录下没有config或破损的留一个模板来创建
def recover_config():
    config = {
            'Setting': {
                'listen': '{{%YOUR_SDK_ADDRESS%}}',
                'port': '{{%YOUR_SDK_PORT%}}',
                'reload': False,
                'debug': False,
                'threaded': False,
                'high_frequency_logs': False,
                'cdkexchange': False,
                'secret_key': 'cokeserver2022'
            },
            'Database': {
                'host': '{{%SDK_ADDRESS%}}',
                'user': '{{%YOUR_MYSQL_USER%}}',
                'port': '{{%YOUR_MYSQL_PORT%}}',
                'autocreate': False,
                'account_library_name': 'hk4e-accounts-cokeserver',
                'exchcdk_library_name': 'hk4e-cdk-cokeserver',
                'password': '{{%YOUR_MYSQL_PASSWD%}}'
            },
            'Login': {
                'disable_mmt': False,
                'disable_regist': False,
                'disable_email_bind_skip': False,
                'enable_email_captcha': False,
                'enable_ps_bind_account': False,
                'email_bind_remind': False,
                'email_verify': False,
                'realperson_required': False,
                'safe_mobile_required': False,
                'device_grant_required': False,
                'initialize_firebase': False,
                'bbs_auth_login': False,
                'fetch_instance_id': False,
                'enable_flash_login': False
            },
            'Player': {
                'disable_ysdk_guard': False,
                'enable_announce_pic_popup': False,
                'protocol': False,
                'qr_enabled': False,
                'qr_bbs': False,
                'qr_cloud': False,
                'enable_user_center': False,
                'guardian_required': False,
                'realname_required': False,
                'heartbeat_required': False
            },
            'Reddot': {
                'display': False
            },
            'Announce': {
                'remind': False,
                'alert': False,
                'extra_remind': False
            },
            'Security': {
                'token_length': 32,
                'min_password_len': 8
            },
            'Auth': {
                'enable_password_verify': False,
                'enable_guest': False
            },
            'Other': {
                'modified': False,
                'serviceworker': False,
                'new_register_page_enable': False,
                'kcp_enable': False,
                'enable_web_dpi': False,
                'list_price_tierv2_enable': False
            },
            'Muipserver':{
                'address': '{{%MUIP_IP%}}',
                'region': '{{%SERVER_REGION%}}',
                'port': '{{%MUIP_PORT%}}',
                'sign': '{{%MUIP_SIGN%}}'
            },
            'Dispatch': {
                'list': {
                    'dev_docker': 'http://{{%DISPATCH_ADDRESS%}}:{{%DISPATCH_PORT%}}',
                    'dev_client': 'http://{{%DISPATCH_ADDRESS%}}:{{%DISPATCH_PORT%}}',
                    'dev_common': 'http://{{%DISPATCH_ADDRESS%}}:{{%DISPATCH_PORT%}}'
                }
            },
            'Gateserver': [
                {
                    'name': 'dev_docker',
                    'title': 'live-FormalChannel',
                    'dispatchUrl': 'http://{{%SDK_ADDRESS%}}:{{%YOUR_SDK_PORT%}}/query_region/dev_docker'
                },
                {
                    'name': 'dev_client',
                    'title': 'Dev-TestChannel-1',
                    'dispatchUrl': 'http://{{%SDK_ADDRESS%}}:{{%YOUR_SDK_PORT%}}/query_region/dev_client'
                },
                {
                    'name': 'dev_common',
                    'title': 'Dev-TestChannel-2',
                    'dispatchUrl': 'http://{{%SDK_ADDRESS%}}:{{%YOUR_SDK_PORT%}}/query_region/dev_common'
                }
            ],
            'Mail': {
                'ENABLE': False,
                'MAIL_SERVER': 'smtp.qq.com',
                'MAIL_PORT': 587,
                'MAIL_USE_TLS': False,
                'MAIL_USE_SSL': False,
                'MAIL_USERNAME': '{{%EMAIL_ADDRESS%}}',
                'MAIL_PASSWORD': '{{%EMAIL_STMP_PASSWORD%}}',
                'MAIL_DEFAULT_SENDER': '{{%EMAIL_SENDER%}}'
            }
        }
    config_file_path = repositories.CONFIG_FILE_PATH
    with open(config_file_path, 'wb') as f:         # 使用二进制写入模式 让文件编排原封不动
        yaml.dump(config, f, encoding='utf-8', sort_keys=False)
settings/restoreconfig.py
CokeSR/Hk4e-SDKSever-Py