峰间云海服务器技术分析

PLUGIN_METADATA = {
    'id': 'cleanitem',
    'version': '1.0.0',
    'name': '扫地僧'
}


from mcdreforged.api.all import *
import time

def clear_drops(server: PluginServerInterface):
    server.execute('tellraw @a {"text":"30秒后清除掉落物","color":"yellow"}')
    time.sleep(30)
    server.execute('kill @e[type=item]')
    server.execute('tellraw @a {"text":"掉落物已清除!","color":"yellow"}')

def on_info(server: PluginServerInterface, info: Info):
    if info.content == '!!cleardrops':
        clear_drops(server)

@new_thread(PLUGIN_METADATA['id'])
def on_load(server, old):
    while True:
        clear_drops(server)
        time.sleep(86400)

PLUGIN_METADATA = {
    'id': 'talksave',
    'version': '1.0.0',
    'name': '记录玩家聊天记录'
}

#配置区域
DATABASE = {
    'host': '127.0.0.1',
    'user': 'user',
    'password': 'password',
    'database': 'database'
}

import mysql.connector
from mcdreforged.api.all import *
# def on_load(server, old):
#     server.logger.info('Hello world!')
def on_user_info(server: PluginServerInterface, info: Info):
    # 获取玩家名和聊天消息内容
    player = info.player
    message = info.content
    #server.logger.info(f'{player} 发送了消息：{message}')
    if player is None or message is None:
        return
    if message.startswith('!!MCDR') or message.startswith('!!plp') or message.startswith('!!day') or message =="stop" or message =="" or message.startswith('!!qb') or message.startswith('tellraw') or message.startswith('list'):
        return
    # 此部分判断是否为MCDR命令，可根据自己插件不同调整
    # 也可与nb联动，同步信息到qq，不过我手抖把代码删了
    conn= mysql.connector.connect(**DATABASE)
    cursor = conn.cursor()
    sql = "INSERT INTO chat (playername, msg, time) VALUES (%s, %s, NOW())"
    val = (player, message)
    cursor.execute(sql, val)
    conn.commit()
    cursor.close()

@app.route('/user/msg', methods=['GET'])
def getmsg():
    num = request.args.get('num', default=0, type=int)
    # 限制访问前200条记录
    if num > 160:
        messages = []
        message = {
            'id': 0,
            'playername': "Null",
            'msg': "200条后不予查看",
            'time': "Null"
        }
        messages.append(message)
        return jsonify(messages), 200
    conn= mysql.connector.connect(**DATABASE)
    cursor = conn.cursor()
    sql = "SELECT id, playername, msg, time FROM chat ORDER BY id DESC LIMIT %s, %s"
    cursor.execute(sql, (num, 40))
    result = cursor.fetchall()
    cursor.close()

    messages = []
    for row in result:
        message = {
            'id': row[0],
            'playername': row[1],
            'msg': row[2],
            'time': row[3].strftime('%Y-%m-%d %H:%M:%S')
        }
        messages.append(message)

    return jsonify(messages), 200

@app.route('/user/photo', methods=['GET'])
def get_data():
    num = request.args.get('num', default=0, type=int)
    
    conn= mysql.connector.connect(**DATABASE)
    cursor = conn.cursor()
    
    try:
        query = "SELECT id, nickname, author, time, url FROM photo ORDER BY id DESC LIMIT %s, %s"
        cursor.execute(query, (num, 24))
        results = cursor.fetchall()
        
        # 将查询结果转换为字典列表
        data = []
        for row in results:
            data.append({
                'id': row[0],
                'nickname': row[1],
                'author': row[2],
                'time': row[3],
                'url': row[4]
            })
        
        return jsonify(data)
    
    except Exception as e:
        return jsonify({'error': str(e)})
    
    finally:
        cursor.close()
        conn.close()

import mysql.connector
import python_nbt.nbt as nbt
import json
import time

