libs.functions.tools.gen_test_data

 1"""
 2libs/functions/tools/gen_test_data.py
 3"""
 4
 5import itertools
 6import logging
 7import random
 8from contextlib import closing
 9from datetime import datetime
10from typing import cast
11
12from tqdm import tqdm
13
14import libs.global_value as g
15from cls.score import GameResult
16from cls.timekit import ExtendedDatetime as ExtDt
17from libs.functions import configuration
18from libs.functions.tools import score_simulator
19from libs.utils import dbutil
20
21
22def main(season_times: int = 1):
23    """テスト用ゲーム結果生成処理
24
25    Args:
26        season_times (int, optional): 総当たり回数. Defaults to 1.
27    """
28
29    configuration.read_memberslist(log=False)
30
31    # 対戦組み合わせ作成
32    teams: list = [x["team"] for x in g.team_list]
33    position: list = ["先鋒", "次鋒", "中堅", "副将", "大将"]
34    teams_data: dict = {x["team"]: x["member"] for x in g.team_list}
35    matchup: list = list(itertools.combinations(teams, 4))
36    teams_count: dict = {x: 0 for x in teams}
37    total_count: int = 0
38
39    now = datetime.now().timestamp() - ((len(matchup) + 7) * 86400 * season_times)
40    dt = now
41
42    with closing(dbutil.get_connection()) as cur:
43        cur.execute("delete from result;")
44        for season in range(1, season_times + 1):
45            random.shuffle(matchup)
46            for count, game in tqdm(enumerate(matchup), total=len(matchup), desc=f"season({season}/{season_times})"):
47                # 対戦メンバーの決定
48                vs_member = {}
49                for team_name in game:
50                    teams_count[team_name] += 1
51                    team_member = teams_data[team_name]
52                    random.shuffle(team_member)
53                    vs_member[team_name] = team_member
54
55                # 試合結果
56                total_count += 1
57                team_name = list(game)
58                random.shuffle(team_name)
59                for idx in range(5):
60                    # 席順シャッフル
61                    member = [
62                        vs_member[team_name[0]][idx],
63                        vs_member[team_name[1]][idx],
64                        vs_member[team_name[2]][idx],
65                        vs_member[team_name[3]][idx],
66                    ]
67                    random.shuffle(member)
68
69                    # スコアデータ生成
70                    dt = now + total_count * 86400 + idx * 3600 + random.random()
71                    vs_score = score_simulator.simulate_game()
72                    result = GameResult(ts=str(dt))
73                    result.calc(
74                        p1_name=member[0], p1_str=str(int(vs_score[0] / 100)),
75                        p2_name=member[1], p2_str=str(int(vs_score[1] / 100)),
76                        p3_name=member[2], p3_str=str(int(vs_score[2] / 100)),
77                        p4_name=member[3], p4_str=str(int(vs_score[3] / 100)),
78                        comment=f"第{season:02d}期{count + 1:04d}試合_{position[idx]}戦",
79                        rule_version=g.cfg.mahjong.rule_version
80                    )
81
82                    # データ投入
83                    param = {
84                        "playtime": ExtDt(dt).format("sql"),
85                    }
86                    param.update(cast(dict, result.to_dict()))
87                    cur.execute(g.sql["RESULT_INSERT"], param)
88
89                    output = f"{position[idx]}: "
90                    output += result.to_text("detail")
91                    logging.debug(output)
92
93        cur.commit()
94
95    with closing(dbutil.get_connection()) as cur:
96        rows = cur.execute("select name, round(sum(point), 1) as point from team_results group by name order by point desc;")
97        logging.notice(dict(rows.fetchall()))  # type: ignore