Redis、日志与工程维护脚本

来源脚本集：参数、游戏 ID、日志格式和路径依赖原环境，尚未逐段运行验证。请按目标环境检查后使用。

redis 获取所有 key 并且打印输出

import redis
import datetime

# 设置 Redis 连接参数
host = '192.0.2.10'
port = 6000
password = ''
db = 12  # 指定要扫描的数据库编号
pattern = '2016:hide:*'  # 匹配以 '2016:' 开头的键

# 创建 Redis 客户端
redis_client = redis.Redis(host=host, port=port, password=password, db=db)

# 切换到指定的数据库
redis_client.select(db)

def get_hide(redis_client, keys, output_file):
    # 打开文件以追加写入
    with open(output_file, 'a') as file:
        for key in keys:
            values = redis_client.hmget(key, "W", "L")
            w = int(values[0]) if values[0] is not None else None
            l = int(values[1]) if values[1] is not None else None
            split_value = key.split(":")
            if w is not None and l is not None:
                r = w / l
                file.write(f"playerId: {split_value[2]} groupId: {split_value[3]} win: {w} lose: {l} r: {r}\n")
                print(f"playerId: {split_value[2]} groupId: {split_value[3]} win: {w} lose: {l} r: {r}")

# 扫描并打印以 '2016:' 开头的键
matching_keys = redis_client.scan_iter(match=pattern, count=1000)
matching_keys = [key.decode() for key in matching_keys]

# 获取当前时间
current_time = datetime.datetime.now()
# 将当前时间格式化为字符串
formatted_time = current_time.strftime("%Y%m%d%H%M%S")

get_hide(redis_client, matching_keys, f"hide_{formatted_time}.txt")

日志分析

import os
import re
from collections import defaultdict

def search_files(directory, pattern):
    matches = []
    for root, dirs, files in os.walk(directory):
        for file in files:
            file_path = os.path.join(root, file)
            try:
                with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
                    matches.extend(re.findall(pattern, f.read()))
            except UnicodeDecodeError:
                print(f"Error decoding file: {file_path}")
    return matches

def rate(directory, pattern):
    results = defaultdict(dict)
    matches = search_files(directory, pattern)
    for match in matches:
        pId, C, W, L, rate = match
        results[pId].update({"C": C, "W": W, "L": L, "rate": round(float(rate), 4)})
    return results

def jackpot_buy(directory, pattern):
    results = defaultdict(int)
    matches = search_files(directory, pattern)
    for match in matches:
        pId, number = match
        results[pId] += int(number)
    return results

def jackpot_win(directory, pattern):
    results = defaultdict(int)
    matches = search_files(directory, pattern)
    for match in matches:
        pId, win = match
        results[pId] += int(win)
    return results

def jackpot_win_type(directory, pattern, output_file):
    matches = search_files(directory, pattern)
    with open(output_file, 'w', encoding='utf-8') as f_out:
        for match in matches:
            cardType, winType, number = match
            result = f"cardType: {cardType} winType: {winType} number: {number}\n"
            f_out.write(result)

def merge_data(rates, jackpot_buy, jackpot_win):
    merged_data = defaultdict(dict)
    for pId, data in rates.items():
        merged_data[pId].update(data)
    for pId, buy in jackpot_buy.items():
        merged_data[pId]['buy'] = buy
    for pId, win in jackpot_win.items():
        merged_data[pId]['win'] = win
    return merged_data

def write_to_file(data, output_file):
    with open(output_file, 'w', encoding='utf-8') as f:
        for pId, values in data.items():
            line = f"pId: {pId} {' '.join(f'{key}: {value}' for key, value in values.items())}\n"
            f.write(line)

rate_search = r"PlayerExit pId:(\d+) C:(\d+) W:(\d+) L:(\d+) rate:([\d.]+)"
jackpot_buy_search = r"BugJp:(\d+) number:(\d+)"
jackpot_win_search = r"pId:(\d+) jpWin:(\d+)"
jackpot_win_type_search = r"JackpotWinType cardType:(\d+) winType:(\d+) number:(\d+)"

# jackpot
jackpot_win_type("./logs/bonustexas_a_1/", jackpot_win_type_search, "./all/1-jackpot_win_type.txt")
jackpot_win_type("./logs/bonustexas_a_2/", jackpot_win_type_search, "./all/2-jackpot_win_type.txt")
jackpot_win_type("./logs/bonustexas_a_3/", jackpot_win_type_search, "./all/3-jackpot_win_type.txt")
jackpot_win_type("./logs/bonustexas_a_4/", jackpot_win_type_search, "./all/4-jackpot_win_type.txt")