def get_scoreboard(scoreboard_name):
# 将字符串转换为 Python 对象
    start_time = time.time()
    file = nbt.read_from_nbt_file("D:\MCC\mcdr_server\server\world\data\scoreboard.dat")
    string_data = str(file)
    # print(string_data)
    python_object = ast.literal_eval(string_data)

    # 将 Python 对象转换为 JSON 字符串
    json_data = json.dumps(python_object)
    data = json.loads(json_data)
    # print(json_data)
    # print(data["value"]["data"]["value"]["PlayerScores"]["value"])
    data = data["value"]["data"]["value"]["PlayerScores"]["value"]
    for i in data:
        if i["Objective"]["value"] == scoreboard_name:
            name = i["Name"]["value"]
            score = i["Score"]["value"]
            #print(name,":",score)
            conn= mysql.connector.connect(**DATABASE)
            cursor = conn.cursor()
            
            try:
                insert_query = "INSERT INTO scoreboard (player, {}) VALUES (%s, %s) ON DUPLICATE KEY UPDATE {} = %s".format(scoreboard_name, scoreboard_name)
                data = (name, score, score)
                cursor.execute(insert_query, data)

                conn.commit()
            
            except Exception as e:
                print(str(e))
            
            finally:
                cursor.close()
                conn.close()


    # 记录结束时间
    end_time = time.time()

    # 计算执行时间
    execution_time = end_time - start_time

    # 打印执行时间
    return(execution_time)

import threading
import time
import extra

def updatabase():
    time_used = extra.get_scoreboard("pickaxe_total")
    time_used = time_used + extra.get_scoreboard("gametime")
    time_used = time_used + extra.get_scoreboard("Kills")
    time_used = time_used + extra.get_scoreboard("fish")
    time_used = time_used + extra.get_scoreboard("xp")
    time_used =" {:.3f}".format(time_used)
    #msg_type = "System-AutoTasks"
    #msg_value = "计分板同步成功，耗时:" + str(time_used) + "秒"
    #extra.eventssave(msg_type,msg_value)
    # 用于日志保存，可去掉
                          
                  
def run_timer():
    while True:
        # 每2h执行一次
        # 简单粗暴，实际使用请使用schedule库
        time.sleep(2* 60 * 60)
        # 在后台线程中执行需要定时执行的函数
        threading.Thread(target=updatabase).start()

@app.route('/user/scoreboard')
def get_scoreboard():
    cnx = mysql.connector.connect(**DATABASE)
    cursor = cnx.cursor()
    query = "SELECT * FROM scoreboard"
    cursor.execute(query)
    data = [{'playername': row[0], 'pickaxe_total': row[1], 'gametime': row[2], 'Kills': row[3], 'fish': row[4], 'xp': row[5]} for row in cursor.fetchall()]
    cursor.close()
    return jsonify(data)

@app.route('/admin/banplayer', methods=['POST'])
def banplayer():
    playername = request.json.get('playername')
    reason = request.json.get('reason')
	...
    if reason == "":
        reason = "未记录原因的封禁，请联系管理员"

    command = 'kick ' + playername + ' ' + reason
    response = extra.sendcommand(command)
    command = 'ban ' + playername + ' ' + reason
    response = extra.sendcommand(command)
    conn = mysql.connector.connect(**DATABASE)
	...

    return jsonify({'code':0,'response': response}), 200

import mcrcon

# 主机 密码 端口
client = mcrcon.MCRcon('127.0.0.1', '*-*-*-*', 2****5)
client.connect()
def sendcommand(command):
    if not command:
        return {'error': 'No command provided'}
    
    response = client.command(command)
    return response

@app.route('/admin/debanplayer', methods=['POST'])
def debanplayer():
    playername = request.json.get('playername')
	...
    command = 'pardon ' + playername
    response = extra.sendcommand(command)
    conn = mysql.connector.connect(**DATABASE)
	...
    return jsonify({'code':0,'response': response}), 200

import ipdb

