libs.commands.results.detail

  1"""
  2libs/commands/results/detail.py
  3"""
  4
  5import re
  6import textwrap
  7from typing import cast
  8
  9import pandas as pd
 10
 11import libs.global_value as g
 12from cls.timekit import ExtendedDatetime as ExtDt
 13from cls.types import GameInfoDict
 14from libs.data import aggregate, loader, lookup
 15from libs.functions import message
 16from libs.utils import formatter, textutil
 17
 18
 19def aggregation():
 20    """個人/チーム成績詳細を集計して返す
 21    Returns:
 22        dict: slackにpostするデータ
 23    """
 24
 25    # 検索動作を合わせる
 26    g.params.update(guest_skip=g.params.get("guest_skip2"))
 27
 28    if g.params["player_name"] in lookup.internal.get_team():
 29        g.params.update(individual=False)
 30    elif g.params["player_name"] in g.member_list:
 31        g.params.update(individual=True)
 32
 33    # --- データ収集
 34    game_info: GameInfoDict = aggregate.game_info()
 35    msg_data: dict = {}
 36    mapping_dict: dict = {}
 37
 38    if game_info["game_count"] == 0:
 39        if g.params.get("individual"):
 40            msg_data["検索範囲"] = f"{ExtDt(g.params["starttime"]).format("ymdhm")}"
 41            msg_data["検索範囲"] += f" ～ {ExtDt(g.params["endtime"]).format("ymdhm")}"
 42            msg_data["特記事項"] = "、".join(message.remarks())
 43            msg_data["検索ワード"] = message.search_word()
 44            msg_data["対戦数"] = f"0 戦 (0 勝 0 敗 0 分) {message.badge_status(0, 0)}"
 45            return (message_build(msg_data), {})
 46        return ("登録されていないチームです", {})
 47
 48    result_df = aggregate.game_results()
 49    record_df = aggregate.ranking_record()
 50
 51    if result_df.empty or record_df.empty:
 52        return (message.reply(message="no_target"), {})
 53
 54    result_df = pd.merge(
 55        result_df, record_df,
 56        on=["name", "name"],
 57        suffixes=["", "_x"]
 58    )
 59
 60    player_name = formatter.name_replace(g.params["player_name"], add_mark=True)
 61    if g.params.get("anonymous"):
 62        mapping_dict = formatter.anonymous_mapping(result_df["name"].unique().tolist())
 63        result_df["name"] = result_df["name"].replace(mapping_dict)
 64        player_name = mapping_dict[player_name]
 65
 66    result_df = formatter.df_rename(result_df)
 67    data = result_df.to_dict(orient="records")[0]
 68
 69    # --- 表示内容
 70    msg_data.update(get_headline(data, game_info, player_name))
 71    msg_data.update(get_totalization(data))
 72
 73    msg2: dict = {}
 74    msg2["座席データ"] = get_seat_data(data)
 75    msg2.update(get_record(data))  # ベスト/ワーストレコード
 76    msg2.update(get_regulations(mapping_dict))  # レギュレーション
 77
 78    if g.params.get("game_results"):  # 戦績
 79        msg2["戦績"] = get_game_results(mapping_dict)
 80
 81    if g.params.get("versus_matrix"):  # 対戦結果
 82        msg2["対戦"] = get_versus_matrix(mapping_dict)
 83
 84    # 非表示項目
 85    if g.cfg.mahjong.ignore_flying:
 86        g.cfg.dropitems.results.append("トビ")
 87    if "トビ" in g.cfg.dropitems.results:
 88        msg2["座席データ"] = re.sub(r"/ .* /", "/", msg2["座席データ"], flags=re.MULTILINE)
 89    if "役満" in g.cfg.dropitems.results:
 90        msg2["座席データ"] = msg2["座席データ"].replace(" / 役満", "")
 91        msg2["座席データ"] = re.sub(r" / [0-9]+$", "", msg2["座席データ"], flags=re.MULTILINE)
 92        msg2.pop("役満和了", None)
 93
 94    if not g.params.get("statistics"):  # 統計
 95        for k in ("座席データ", "ベストレコード", "ワーストレコード"):
 96            msg2.pop(k, None)
 97
 98    for k in list(msg2.keys()):
 99        if k in g.cfg.dropitems.results:
