libs.commands.ranking.ranking

  1"""
  2libs/commands/results/ranking.py
  3"""
  4
  5import re
  6
  7import pandas as pd
  8from tabulate import tabulate
  9
 10import libs.global_value as g
 11from cls.types import GameInfoDict
 12from libs.data import aggregate, loader
 13from libs.functions import message
 14from libs.utils import formatter
 15
 16
 17def aggregation() -> tuple[str, dict]:
 18    """ランキングデータを生成
 19
 20    Returns:
 21        tuple[str, dict]: 集計結果
 22        - str: ランキングの集計情報
 23        - dict: 各ランキングの情報
 24    """
 25
 26    # 情報ヘッダ
 27    if g.params.get("individual"):  # 個人集計
 28        msg = "\n*【ランキング】*\n"
 29    else:  # チーム集計
 30        msg = "\n*【チームランキング】*\n"
 31
 32    # データ取得
 33    game_info: GameInfoDict = aggregate.game_info()
 34    if not game_info["game_count"]:  # 検索結果が0件のとき
 35        msg += "\t" + message.reply(message="no_hits")
 36        return (msg, {})
 37
 38    result_df = loader.read_data("ranking/aggregate.sql")
 39    if result_df.empty:
 40        msg += "\t" + message.reply(message="no_target")
 41        return (msg, {})
 42
 43    df = pd.merge(
 44        result_df, aggregate.ranking_record(),
 45        on=["name", "name"],
 46        suffixes=["", "_x"]
 47    )
 48
 49    if g.params.get("anonymous"):
 50        mapping_dict = formatter.anonymous_mapping(df["name"].unique().tolist())
 51        df["name"] = df["name"].replace(mapping_dict)
 52
 53    # 集計
 54    data: dict = {}
 55
 56    # ゲーム参加率
 57    df["participation_rate"] = df["game_count"] / game_info["game_count"]
 58    df["rank"] = df["participation_rate"].rank(ascending=False, method="dense").astype("int")
 59    df["disp"] = df.apply(lambda row: f"<>{row["participation_rate"]:>7.2%} ({row["game_count"]:3d}G / {game_info["game_count"]:4d}G)", axis=1)
 60    data["ゲーム参加率"] = table_conversion(df)
 61
 62    # 通算ポイント
 63    df["rank"] = df["point_sum"].rank(ascending=False, method="dense").astype("int")
 64    df["disp"] = df.apply(lambda row: f"<>{row["point_sum"]:>+7.1f}pt ({row["game_count"]:3d}G)", axis=1)
 65    data["通算ポイント"] = table_conversion(df)
 66
 67    # 平均ポイント
 68    df["rank"] = df["point_avg"].rank(ascending=False, method="dense").astype("int")
 69    df["disp"] = df.apply(lambda row: f"<>{row["point_avg"]:>+7.1f}pt ({row["point_sum"]:>+7.1f}pt / {row["game_count"]:3d}G)", axis=1)
 70    data["平均ポイント"] = table_conversion(df)
 71
 72    # 平均収支
 73    df["rank"] = df["rpoint_avg"].rank(ascending=False, method="dense").astype("int")
 74    df["disp"] = df.apply(lambda row: f"<>{row["rpoint_avg"] - 25000:>6.0f}点 ({row["rpoint_avg"]:>6.0f}点 / {row["game_count"]:3d}G)", axis=1)
 75    data["平均収支"] = table_conversion(df)
 76
 77    # トップ率
 78    df["rank"] = df["rank1_rate"].rank(ascending=False, method="dense").astype("int")
 79    df["disp"] = df.apply(lambda row: f"<>{row["rank1_rate"]:>7.2%} ({row["rank1"]:3d} / {row["game_count"]:3d}G)", axis=1)
 80    data["トップ率"] = table_conversion(df)
 81
 82    # 連対率
 83    df["rank"] = df["top2_rate"].rank(ascending=False, method="dense").astype("int")
 84    df["disp"] = df.apply(lambda row: f"<>{row["top2_rate"]:>7.2%} ({row["top2"]:3d} / {row["game_count"]:3d}G)", axis=1)
 85    data["連対率"] = table_conversion(df)
 86
 87    # ラス回避率
 88    df["rank"] = df["top3_rate"].rank(ascending=False, method="dense").astype("int")
 89    df["disp"] = df.apply(lambda row: f"<>{row["top3_rate"]:>7.2%} ({row["top3"]:3d} / {row["game_count"]:3d}G)", axis=1)
 90    data["ラス回避率"] = table_conversion(df)
 91
 92    # トビ率
 93    df["rank"] = df["flying_rate"].rank(ascending=True, method="dense").astype("int")
 94    df["disp"] = df.apply(lambda row: f"<>{row["flying_rate"]:>7.2%} ({row["flying"]:3d} / {row["game_count"]:3d}G)", axis=1)
 95    data["トビ率"] = table_conversion(df)
 96
 97    # 平均順位
 98    df["rank"] = df["rank_avg"].rank(ascending=True, method="dense").astype("int")
 99    df["disp"] = df.apply(lambda row: f"<>{row["rank_avg"]:>4.2f} ({row["rank_dist"]})", axis=1)
