the server utils of Rock Minecraft server

server_util.py

import datetime
import os
import time
from argparse import ArgumentParser
from typing import List


def send_server_command(command) -> None:
    os.system(f'screen -S rock-server -p 0 -X eval \'stuff "{command}"\\015\'')


def get_newest_log_name():
    list_of_files = os.listdir('run_log')
    path_list = ["run_log/{0}".format(x) for x in list_of_files]

    newest_file = max(path_list, key=os.path.getctime)
    return newest_file


def get_server_run_log(return_lines: int = 0) -> List[str]:
    newest_file = get_newest_log_name()

    with open(newest_file, 'r', encoding='utf-8') as f:
        lines = f.read().split('\n')
    lines = [x.strip() for x in lines]
    lines = list(filter(None, lines))
    for i in range(len(lines)):
        if lines[i].startswith(''):
            lines[i] = lines[i][3:]

    if return_lines == 0:
        return lines
    return lines[-return_lines:]


def server_cmd(command, return_lines: int = 3) -> List[str]:
    send_server_command(command)
    time.sleep(0.5)
    return get_server_run_log(return_lines)


def wait_server_stop() -> None:
    start_time = time.time()
    while True:
        end_time = time.time()
        if end_time - start_time >= 3 * 60:
            # 超過三分鐘還沒關機，啟動強制關閉程序
            check_result = os.popen('ps aux | grep jar').read()
            if 'Xms' in check_result and 'Xmx' in check_result:
                os.system('sudo reboot')
            else:
                # server is closed
                break

        check_result = os.popen('ps aux').read()
        if 'Xms' in check_result or 'Xmx' in check_result:
            print('Wait for server close..')
            time.sleep(3)
        else:
            print('server close!')
            break


def wait_service_stop() -> None:
    while True:
        check_result = os.popen('ps aux').read()
        if 'Xms' in check_result or 'Xmx' in check_result or 'start_service.py' in check_result:
            print('Wait for server close..')
            time.sleep(3)
        else:
            print('server close!')
            break


def backup_map(force: bool = False):
    backup_root = '/data/backup'
    mc_root = f'/home/codingman/server'

    now = datetime.datetime.now()

    date_mark = now.strftime("%Y-%m-%d")
    backup_path = f'{backup_root}/{datetime.datetime.today().weekday()}'

    if os.path.exists(f"{backup_path}/{date_mark}") and not force:
        # backup_path should not exist
        return
    os.system(f'touch {backup_path}/{date_mark}')
        
    for count_down in reversed(range(0, 31)):
        send_server_command(f'say 伺服器即將開始備份，倒數...{count_down}')

        time.sleep(1)

    send_server_command('kickall')
    time.sleep(3)
    send_server_command('stop')
    wait_server_stop()

    # rsync -auvz world/ /data/backup/qq/world/

    os.system(f'mkdir -p {backup_path}')

    backup_folders = ['world', 'world_nether', 'world_the_end']
    for backup_folder in backup_folders:
        backup_folder_path = f'{backup_path}/{backup_folder}'
        # print(backup_world_path)
        os.system(f'mkdir -p {backup_folder_path}')
        os.system(f'rsync -avh {mc_root}/{backup_folder}/ {backup_folder_path}')

    # delete old folder

    # for day in range(8, 15):
    #     old_day = datetime.datetime.now() - datetime.timedelta(days=day)
    #     old_day_folder = old_day.strftime('%Y-%m-%d')
    #     old_day_backup_path = f'{backup_root}/{old_day_folder}'
    #
    #     if not os.path.exists(old_day_backup_path):
    #         continue
    #
    #     os.system(f'rm -r {old_day_backup_path}')

    os.system('sudo reboot')
    time.sleep(30)


def reset_command():
    try:
        os.remove('/tmp/server_close')
    except FileNotFoundError:
        pass


def is_in_time_range(start, end):
    now = datetime.datetime.now()

    if start <= end:
        return start <= now <= end
    else:
        return start <= now or now <= end


if __name__ == '__main__':
    # parser = ArgumentParser()
    # parser.add_argument('--test', help="test", action="store_true")
    # parser.add_argument('--backup', help="backup maps", action="store_true")
    # args = parser.parse_args()

    # if args.test:

    #     now = datetime.datetime.now()
    #     for day in range(8, 15):
    #         old_day = now - datetime.timedelta(days=day)
    #         old_day_folder = old_day.strftime('%Y-%m-%d')
    #         old_day_backup_path = f'/data/backup/{old_day_folder}'

    #         print(old_day_backup_path)
    # elif args.backup:
    #     backup_map(force=True)

    backup_day = [
        # Monday and friday
        0, 4
    ]

    print(datetime.datetime.today().weekday())
    print(type(datetime.datetime.today().weekday()))
    print(datetime.datetime.today().weekday() in backup_day)