100            msg2.pop(k)
101
102    return (message_build(msg_data), msg2)
103
104
105def get_headline(data: dict, game_info: GameInfoDict, player_name: str) -> dict:
106    """ヘッダメッセージ生成
107
108    Args:
109        data (dict): 生成内容が格納された辞書
110        game_info (GameInfoDict): ゲーム集計情報
111        player_name (str): プレイヤー名
112
113    Returns:
114        dict: 集計データ
115    """
116
117    ret: dict = {}
118
119    if g.params.get("individual"):
120        ret["title"] = "*【個人成績】*"
121        ret["プレイヤー名"] = f"{player_name} {message.badge_degree(data["ゲーム数"])}"
122        if (team_list := lookup.internal.which_team(g.params["player_name"])):
123            ret["所属チーム"] = team_list
124    else:
125        ret["title"] = "*【チーム成績】*"
126        ret["チーム名"] = f"{g.params["player_name"]} {message.badge_degree(data["ゲーム数"])}"
127        ret["登録メンバー"] = "、".join(lookup.internal.get_teammates(g.params["player_name"]))
128
129    badge_status = message.badge_status(data["ゲーム数"], data["win"])
130    ret["検索範囲"] = message.item_search_range(kind="str", time_pattern="time").strip()
131    ret["集計範囲"] = message.item_aggregation_range(game_info, kind="str").strip()
132    ret["特記事項"] = "、".join(message.remarks())
133    ret["検索ワード"] = message.search_word()
134    ret["対戦数"] = f"{data["ゲーム数"]} 戦 ({data["win"]} 勝 {data["lose"]} 敗 {data["draw"]} 分) {badge_status}"
135    ret["_blank1"] = True
136
137    return ret
138
139
140def get_totalization(data: dict) -> dict:
141    """集計トータルメッセージ生成
142
143    Args:
144        data (dict): 生成内容が格納された辞書
145
146    Returns:
147        dict: 生成メッセージ
148    """
149
150    ret: dict = {}
151
152    ret["通算ポイント"] = f"{data["通算ポイント"]:+.1f}pt".replace("-", "▲")
153    ret["平均ポイント"] = f"{data["平均ポイント"]:+.1f}pt".replace("-", "▲")
154    ret["平均順位"] = f"{data["平均順位"]:1.2f}"
155    if g.params.get("individual") and g.cfg.badge.grade.display:
156        ret["段位"] = message.badge_grade(g.params["player_name"])
157    ret["_blank2"] = True
158    ret["1位"] = f"{data["1位"]:2} 回 ({data["1位率"]:6.2f}%)"
159    ret["2位"] = f"{data["2位"]:2} 回 ({data["2位率"]:6.2f}%)"
160    ret["3位"] = f"{data["3位"]:2} 回 ({data["3位率"]:6.2f}%)"
161    ret["4位"] = f"{data["4位"]:2} 回 ({data["4位率"]:6.2f}%)"
162    ret["トビ"] = f"{data["トビ"]:2} 回 ({data["トビ率"]:6.2f}%)"
163    ret["役満"] = f"{data["役満和了"]:2} 回 ({data["役満和了率"]:6.2f}%)"
164
165    return ret
166
167
168def get_seat_data(data: dict) -> str:
169    """座席データメッセージ生成
170
171    Args:
172        data (dict): 生成内容が格納された辞書
173
174    Returns:
175        str: 生成メッセージ
176    """
177
178    ret: str = textwrap.dedent(f"""\
179        *【座席データ】*
180        \t# 席：順位分布(平均順位) / トビ / 役満 #
181        \t{data["東家-順位分布"]:22s} / {data["東家-トビ"]} / {data["東家-役満和了"]}
182        \t{data["南家-順位分布"]:22s} / {data["南家-トビ"]} / {data["南家-役満和了"]}
183        \t{data["西家-順位分布"]:22s} / {data["西家-トビ"]} / {data["西家-役満和了"]}
184        \t{data["北家-順位分布"]:22s} / {data["北家-トビ"]} / {data["北家-役満和了"]}
185    """).replace("0.00", "-.--")
186
187    return ret
188
189
190def get_record(data: dict) -> dict:
191    """レコード情報メッセージ生成
192
193    Args:
194        data (dict): 生成内容が格納された辞書
195
196    Returns:
197        dict: 集計データ
198    """
199
200    def current_data(count: int) -> str:
201        if count == 0:
202            ret = "0 回"
203        elif count == 1:
204            ret = "1 回目"
205        else:
206            ret = f"{count} 連続中"
207        return ret
208
209    def max_data(count: int, current: int) -> str:
210        if count == 0:
211            ret = "*****"
212        elif count == 1:
213            ret = "最大 1 回"
214        else:
215            ret = f"最大 {count} 連続"
216
217        if count == current:
218            if count:
219                ret = "記録更新中"
220            else:
221                ret = "記録なし"
222
223        return ret
224
225    ret: dict = {}
226    ret["ベストレコード"] = textwrap.dedent(f"""\
227        *【ベストレコード】*
228        \t連続トップ：{current_data(data["c_top"])} ({max_data(data["連続トップ"], data["c_top"])})
229        \t連続連対：{current_data(data["c_top2"])} ({max_data(data["連続連対"], data["c_top2"])})
230        \t連続ラス回避：{current_data(data["c_top3"])} ({max_data(data["連続ラス回避"], data["c_top3"])})
231        \t最大素点：{data["最大素点"] * 100}点
232        \t最大獲得ポイント：{data["最大獲得ポイント"]}pt
233    """).replace("-", "▲").replace("*****", "-----")
234
235    ret["ワーストレコード"] = textwrap.dedent(f"""\
236        *【ワーストレコード】*
237        \t連続ラス：{current_data(data["c_low4"])} ({max_data(data["連続ラス"], data["c_low4"])})
238        \t連続逆連対：{current_data(data["c_low2"])} ({max_data(data["連続逆連対"], data["c_low2"])})
239        \t連続トップなし：{current_data(data["c_low"])} ({max_data(data["連続トップなし"], data["c_low"])})
240        \t最小素点：{data["最小素点"] * 100}点
241        \t最小獲得ポイント：{data["最小獲得ポイント"]}pt
242    """).replace("-", "▲").replace("*****", "-----")
243
244    return ret
245
246
247def get_regulations(mapping_dict: dict) -> dict:
248    """レギュレーション情報メッセージ生成
249
250    Returns:
251        dict: 集計データ
252    """
253
254    ret: dict = {}
255
256    df_grandslam = aggregate.remark_count("grandslam")
257    df_regulations = aggregate.remark_count("regulation")
258
259    if g.params.get("anonymous"):
260        new_list = list(set(df_grandslam["name"].unique().tolist() + df_regulations["name"].unique().tolist()))
261        for name in new_list:
262            if name in mapping_dict:
263                new_list.remove(name)
264
265        mapping_dict.update(formatter.anonymous_mapping(new_list, len(mapping_dict)))
266        df_grandslam["name"] = df_grandslam["name"].replace(mapping_dict)
267        df_regulations["name"] = df_regulations["name"].replace(mapping_dict)
268
269    if not df_grandslam.empty:
270        ret["役満和了"] = "\n*【役満和了】*\n"
271        for x in df_grandslam.itertuples():
272            ret["役満和了"] += f"\t{x.matter}\t{x.count}回\n"
273
274    if not df_regulations.query("type == 1").empty:
275        ret["卓外ポイント"] = "\n*【卓外ポイント】*\n"
276        for x in df_regulations.query("type == 1").itertuples():
277            ex_point = str(x.ex_point).replace("-", "▲")
278            ret["卓外ポイント"] += f"\t{x.matter}\t{x.count}回 ({ex_point}pt)\n"
279
280    if not df_regulations.query("type != 1").empty:
281        ret["その他"] = "\n*【その他】*\n"
282        for x in df_regulations.query("type != 1").itertuples():
283            ret["その他"] += f"\t{x.matter}\t{x.count}回\n"
284
285    return ret
286
287
288def get_game_results(mapping_dict: dict) -> str:
289    """戦績データ出力用メッセージ生成
290
291    Returns:
292        str: 出力メッセージ
293    """
294
295    ret: str = "\n*【戦績】*\n"
296    data: dict = {}
297
298    target_player = formatter.name_replace(g.params["target_player"][0], add_mark=True)
299    df = loader.read_data("summary/details.sql").fillna(value="")
300
301    if g.params.get("anonymous"):
302        mapping_dict.update(formatter.anonymous_mapping(df["name"].unique().tolist(), len(mapping_dict)))
303        df["name"] = df["name"].replace(mapping_dict)
304        target_player = mapping_dict.get(target_player, target_player)
305
306    p_list: dict = df["name"].unique().tolist()
307    if g.params.get("verbose"):
308        data["p0"] = df.filter(items=["playtime", "guest_count", "same_team"]).drop_duplicates().set_index("playtime")
309        for idx, prefix in enumerate(["p1", "p2", "p3", "p4"]):  # pylint: disable=unused-variable  # noqa: F841
310            tmp_df = df.query("seat == @idx + 1").filter(
311                items=["playtime", "name", "rpoint", "rank", "point", "grandslam", "name"]
312            )
313
314            data[prefix] = tmp_df.rename(
315                columns={
316                    "name": f"{prefix}_name",
317                    "rpoint": f"{prefix}_rpoint",
318                    "rank": f"{prefix}_rank",
319                    "point": f"{prefix}_point",
320                    "grandslam": f"{prefix}_gs",
321                }
322            ).set_index("playtime")
323
324        max_len = textutil.count_padding(p_list)
325        df_data = pd.concat([data["p1"], data["p2"], data["p3"], data["p4"], data["p0"]], axis=1)
326        df_data = df_data.query("p1_name == @target_player or p2_name == @target_player or p3_name == @target_player or p4_name == @target_player")
327
328        for x in df_data.itertuples():
329            vs_guest = ""
330            if x.guest_count >= 2 and g.params["individual"]:
331                vs_guest = "(2ゲスト戦)"
332            if x.same_team == 1 and not g.params["individual"]:
333                vs_guest = "(チーム同卓)"
334
335            ret += textwrap.dedent(
336                """\
337                {} {}
338                \t東家：{} {} {}位 {:8d}点 ({:7.1f}pt) {}
339                \t南家：{} {} {}位 {:8d}点 ({:7.1f}pt) {}
340                \t西家：{} {} {}位 {:8d}点 ({:7.1f}pt) {}
341                \t北家：{} {} {}位 {:8d}点 ({:7.1f}pt) {}
342                """
343            ).format(
344                x.Index.replace("-", "/"), vs_guest,
345                x.p1_name, " " * (max_len - textutil.len_count(x.p1_name)), x.p1_rank, int(x.p1_rpoint) * 100, x.p1_point, x.p1_gs,
346                x.p2_name, " " * (max_len - textutil.len_count(x.p2_name)), x.p2_rank, int(x.p2_rpoint) * 100, x.p2_point, x.p2_gs,
347                x.p3_name, " " * (max_len - textutil.len_count(x.p3_name)), x.p3_rank, int(x.p3_rpoint) * 100, x.p3_point, x.p3_gs,
348                x.p4_name, " " * (max_len - textutil.len_count(x.p4_name)), x.p4_rank, int(x.p4_rpoint) * 100, x.p4_point, x.p4_gs,
349            ).replace(" -", "▲")
350    else:
351        df_data = df.query("name == @target_player").set_index("playtime")
352        for x in df_data.itertuples():
353            play_time = str(x.Index).replace("-", "/")
354            rpoint = cast(int, x.rpoint) * 100
355            point = cast(float, x.point)
356            vs_guest = ""
357            guest_count = cast(int, x.guest_count)
358            same_team = cast(int, x.same_team)
359
360            if guest_count >= 2 and g.params.get("individual"):
361                vs_guest = g.cfg.setting.guest_mark
362            if same_team == 1 and not g.params.get("individual"):
363                vs_guest = g.cfg.setting.guest_mark
364
365            ret += f"\t{vs_guest}{play_time}  {x.rank}位 {rpoint:8d}点 ({point:7.1f}pt) {x.grandslam}\n".replace("-", "▲")
366
367    return ret
368
369
370def get_versus_matrix(mapping_dict: dict) -> str:
371    """対戦結果データ出力用メッセージ生成
372
373    Returns:
374        str: 出力メッセージ
375    """
376
377    ret: str = "\n*【対戦結果】*\n"
378    df = loader.read_data("summary/versus_matrix.sql")
379
380    if g.params.get("anonymous"):
381        mapping_dict.update(formatter.anonymous_mapping(df["vs_name"].unique().tolist(), len(mapping_dict)))
382        df["my_name"] = df["my_name"].replace(mapping_dict)
383        df["vs_name"] = df["vs_name"].replace(mapping_dict)
384
385    max_len = textutil.count_padding(df["vs_name"].unique().tolist())
386
387    for _, r in df.iterrows():
388        padding = max_len - textutil.len_count(r["vs_name"])
389        ret += f"\t{r["vs_name"]}{" " * padding} ： "
390        ret += f"{r["game"]:3d} 戦 {r["win"]:3d} 勝 {r["lose"]:3d} 敗 ({r["win%"]:6.2f}%)\n"
391
392    return ret
393
394
395def message_build(data: dict) -> str:
396    """表示する内容をテキストに起こす
397
398    Args:
399        data (dict): 内容
400
401    Returns:
402        str: 表示するテキスト
403    """
404
405    msg = ""
406    for k, v in data.items():
407        if not v:  # 値がない項目は削除
408            continue
409        match k:
410            case k if k in g.cfg.dropitems.results:  # 非表示
411                pass
412            case k if str(k).startswith("_blank"):
413                msg += "\t\n"
414            case "title":
415                msg += f"{v}\n"
416            case _:
417                msg += f"\t{k}：{v}\n"
418
419    return msg.strip()