100    data["平均順位"] = table_conversion(df)
101
102    # 役満和了率
103    df["rank"] = df["gs_rate"].rank(ascending=False, method="dense").astype("int")
104    df["disp"] = df.apply(lambda row: f"<>{row["gs_rate"]:>7.2%} ({row["gs_count"]:3d} / {row["game_count"]:3d}G)", axis=1)
105    data["役満和了率"] = table_conversion(df, ["gs_count", 1])
106
107    # 最大素点
108    df["rank"] = df["rpoint_max"].rank(ascending=False, method="dense").astype("int")
109    df["disp"] = df.apply(lambda row: f"<>{row["rpoint_max"]:>6.0f}点 ({row["point_max"]:>+7.1f}pt)", axis=1)
110    data["最大素点"] = table_conversion(df)
111
112    # 連続トップ
113    df["rank"] = df["c_top"].rank(ascending=False, method="dense").astype("int")
114    df["disp"] = df.apply(lambda row: f"<>{row["c_top"]:>2d}連続 ({row["game_count"]:3d}G)", axis=1)
115    data["連続トップ"] = table_conversion(df, ["c_top", 2])
116
117    # 連続連対
118    df["rank"] = df["c_top2"].rank(ascending=False, method="dense").astype("int")
119    df["disp"] = df.apply(lambda row: f"<>{row["c_top2"]:>2d}連続 ({row["game_count"]:3d}G)", axis=1)
120    data["連続連対"] = table_conversion(df, ["c_top2", 2])
121
122    # 連続ラス回避
123    df["rank"] = df["c_top3"].rank(ascending=False, method="dense").astype("int")
124    df["disp"] = df.apply(lambda row: f"<>{row["c_top3"]:>2d}連続 ({row["game_count"]:3d}G)", axis=1)
125    data["連続ラス回避"] = table_conversion(df, ["c_top3", 2])
126
127    # 表示
128    msg += message.header(game_info, "", 1)
129
130    for key in list(data.keys()):
131        if key in g.cfg.dropitems.ranking:  # 非表示項目
132            data.pop(key)
133            continue
134
135        if key in data:  # 対象者がいなければ項目を削除
136            if not data[key]:
137                data.pop(key)
138                continue
139
140        data[key] = f"*{key}*\n" + data[key]
141
142    return (msg, data)
143
144
145def table_conversion(df: pd.DataFrame, threshold: list | None = None) -> str:
146    """テーブル変換
147
148    Args:
149        df (pd.DataFrame): 変換対象データ
150        threshold (list | None, optional): 非表示にする閾値. Defaults to None.
151
152    Returns:
153        str: 作成したテーブル
154    """
155
156    if isinstance(threshold, list):
157        df = df.query(f"{threshold[0]} >= @threshold[1]").copy()
158
159    if df.empty:
160        return ""
161
162    df.sort_values(by=["rank", "game_count"], ascending=[True, False], inplace=True)
163    df = df.query("rank <= @g.cfg.ranking.ranked")
164    tbl = tabulate(df.filter(items=["rank", "name", "disp"]).values)
165    tbl = re.sub(r"( *[0-9]+)\s(.*)<>(.*)", r"\1:\2\3", tbl)
166    tbl = "\n".join(tbl.splitlines()[1:-1]).replace(" -", "▲")
167    tbl = f"\n```\n{tbl}\n```\n"
168
169    return tbl