directory_1 = './logs/bonustexas_a_1/'  # 当前目录路径
output_file_1 = './all/1-all.txt'  # 输出文件名

directory_2 = './logs/bonustexas_a_2/'  # 当前目录路径
output_file_2 = './all/2-all.txt'  # 输出文件名

directory_3 = './logs/bonustexas_a_3/'  # 当前目录路径
output_file_3 = './all/3-all.txt'  # 输出文件名

directory_4 = './logs/bonustexas_a_4/'  # 当前目录路径
output_file_4 = './all/4-all.txt'  # 输出文件名

rates_data_1 = rate(directory_1, rate_search)
jackpot_buy_data_1 = jackpot_buy(directory_1, jackpot_buy_search)
jackpot_win_data_1 = jackpot_win(directory_1, jackpot_win_search)
merged_data = merge_data(rates_data_1, jackpot_buy_data_1, jackpot_win_data_1)
write_to_file(merged_data, output_file_1)

rates_data_2 = rate(directory_2, rate_search)
jackpot_buy_data_2 = jackpot_buy(directory_2, jackpot_buy_search)
jackpot_win_data_2 = jackpot_win(directory_2, jackpot_win_search)
merged_data = merge_data(rates_data_2, jackpot_buy_data_2, jackpot_win_data_2)
write_to_file(merged_data, output_file_2)

rates_data_3 = rate(directory_3, rate_search)
jackpot_buy_data_3 = jackpot_buy(directory_3, jackpot_buy_search)
jackpot_win_data_3 = jackpot_win(directory_3, jackpot_win_search)
merged_data = merge_data(rates_data_3, jackpot_buy_data_3, jackpot_win_data_3)
write_to_file(merged_data, output_file_3)

rates_data_4 = rate(directory_4, rate_search)
jackpot_buy_data_4 = jackpot_buy(directory_4, jackpot_buy_search)
jackpot_win_data_4 = jackpot_win(directory_4, jackpot_win_search)
merged_data = merge_data(rates_data_4, jackpot_buy_data_4, jackpot_win_data_4)
write_to_file(merged_data, output_file_4)

Git Pull

#!/bin/bash

function check_and_pull_git_repositories() {
    for repo in $(find "$1" -type d -name '.git'); do
        repo_dir=$(dirname "$repo")
        # 切换到Git仓库所在的目录
        cd "$repo_dir" || continue
        # 执行git pull命令
        if ! git pull; then
            echo "执行git pull时出错：$repo_dir"
        fi
    done
}

# 指定要检查的目录
directory_to_check="/tmp"

check_and_pull_git_repositories "$directory_to_check"

Grafana 配置备份

import os
import requests
import json

def export_dashboard_config(grafana_url, api_key, folder_path):
    # 设置API请求头
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }

    # 发送API请求获取仪表盘列表
    response = requests.get(f'{grafana_url}/search', headers=headers)
    if response.status_code == 200:
        dashboards = response.json()

        # 遍历仪表盘列表
        for dashboard in dashboards:
            dashboard_uid = dashboard['uid']
            dashboard_title = dashboard['title']

            # 发送API请求获取仪表盘配置
            response = requests.get(f'{grafana_url}/dashboards/uid/{dashboard_uid}', headers=headers)
            if response.status_code == 200:
                dashboard_data = response.json()

                # 提取仪表盘配置
                dashboard_config = dashboard_data['dashboard']

                # 在这里执行导出仪表盘配置的操作
                export_dashboard(dashboard_config, folder_path, dashboard_title)

def export_dashboard(dashboard_config, folder_path, dashboard_title):
    # 创建文件夹（如果不存在）
    if not os.path.exists(folder_path):
        os.makedirs(folder_path)

    # 生成导出文件路径
    file_path = os.path.join(folder_path, f'{dashboard_title}.json')

    # 将仪表盘配置保存到文件
    with open(file_path, 'w') as f:
        json.dump(dashboard_config, f, indent=4)

# 调用函数并传入Grafana URL、API密钥和文件夹路径
grafana_url = 'http://localhost/api'
api_key = 'api'
folder_path = 'path'

export_dashboard_config(grafana_url, api_key, folder_path)