@app.route('/admin/getaccount', methods=['POST'])
def getaccount():
    accountinfo = request.json.get('accountinfo')
    mode = int(request.json.get('mode'))
	...
    conn = mysql.connector.connect(**DATABASE)
    c = conn.cursor()
    if mode == 1:
        c.execute("SELECT * FROM `qq_user` where qq_id = %s", (accountinfo,))
    elif mode == 2:
        c.execute("SELECT * FROM `qq_user` where user_name = %s", (accountinfo,))
    elif mode == 3:
        c.execute("SELECT * FROM `qq_user` where be_name = %s", (accountinfo,))
    else:
        c.execute("SELECT * FROM `qq_user` where qq_id = %s", (accountinfo,))
    acc_result = c.fetchall()

    
    if acc_result:
        if acc_result[0][5] is None:
            be_name = "!未绑定!"
        else:
            be_name = acc_result[0][5]
        playername = acc_result[0][1]
        # 用的MySQL数据库，Authme和该服务共用一个数据库，因此直接调用其获取IP
        c.execute("SELECT * FROM `authme` where realname = %s or username = %s", (playername,playername,))
        ip_result = c.fetchall()
        conn.close()
        #嵌套有点多，整活部分可以改掉
        if ip_result:
            ip = ip_result[0][4]
            if ipaddress.ip_address(ip).is_private:
                ip = "内网地址"
                country="南极洲"
                city="中山站"
            else:
                ipinfo=db.find_map(ip, "CN")
                country=ipinfo["country_name"]
                city= ipinfo["region_name"]+"，"+ipinfo["city_name"]
        else:
            ip = "!未记录在数据库!"
            country="北极点"
            city="鹦鹉螺号"
        return jsonify({'code':0,'qq': acc_result[0][0],'user_name': playername,'bind_time': acc_result[0][2],'group': acc_result[0][3],'status': acc_result[0][4],'be_name': be_name,'ip': ip,'country':country,'city':city}), 200
    else:
        conn.close()
        return jsonify({'code':1000,'msg': '没有相应玩家消息'}), 200

@bind.handle()
async def handle_first_receive(matcher: Matcher, args: Message = CommandArg()):
    plain_text = args.extract_plain_text()  # 首次发送命令时跟随的参数，例：/绑定 username，则args为username
    if plain_text:
        matcher.set_arg("username", args)  # 如果用户发送了参数则直接赋值


@bind.got("username", prompt="请输入你的用户名")
async def handle_username(event: GroupMessageEvent, username: Message = Arg(), username_str: str = ArgPlainText("username")):
    username = str(username)
    user_id = event.user_id

    pattern = r'^[a-zA-Z0-9_]+$'
    if re.match(pattern, username):
        pass
    else:
        extra.settimeout()
        await bind.finish(f"用户名只可以包括英文字母，数字，下划线")
	...
    binduser(username)
	...

    await bind.finish(f"{username} 绑定成功（若仍无法进入服务器，可尝试发送“正常流程绑定”）")
    # 即用whitelist add
            
