libs.commands.results.detail

  1"""
  2libs/commands/results/detail.py
  3"""
  4
  5import textwrap
  6from typing import TYPE_CHECKING, Any
  7
  8import numpy as np
  9import pandas as pd
 10from table2ascii import Alignment, PresetStyle, table2ascii
 11
 12import libs.global_value as g
 13from libs.domain.datamodels import GameInfo
 14from libs.domain.stats import StatsInfo
 15from libs.functions import message
 16from libs.functions.compose import badge, text_item
 17from libs.types import StyleOptions
 18from libs.utils import converter, formatter
 19
 20if TYPE_CHECKING:
 21    from integrations.protocols import MessageParserProtocol
 22    from libs.types import MessageType
 23
 24
 25def aggregation(m: "MessageParserProtocol") -> None:
 26    """
 27    成績詳細を集計
 28
 29    Args:
 30        m (MessageParserProtocol): メッセージデータ
 31
 32    """
 33    # --- パラメータ更新
 34    g.params.guest_skip = g.params.guest_skip2  # 検索動作を合わせる
 35
 36    if rule_version := g.params.rule_version:
 37        g.params.update_from_dict(
 38            {
 39                "mode": int(g.cfg.rule.to_dict(rule_version).get("mode", 4)),
 40                "rule_version": str(g.cfg.rule.to_dict(rule_version).get("rule_version", "")),
 41                "origin_point": int(g.cfg.rule.to_dict(rule_version).get("origin_point", 250)),
 42                "return_point": int(g.cfg.rule.to_dict(rule_version).get("return_point", 300)),
 43            }
 44        )
 45        if (target_mode := g.params.target_mode) and target_mode != g.cfg.rule.get_mode(rule_version):
 46            m.set_headline(message.random_reply(m, "rule_mismatch"), StyleOptions(title="集計矛盾検出"))
 47            m.status.result = False
 48            return
 49    if g.params.player_name in g.cfg.team.lists:
 50        g.params.individual = False
 51    elif g.params.player_name in g.cfg.member.lists:
 52        g.params.individual = True
 53
 54    # --- データ収集
 55    game_info = GameInfo()
 56    msg_data: dict[str, str] = {}
 57    mapping_dict: dict[str, str] = {}
 58
 59    # タイトル
 60    if g.params.individual:
 61        title = "個人成績詳細"
 62    else:
 63        title = "チーム成績詳細"
 64
 65    if game_info.count == 0:
 66        if g.params.individual:
 67            msg_data["検索範囲"] = f"{text_item.search_range(time_pattern='time')}"
 68            msg_data["特記事項"] = "、".join(text_item.remarks())
 69            msg_data["検索ワード"] = text_item.search_word()
 70            msg_data["対戦数"] = f"0 戦 (0 勝 0 敗 0 分) {badge.status(0, 0)}"
 71            m.set_headline(message_build(msg_data), StyleOptions(title=title))
 72        else:
 73            m.set_headline("登録されていないチームです。", StyleOptions(title=title))
 74        m.status.result = False
 75        return
 76
 77    stats = StatsInfo()
 78    stats.read(g.params)
 79
 80    if stats.result_df.empty or stats.record_df.empty:
 81        m.set_headline(message.random_reply(m, "no_target"), StyleOptions(title=title))
 82        m.status.result = False
 83        return
 84
 85    player_name = formatter.name_replace(g.params.player_name, add_mark=True)
 86    if g.params.anonymous:
 87        mapping_dict = formatter.anonymous_mapping(stats.result_df["name"].unique().tolist())
 88        stats.result_df["name"] = stats.result_df["name"].replace(mapping_dict)
 89        player_name = mapping_dict[player_name]
 90
 91    # --- 表示内容
 92    msg_data.update(get_headline(stats, game_info, player_name))
 93    msg_data.update(get_totalization(stats))
 94    mode = g.params.mode
 95
 96    # 統計
 97    seat_data = pd.DataFrame(
 98        {  # 座席データ
 99            "席": ["東家", "南家", "西家", "北家"][:mode],
100            "順位分布": stats.rank_distr_list2,
101            "平均順位": [f"{x:.2f}".replace("0.00", "-.--") for x in stats.rank_avg_list],
102            "トビ": stats.flying_list,
103            "役満和了": stats.yakuman_list,
104        }
105    )
106
107    if g.params.ignore_flying or g.cfg.rule.dropitems(g.params.rule_version) & g.cfg.dropitems.flying:
108        seat_data.drop(columns=["トビ"], inplace=True)
109    if g.cfg.rule.dropitems(g.params.rule_version) & g.cfg.dropitems.yakuman:
110        seat_data.drop(columns=["役満和了"], inplace=True)
111
112    if mode == 3:
113        balance_data = textwrap.dedent(
114            f"""\
115            全体：{stats.seat0.avg_balance("all"):+.1f}点
116            1着終了時：{stats.seat0.avg_balance("rank1"):+.1f}点
117            2着終了時：{stats.seat0.avg_balance("rank2"):+.1f}点
118            3着終了時：{stats.seat0.avg_balance("rank3"):+.1f}点
119            """.replace("+0.0点", "記録なし")
120        ).replace("-", "▲")
121    else:
122        balance_data = textwrap.dedent(
123            f"""\
124            全体：{stats.seat0.avg_balance("all"):+.1f}点
125            連対時：{stats.seat0.avg_balance("top2"):+.1f}点
126            逆連対時：{stats.seat0.avg_balance("lose2"):+.1f}点
127            1着終了時：{stats.seat0.avg_balance("rank1"):+.1f}点
128            2着終了時：{stats.seat0.avg_balance("rank2"):+.1f}点
129            3着終了時：{stats.seat0.avg_balance("rank3"):+.1f}点
130            4着終了時：{stats.seat0.avg_balance("rank4"):+.1f}点
131            """.replace("+0.0点", "記録なし")
132        ).replace("-", "▲")
133
134    if g.params.statistics:
135        m.set_message(seat_data, StyleOptions(title="座席データ", data_kind=StyleOptions.DataKind.SEAT_DATA))
136        m.set_message(textwrap.indent(stats.seat0.best_record(), "\t"), StyleOptions(title="ベストレコード"))
137        m.set_message(textwrap.indent(stats.seat0.worst_record(), "\t"), StyleOptions(title="ワーストレコード"))
138        m.set_message(textwrap.indent(balance_data.strip(), "\t"), StyleOptions(title="平均収支"))
139
140    # レギュレーション
141    remarks_df = g.params.read_data("REMARKS_INFO")
142    count_df = remarks_df.groupby("matter").agg(matter_count=("matter", "count"), ex_total=("ex_point", "sum"), type=("type", "max"))
143    count_df["matter"] = count_df.index
144
145    if not g.cfg.rule.dropitems(g.params.rule_version) & g.cfg.dropitems.yakuman:
146        work_df = count_df.query("type == 0").filter(items=["matter", "matter_count"])
147        m.set_message(work_df, StyleOptions(title="役満和了", data_kind=StyleOptions.DataKind.REMARKS_YAKUMAN))
148
149    if not g.cfg.rule.dropitems(g.params.rule_version) & g.cfg.dropitems.regulation:
150        if g.params.individual:
151            work_df = count_df.query("type == 2").filter(items=["matter", "matter_count", "ex_total"])
152        else:
153            work_df = count_df.query("type == 2 or type == 3").filter(items=["matter", "matter_count", "ex_total"])
154        m.set_message(work_df, StyleOptions(title="卓外清算", data_kind=StyleOptions.DataKind.REMARKS_REGULATION))
155
156    if not g.cfg.rule.dropitems(g.params.rule_version) & g.cfg.dropitems.other:
157        work_df = count_df.query("type == 1").filter(items=["matter", "matter_count"])
158        m.set_message(work_df, StyleOptions(title="その他", data_kind=StyleOptions.DataKind.REMARKS_OTHER))
159
160    # 対戦結果
161    if g.params.versus_matrix:
162        m.set_message(get_versus_matrix(mapping_dict), StyleOptions(title="対戦結果", indent=1))
163
164    # 戦績
165    if g.params.game_results:
166        if g.params.verbose:
167            m.set_message(get_results_details(mapping_dict), StyleOptions(title="戦績", data_kind=StyleOptions.DataKind.RECORD_DATA_ALL, codeblock=False))
168        else:
169            m.set_message(get_results_simple(mapping_dict), StyleOptions(title="戦績", data_kind=StyleOptions.DataKind.RECORD_DATA, codeblock=False))
170
171    # 非表示項目を除外
172    m.post.message = [(data, options) for data, options in m.post.message if options.title not in g.cfg.rule.dropitems(g.params.rule_version)]
173
174    m.set_headline(message_build(msg_data), StyleOptions(title=title))
175
176
177def comparison(m: "MessageParserProtocol") -> None:
178    """
179    成績詳細を比較
180
181    Args:
182        m (MessageParserProtocol): メッセージデータ
183
184    """
185    # 検索動作を合わせる
186    g.params.guest_skip = g.params.guest_skip2
187
188    if g.params.player_name in g.cfg.team.lists:
189        g.params.update_from_dict({"individual": False})
190    elif g.params.player_name in g.cfg.member.lists:
191        g.params.update_from_dict({"individual": True})
192
193    # データ収集
194    data: "MessageType"
195    game_info = GameInfo()
196
197    # タイトル
198    title = "成績詳細比較"
199    m.set_headline(message.header(game_info, m, "", 1), StyleOptions(title=title))
200
201    if not game_info.count:
202        m.status.result = False
203        return
204
205    stats_df = pd.DataFrame()
206    result_df = g.params.read_data("RESULTS_INFO")
207    record_df = g.params.read_data("RECORD_INFO")
208
209    for name in result_df.query("id==0").sort_values("total_point", ascending=False)["name"]:
210        work_stats = StatsInfo()
211        if str(name) not in g.params.player_list:
212            continue
213        work_stats.set_parameter(**g.params.placeholder())
214        work_stats.name = str(name)
215        work_stats.set_data(result_df.query("name == @name"))
216        work_stats.set_data(record_df.query("name == @name"))
217        stats_df = pd.concat([stats_df, work_stats.summary])
218
219    if stats_df.empty:
220        m.set_headline(message.random_reply(m, "no_hits"), StyleOptions())
221        m.status.result = False
222        return
223
224    # 規定打数足切り
225    stats_df.query("count >= @g.params.stipulated", inplace=True)
226    if stats_df.empty:
227        m.set_headline(message.random_reply(m, "no_target"), StyleOptions())
228        m.status.result = False
229        return
230
231    if g.params.anonymous:
232        mapping_dict = formatter.anonymous_mapping(stats_df["name"].unique().tolist())
233        stats_df["name"] = stats_df["name"].replace(mapping_dict)
234
235    # 非表示項目
236    stats_df = stats_df.drop(columns=[x for x in g.cfg.rule.dropitems(g.params.rule_version) if x in stats_df.columns.to_list()])
237    if g.params.ignore_flying or g.cfg.rule.dropitems(g.params.rule_version) & g.cfg.dropitems.flying:
238        stats_df = stats_df.drop(columns=["flying_rate-count"])
239    if g.cfg.rule.dropitems(g.params.rule_version) & g.cfg.dropitems.yakuman:
240        stats_df = stats_df.drop(columns=["yakuman_rate-count"])
241
242    # 出力
243    options: StyleOptions = StyleOptions(
244        title=title,
245        data_kind=StyleOptions.DataKind.DETAILED_COMPARISON,
246        base_name=title,
247        show_index=True,
248        codeblock=True,
249        transpose=True,
250    )
251
252    match g.params.format.lower():
253        case "csv":
254            options.format_type = "csv"
255            data = converter.save_output(stats_df, options, m.post.headline)
256        case "text" | "txt":
257            options.format_type = "txt"
258            data = converter.save_output(stats_df, options, m.post.headline)
259        case _:
260            options.key_title = False
261            data = formatter.df_rename(stats_df, options).T
262
263    m.set_message(data, options)
264    m.post.thread = True
265
266
267def get_headline(data: StatsInfo, game_info: GameInfo, player_name: str) -> dict[str, Any]:
268    """
269    ヘッダメッセージ生成
270
271    Args:
272        data (dict): 生成内容が格納された辞書
273        game_info (GameInfo): ゲーム集計情報
274        player_name (str): プレイヤー名
275
276    Returns:
277        dict[str, Any]: 集計データ
278
279    """
280    ret: dict[str, Any] = {}
281
282    if g.params.individual:
283        ret["プレイヤー名"] = f"{player_name} {badge.degree(data.seat0.count)}"
284        if team_name := g.cfg.team.which(g.params.player_name):
285            ret["所属チーム"] = team_name
286    else:
287        ret["チーム名"] = f"{g.params.player_name} {badge.degree(data.seat0.count)}"
288        ret["登録メンバー"] = "、".join(g.cfg.team.member(g.params.player_name))
289
290    badge_status = badge.status(data.seat0.count, data.seat0.win)
291    ret["検索範囲"] = str(text_item.search_range(time_pattern="time"))
292    ret["集計範囲"] = str(text_item.aggregation_range(game_info))
293    ret["特記事項"] = "、".join(text_item.remarks())
294    ret["検索ワード"] = text_item.search_word()
295    ret["対戦数"] = f"{data.seat0.war_record()} {badge_status}"
296    ret["_blank1"] = True
297
298    return ret
299
300
301def get_totalization(data: StatsInfo) -> dict[str, Any]:
302    """
303    集計トータルメッセージ生成
304
305    Args:
306        data (StatsInfo): 成績情報
307
308    Returns:
309        dict[str, Any]: 生成メッセージ
310
311    """
312    ret: dict[str, Any] = {}
313
314    ret["通算ポイント"] = f"{data.seat0.total_point:+.1f}pt".replace("-", "▲")
315    ret["平均ポイント"] = f"{data.seat0.avg_point:+.1f}pt".replace("-", "▲")
316    ret["平均順位"] = f"{data.seat0.rank_avg:1.2f}"
317    if g.params.individual and g.adapter.conf.badge_grade:
318        ret["段位"] = badge.grade(g.params.player_name)
319    ret["_blank2"] = True
320    ret["1位"] = f"{data.seat0.rank1:2} 回 ({data.seat0.rank1_rate:7.2%})"
321    ret["2位"] = f"{data.seat0.rank2:2} 回 ({data.seat0.rank2_rate:7.2%})"
322    ret["3位"] = f"{data.seat0.rank3:2} 回 ({data.seat0.rank3_rate:7.2%})"
323    if g.params.mode == 4:
324        ret["4位"] = f"{data.seat0.rank4:2} 回 ({data.seat0.rank4_rate:7.2%})"
325    ret["トビ"] = f"{data.seat0.flying:2} 回 ({data.seat0.flying_rate:7.2%})"
326    ret["役満"] = f"{data.seat0.yakuman:2} 回 ({data.seat0.yakuman_rate:7.2%})"
327
328    return ret
329
330
331def get_results_simple(mapping_dict: dict[str, str]) -> pd.DataFrame:
332    """
333    戦績(簡易)データ取得
334
335    Args:
336        mapping_dict (dict[str, str]): 匿名化オプション用マップ
337
338    Returns:
339        pd.DataFrame: 戦績データ
340
341    """
342    target_player = formatter.name_replace(g.params.target_player[0], add_mark=True)
343
344    df = g.params.read_data("SUMMARY_DETAILS").fillna(value="")
345    if g.params.anonymous:
346        mapping_dict.update(formatter.anonymous_mapping(df["name"].unique().tolist(), len(mapping_dict)))
347        df["name"] = df["name"].replace(mapping_dict)
348        target_player = mapping_dict.get(target_player, target_player)
349
350    df_data = df.query("name == @target_player")
351    df_data["seat"] = df_data.apply(lambda v: ["東家", "南家", "西家", "北家"][(v["seat"] - 1)], axis=1)
352    df_data["rpoint"] = df_data["rpoint"] * 100
353    pd.options.mode.copy_on_write = True
354    if g.params.individual:
355        df_data.loc[:, "memo"] = np.where(df_data["guest_count"] >= 2, "2ゲスト戦", "")
356    else:
357        df_data.loc[:, "memo"] = np.where(df_data["same_team"] == 1, "チーム同卓", "")
358    df_data = df_data.filter(items=["playtime", "seat", "rank", "rpoint", "point", "remarks", "memo"])
359
360    return df_data
361
362
363def get_results_details(mapping_dict: dict[str, str]) -> pd.DataFrame:
364    """
365    戦績(詳細)データ取得
366
367    Args:
368        mapping_dict (dict[str, str]): 匿名化オプション用マップ
369
370    Returns:
371        pd.DataFrame: 戦績データ
372
373    """
374    target_player = formatter.name_replace(g.params.target_player[0], add_mark=True)  # noqa: F841
375
376    df = g.params.read_data("SUMMARY_DETAILS2").fillna(value="")
377    if g.params.anonymous:
378        name_list: list[str] = []
379        name_list.extend(df["p1_name"].unique().tolist())
380        name_list.extend(df["p2_name"].unique().tolist())
381        name_list.extend(df["p3_name"].unique().tolist())
382        name_list.extend(df["p4_name"].unique().tolist())
383        mapping_dict.update(formatter.anonymous_mapping(list(set(name_list)), len(mapping_dict)))
384        df["p1_name"] = df["p1_name"].replace(mapping_dict)
385        df["p2_name"] = df["p2_name"].replace(mapping_dict)
386        df["p3_name"] = df["p3_name"].replace(mapping_dict)
387        df["p4_name"] = df["p4_name"].replace(mapping_dict)
388        target_player = mapping_dict.get(target_player, target_player)
389
390    match g.params.mode:
391        case 3:
392            df.drop(columns=["p4_name", "p4_rpoint", "p4_rank", "p4_point", "p4_remarks"], inplace=True)
393            df_data = df.query(
394                "p1_name == @target_player or p2_name == @target_player or p3_name == @target_player"  # noqa: E501
395            )
396        case 4:
397            df_data = df.query(
398                "p1_name == @target_player or p2_name == @target_player or p3_name == @target_player or p4_name == @target_player"  # noqa: E501
399            )
400        case _:
401            return pd.DataFrame()
402
403    pd.options.mode.copy_on_write = True
404    if g.params.individual:
405        df_data.loc[:, "memo"] = np.where(df_data["guest_count"] >= 2, "2ゲスト戦", "")
406    else:
407        df_data.loc[:, "memo"] = np.where(df_data["same_team"] == 1, "チーム同卓", "")
408    df_data = df_data.drop(columns=["guest_count", "same_team"])
409
410    return df_data
411
412
413def get_versus_matrix(mapping_dict: dict[str, str]) -> str:
414    """
415    対戦結果データ出力用メッセージ生成
416
417    Args:
418        mapping_dict (dict[str, str]): 匿名化用マッピングデータ
419
420    Returns:
421        str: 出力メッセージ
422
423    """
424    df = g.params.read_data("SUMMARY_VERSUS_MATRIX")
425
426    if df.empty:
427        return ""
428
429    if g.params.anonymous:
430        mapping_dict.update(formatter.anonymous_mapping(df["vs_name"].unique().tolist(), len(mapping_dict)))
431        df["my_name"] = df["my_name"].replace(mapping_dict)
432        df["vs_name"] = df["vs_name"].replace(mapping_dict)
433
434    data_list: list[list[str]] = []
435    for _, r in df.iterrows():
436        data_list.append([r["vs_name"], f"{r['game']} 戦", f"{r['win']} 勝", f"{r['lose']} 敗", f"({r['win%']:6.2f}%)"])
437
438    output = table2ascii(
439        # header=["対戦相手", "ゲーム数", "勝", "負", "勝率"],
440        body=data_list,
441        alignments=[Alignment.LEFT, Alignment.RIGHT, Alignment.RIGHT, Alignment.RIGHT, Alignment.RIGHT],
442        style=PresetStyle.ascii_borderless,
443        cell_padding=0,
444    )
445
446    return output
447
448
449def message_build(data: dict[str, str]) -> str:
450    """
451    表示する内容をテキストに起こす
452
453    Args:
454        data (dict[str, str]): 内容
455
456    Returns:
457        str: 表示するテキスト
458
459    """
460    msg = ""
461    for k, v in data.items():
462        if not v:  # 値がない項目は削除
463            continue
464        match k:
465            case k if k in g.cfg.rule.dropitems(g.params.rule_version):  # 非表示
466                pass
467            case k if str(k).startswith("_blank"):
468                msg += "\n"
469            case "title":
470                msg += f"{v}\n"
471            case _:
472                msg += f"{k}：{v}\n"
473
474    return textwrap.indent(msg.strip(), "\t")