def binduser(playername):
    offline_uuid = str(extra.touuid(playername))
    #offline_uuid = offline_uuid_raw[:8] + "-" + offline_uuid_raw[8:12] + "-" + offline_uuid_raw[12:16] + "-" + offline_uuid_raw[16:20] + "-" + offline_uuid_raw[20:]
    #读取 ***\whitelist.json,找到playername对应的uuid，用offline_uuid替换
    whitelist_file_path = r"Z:\server\whitelist.json"
    with open(whitelist_file_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    # 先尝试是否能找到对应玩家名称的条目，并替换uuid字段，如果没有，则新增
    found = False # 定义一个变量，用来标记是否找到了对应的玩家条目
    for player in data:
        if player["name"] == playername:
            player["uuid"] = offline_uuid # 替换uuid字段
            found = True # 将变量设为True，表示找到了对应的玩家条目
            break # 跳出循环，不再继续查找
    if not found: # 如果变量为False，表示没有找到对应的玩家条目
        data.append({"name": playername, "uuid": offline_uuid}) # 在data列表中添加一个新的字典，包含玩家的名字和uuid

    # 将修改后的内容写回whitelist.json文件
    with open(whitelist_file_path, "w", encoding="utf-8") as f:
        json.dump(data, f)
    # 调用外部命令来重载白名单
    client.connect()
    command = "whitelist reload"
    response = client.command(command)
    client.disconnect()

import hashlib
import uuid

#该函数用于计算离线用户的uuid
def touuid(name):
    name = "OfflinePlayer:" + name
    md5 = hashlib.md5()
    md5.update(name.encode('utf-8'))
    md5_bytes = md5.digest()

    md5_bytes = bytearray(md5_bytes) 

    md5_bytes[6] &= 0x0f
    md5_bytes[6] |= 0x30
    md5_bytes[8] &= 0x3f
    md5_bytes[8] |= 0x80 

    return uuid.UUID(bytes=bytes(md5_bytes))

def debinduser(playername):
    offline_uuid = str(extra.touuid(playername))
    whitelist_file_path = r"Z:\server\whitelist.json"
    with open(whitelist_file_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    for player in data:
        if player["name"] == playername:
            data.remove(player) # 移除该条目

    # 将修改后的内容写回whitelist.json文件
    with open(whitelist_file_path, "w", encoding="utf-8") as f:
        json.dump(data, f)
    # 调用外部命令来重载白名单
    client.connect()
    command = "whitelist reload"
    response = client.command(command)
    client.disconnect()

upphoto = on_message(rule=to_me(), priority=5, block=True)

@upphoto.handle()
async def handle_upphoto(bot: Bot, event: GroupMessageEvent, state: T_State):
    user_id = event.user_id
    message = str(event.get_message())
    conn = mysql.connector.connect(**DATABASE)

    # print(message)
    if message:
        if matches:
            pattern = r"\[CQ:image,file=\w+\.image,subType=\d+,url=(https?://\S+?)]"
            matches = re.findall(pattern, message)
            current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))
            # print(matches)
            for match in matches:
                # print(match)
                
                file_name =  str(user_id) + str(time.time() * 1000)+".jpg"
                save_path = './tmp/' + file_name
                response = requests.get(match, stream=True)
                if response.status_code == 200:
                    with open(save_path, 'wb') as file:
                        for chunk in response.iter_content(1024):
                            file.write(chunk)
                    path = "Z:\\nb2\\mc\\tmp\\" + file_name
                    response = extra.upload_image(path)
                    if response.startswith("https://"):
                        url = response
                        c = conn.cursor()
                        c.execute("INSERT INTO photo VALUES (%s,%s,%s,%s,%s)", (None, str(event.sender.card),user_id, current_time, url))
                        conn.commit()
                    else:
                        conn.disconnect()
                        await upphoto.finish("上传失败（调用上传工具错误）")
                    os.remove(path)
                else:
                    conn.disconnect()
                    await upphoto.finish("上传失败(请求图片错误)")
            conn.disconnect()
            await upphoto.finish("上传成功")
        else:
            conn.disconnect()
            return  # 没有图片链接，不执行回复操作

from nonebot import on_message, logger
from nonebot.adapters.onebot.v11 import GroupMessageEvent, Bot,MessageSegment
from nonebot.rule import startswith
from nonebot import get_driver
import mcrcon
import threading
import time
import re



msg_matcher = on_message(priority=10, block=False)
client = mcrcon.MCRcon('127.0.0.1', '******', 2***5)



def sendcommand(command):
    response = client.command(command)
    return response

@msg_matcher.handle()
async def _(bot: Bot, event: GroupMessageEvent):
    if event.group_id == 790118052:
        nick = str(event.sender.card)
        if nick == "":
            nick = str(event.sender.nickname)
        msg = nick+":"+replace(event.message)
        qqqq = "[Q群]"
        msg =f'[{{"text":"{qqqq}","color":"gold"}},{{"text":"{msg}","color":"white"}}]'
        command=f"tellraw @a [{msg}]"
        # print(command)
        sendcommand(command)

# 定义一个函数，接受一个文本参数，返回替换后的文本
def replace(text):
  text = str(text)

  image_pattern = r"\[CQ:image,file=.+?image,subType=0,url=.+?\]"

  at_pattern = r"\[CQ:at,qq=\d+\]"

  picface_pattern = r"\[CQ:image,file=.+?image,subType=1,url=.+?\]"
  face_pattern = r"\[CQ:face,id=.+?]"

  record_pattern = r"\[CQ:record,file=.+?amr,url=.+?\]"
  redbag_pattern = r"\[CQ:redbag,title=.+?\]"
  shit_pattern = r"请使用最新版手机QQ体验新功能"

  result = re.sub(image_pattern, "[图片]", text)

  result = re.sub(picface_pattern, "[图片表情]", result)
  result = re.sub(face_pattern, "[表情]", result)
  result = re.sub(at_pattern, "@", result)

  result = re.sub(record_pattern, "[语音]", result)

  result = re.sub(redbag_pattern, "[红包]", result)

  result = re.sub(shit_pattern, "不支持的信息类型", result)
  result = result.replace("<", "《").replace(">", "》")
  # 返回替换后的结果
  return result

def connectagain():
    client.connect()


def run_timer():
    while True:
        # 每5分钟执行一次
        time.sleep(5 * 60)
        # 在后台线程中执行需要定时执行的函数
        threading.Thread(target=connectagain).start()
# 避免重复执行使得后台刷屏
connectagain()
timer_thread = threading.Thread(target=run_timer)
timer_thread.daemon = True
timer_thread.start()

import subprocess
import psutil
import time
from mcstatus import JavaServer
import redis

BALANCE = '381b4222-f694-41f0-9685-ff5bb260df2e'
HIGH_PERFORMANCE = '8c5e7fda-e8bf-4a96-9a85-a6e23a8c635c'
ENERGY_SAVER = 'a1841308-3541-4fab-bc81-f71556f20b4a'
redis_client = redis.Redis(host='localhost', port=6379, db=1)

server = JavaServer.lookup("127.0.0.1:26565")
people = 1
def change_power_plan(plan):
    subprocess.run(f'powercfg /S {plan}')

def get_power_plan():
    output = subprocess.run("powercfg /GETACTIVESCHEME", shell=True, stdout=subprocess.PIPE).stdout.decode("GBK")
    return output.split(" ")[-3]
def what_power_plan(plan):
    if plan == BALANCE:
        return "BALANCE"
    elif plan == ENERGY_SAVER:
        return "ENERGY_SAVER"
    elif plan == HIGH_PERFORMANCE:
        return "HIGH_PERFORMANCE"
    else:
        return "Unknown"
def what_power_planid(plan):
    if plan == "BALANCE":
        return BALANCE
    elif plan == "ENERGY_SAVER":
        return ENERGY_SAVER
    elif plan == "HIGH_PERFORMANCE":
        return HIGH_PERFORMANCE
    else:
        return "Unknown"
while True:
    current_time = time.strftime("%H:%M:%S", time.localtime())
    plan = get_power_plan()
    mcserve_plan = redis_client.get('powerplan')
    if mcserve_plan is None:
        cpu_percent = psutil.cpu_percent()
        status = server.status()
        hour = int(current_time.split(":")[0])
        if hour >= 0 and hour <= 5:
            if plan != ENERGY_SAVER:
                print(str(current_time) + "：切换到节能模式")
                change_power_plan(ENERGY_SAVER)

        elif status.players.online > 3:
            if plan != HIGH_PERFORMANCE:
                print(str(current_time) + "：多人，切换到高性能模式")
                change_power_plan(HIGH_PERFORMANCE)
        elif status.players.online > people:
            if plan != BALANCE:
                print(str(current_time) + "：有人，切换到平衡模式")
                change_power_plan(BALANCE)
        elif cpu_percent > 70:
            if plan != HIGH_PERFORMANCE:
                print(str(current_time) + "：CPU占用过高，切换到平衡模式")
                change_power_plan(BALANCE)
        elif status.players.online <= people:
            if plan != ENERGY_SAVER:
                print(str(current_time) + "：无人，切换到节能模式")
                change_power_plan(ENERGY_SAVER)
        else:
            if plan != BALANCE:
                print(str(current_time) + "：切换到平衡模式")
                change_power_plan(BALANCE)
    if mcserve_plan is not None:
        decoded_mcserve_plan = mcserve_plan.decode('utf-8')
        if what_power_plan(plan) == decoded_mcserve_plan:
            continue
        else:
            power_plan_id = what_power_planid(decoded_mcserve_plan)
            change_power_plan(power_plan_id)
            print(str(current_time) + "：MCSERVE覆盖配置，切换到" + decoded_mcserve_plan + "计划")

    time.sleep(300)