libs.domain.score

  1"""
  2libs/domain/score.py
  3"""
  4
  5import re
  6from dataclasses import dataclass, field
  7from typing import TYPE_CHECKING, Any, Literal, Optional, cast
  8
  9import pandas as pd
 10
 11if TYPE_CHECKING:
 12    from libs.types import ScoreDict
 13
 14
 15@dataclass
 16class Score:
 17    """
 18    プレイヤー成績
 19
 20    Note:
 21        フィールド名の 'r_' プレフィックス (r_str, rpoint など) は、
 22        GameResult.to_dict() によって 'p1_', 'p2_', 'p3_', 'p4_' に置換され、
 23        DBテーブルのカラム名 (p1_str, p2_str など) として使用する。
 24
 25    """
 26
 27    name: str = field(default="")
 28    """プレイヤー名"""
 29    r_str: str = field(default="")
 30    """素点(ユーザー入力文字列/未計算)"""
 31    rpoint: int = field(default=0)
 32    """素点(ユーザー入力文字列評価後)"""
 33    point: float = field(default=0.0)
 34    """獲得ポイント"""
 35    rank: int = field(default=0)
 36    """獲得順位"""
 37
 38    def has_valid_data(self) -> bool:
 39        """有効なデータを持っているかチェック"""
 40        return self != Score()
 41
 42    def to_dict(self, prefix: str) -> "ScoreDict":
 43        """
 44        データを辞書で返す
 45
 46        Args:
 47            prefix (str): キーに付与する接頭辞 (p1, p2, p3, p4)
 48
 49        Returns:
 50            ScoreDict: 返却する辞書
 51
 52        Note:
 53            フィールド名の 'r_' プレフィックスは、指定された prefix に置換される。
 54            例: r_str -> p1_str (prefix='p1' の場合)
 55
 56        """
 57        return cast(
 58            "ScoreDict",
 59            {
 60                f"{prefix}_name": self.name,
 61                f"{prefix}_str": self.r_str,
 62                f"{prefix}_rpoint": self.rpoint,
 63                f"{prefix}_point": self.point,
 64                f"{prefix}_rank": self.rank,
 65            },
 66        )
 67
 68
 69class GameResult:
 70    """スコアデータ"""
 71
 72    def __init__(self, **kwargs: Any) -> None:
 73        # ゲーム結果
 74        self.ts: str = ""
 75        """タイムスタンプ"""
 76        self.p1: Score = Score()
 77        """東家成績"""
 78        self.p2: Score = Score()
 79        """南家成績"""
 80        self.p3: Score = Score()
 81        """西家成績"""
 82        self.p4: Score = Score()
 83        """北家成績"""
 84        self.comment: Optional[str] = None
 85        """ゲームコメント"""
 86        self.deposit: int = 0
 87        """供託"""
 88
 89        # 付属情報
 90        self.mode: Literal[3, 4] = 4
 91        """集計モード(三人打ち/四人打ち)"""
 92        self.rule_version: str = ""
 93        """ルール識別子"""
 94        self.origin_point: int = 250
 95        """配給原点"""
 96        self.return_point: int = 300
 97        """返し点"""
 98        self.rank_point: list[int] = [30, 10, -10, -30]
 99        """順位点"""
100        self.draw_split: bool = False
101        """同着時に順位点を山分けにするか"""
102        self.source: Optional[str] = None
103        """データ入力元識別子"""
104
105        self.calc(**kwargs)
106
107    def __bool__(self) -> bool:
108        return all(self.to_list("name") + self.to_list("str"))
109
110    def __eq__(self, other: Any) -> bool:
111        if not isinstance(other, GameResult):
112            return NotImplemented
113
114        return all(
115            [
116                self.mode == other.mode,
117                self.ts == other.ts,
118                self.p1.name == other.p1.name,
119                self.p1.rpoint == other.p1.rpoint,
120                self.p2.name == other.p2.name,
121                self.p2.rpoint == other.p2.rpoint,
122                self.p3.name == other.p3.name,
123                self.p3.rpoint == other.p3.rpoint,
124                self.p4.name == other.p4.name,
125                self.p4.rpoint == other.p4.rpoint,
126                self.rule_version == other.rule_version,
127                self.comment == other.comment,
128                self.source == other.source,
129            ]
130        )
131
132    def __lt__(self, other: Any) -> bool:
133        if not isinstance(other, GameResult):
134            return NotImplemented
135        return self.ts < other.ts
136
137    def has_valid_data(self) -> bool:
138        """DB更新に必要なデータを持っているかチェック"""
139        # スコアデータ
140        match self.mode:
141            case 3:
142                score_data = all(
143                    [
144                        self.p1.has_valid_data(),
145                        self.p2.has_valid_data(),
146                        self.p3.has_valid_data(),
147                    ]
148                )
149            case 4:
150                score_data = all(
151                    [
152                        self.p1.has_valid_data(),
153                        self.p2.has_valid_data(),
154                        self.p3.has_valid_data(),
155                        self.p4.has_valid_data(),
156                    ]
157                )
158            case _:
159                score_data = False
160
161        return all([self.ts, isinstance(self.ts, str), score_data, all(self.to_list("rank"))])
162
163    def set(self, **kwargs: Any) -> None:
164        """データ取り込み"""
165
166        def _normalize_score_string(s: str) -> str:
167            """
168            素点文字列の正規化
169
170            Args:
171                s (str): 入力文字列
172
173            Returns:
174                str: 正規化後の文字列
175
176            """
177            s = s.strip()
178            s = re.sub(r"(-)+|(\+)+", r"\1\2", s)  # 連続した符号を集約
179            s = re.sub(r"(-|\+)0+", r"\1", s)  # 符号の直後のゼロを削除
180            if s != "0":  # 先頭のゼロとプラス記号を削除
181                s = re.sub(r"^[0+]+", "", s)
182            return s
183
184        def _set_score_attr(score: Score, prefix: str, key: str, value: object) -> None:
185            """
186            Scoreオブジェクトに属性を設定
187
188            Args:
189                score (Score): 対象スコアオブジェクト
190                prefix (str): プレイヤーポジション (p1-p4)
191                key (str): 属性名
192                value (object): 設定値
193
194            """
195            match key:
196                case "name":
197                    score.name = str(value)
198                case "str":
199                    score.r_str = _normalize_score_string(str(value))
200                case "r_str":
201                    score.r_str = str(value)
202                case "rpoint" if isinstance(value, int):
203                    score.rpoint = value
204                case "point" if isinstance(value, (float, int)):
205                    score.point = float(value)
206                case "rank" if isinstance(value, int):
207                    score.rank = value
208
209        # プレイヤースコア設定
210        for prefix in ("p1", "p2", "p3", "p4"):
211            score_obj = cast(Score, getattr(self, prefix))
212            for attr in ("name", "str", "r_str", "rpoint", "point", "rank"):
213                key = f"{prefix}_{attr}"
214                if key in kwargs:
215                    _set_score_attr(score_obj, prefix, attr, kwargs[key])
216
217        # ゲーム設定
218        if "mode" in kwargs and isinstance(kwargs["mode"], int):
219            if kwargs["mode"] in (3, 4):
220                self.mode = kwargs["mode"]  # type: ignore[assignment]
221        if "ts" in kwargs and isinstance(kwargs["ts"], str):
222            self.ts = kwargs["ts"]
223        if "rule_version" in kwargs and isinstance(kwargs["rule_version"], str):
224            self.rule_version = str(kwargs["rule_version"])
225        if "deposit" in kwargs and isinstance(kwargs["deposit"], int):
226            self.deposit = int(kwargs["deposit"])
227        if "origin_point" in kwargs and isinstance(kwargs["origin_point"], int):
228            self.origin_point = int(kwargs["origin_point"])
229        if "return_point" in kwargs and isinstance(kwargs["return_point"], int):
230            self.return_point = int(kwargs["return_point"])
231        if "rank_point" in kwargs and isinstance(kwargs["rank_point"], list):
232            self.rank_point = kwargs["rank_point"]
233        if "draw_split" in kwargs and isinstance(kwargs["draw_split"], bool):
234            self.draw_split = kwargs["draw_split"]
235        if "comment" in kwargs:
236            self.comment = kwargs["comment"]
237        if "source" in kwargs:
238            self.source = kwargs["source"]
239
240    def to_dict(self) -> "ScoreDict":
241        """
242        データを辞書で返す
243
244        Returns:
245            ScoreDict: スコアデータ
246
247        """
248        return {
249            "ts": self.ts,
250            **self.p1.to_dict("p1"),
251            **self.p2.to_dict("p2"),
252            **self.p3.to_dict("p3"),
253            **self.p4.to_dict("p4"),
254            "deposit": self.deposit,
255            "comment": self.comment,
256            "rule_version": self.rule_version,
257            "source": self.source,
258            "mode": self.mode,
259        }
260
261    def to_text(self, kind: Literal["simple", "detail", "logging"] = "simple") -> str:
262        """
263        データをテキストで返す
264
265        Args:
266            kind (Literal, optional): 表示形式
267                - *simple*: 簡易情報 (Default)
268                - *detail*: 詳細情報
269                - *logging*: ロギング用
270
271        Returns:
272            str: スコアデータ
273
274        """
275        ret_text: str = ""
276        match kind:
277            case "simple":
278                ret_text += f"[{self.p1.name} {self.p1.r_str}] "
279                ret_text += f"[{self.p2.name} {self.p2.r_str}] "
280                ret_text += f"[{self.p3.name} {self.p3.r_str}] "
281                ret_text += f"[{self.p4.name} {self.p4.r_str}] " if self.mode == 4 else ""
282                ret_text += f"[供託 {self.deposit}] [{self.comment if self.comment else None}]"
283            case "detail":
284                ret_text += f"[{self.p1.rank}位 {self.p1.name} {self.p1.rpoint * 100}点 ({self.p1.point}pt)] ".replace("-", "▲")
285                ret_text += f"[{self.p2.rank}位 {self.p2.name} {self.p2.rpoint * 100}点 ({self.p2.point}pt)] ".replace("-", "▲")
286                ret_text += f"[{self.p3.rank}位 {self.p3.name} {self.p3.rpoint * 100}点 ({self.p3.point}pt)] ".replace("-", "▲")
287                ret_text += f"[{self.p4.rank}位 {self.p4.name} {self.p4.rpoint * 100}点 ({self.p4.point}pt)] ".replace("-", "▲") if self.mode == 4 else ""
288                ret_text += f"[供託 {self.deposit * 100}点] "
289                ret_text += f"[{self.comment if self.comment else None}]"
290            case "logging":
291                ret_text += f"ts={self.ts}, deposit={self.deposit}, rule_version={self.rule_version}, "
292                ret_text += f"p1={self.p1.to_dict('p1')}, p2={self.p2.to_dict('p2')}, p3={self.p3.to_dict('p3')}, "
293                ret_text += f"p4={self.p4.to_dict('p4')}, " if self.mode == 4 else ""
294                ret_text += f"comment={self.comment if self.comment else None}, source={self.source}"
295
296        return ret_text
297
298    def to_list(self, kind: Literal["name", "str", "rpoint", "point", "rank"] = "name") -> list[str | int | float]:
299        """
300        指定データをリストで返す
301
302        Args:
303            kind (Literal, optional): 取得内容
304                - *name*: プレイヤー名 (Default)
305                - *str*: 入力された素点情報
306                - *rpoint*: 素点
307                - *point*: ポイント
308                - *rank*: 順位
309
310        Returns:
311            list[str | int | float]: リスト
312
313        """
314        ret_list: list[str | int | float] = []
315        match kind:
316            case "name":
317                ret_list = [self.p1.name, self.p2.name, self.p3.name, self.p4.name]
318            case "str":
319                ret_list = [self.p1.r_str, self.p2.r_str, self.p3.r_str, self.p4.r_str]
320            case "rpoint":
321                ret_list = [self.p1.rpoint, self.p2.rpoint, self.p3.rpoint, self.p4.rpoint]
322            case "point":
323                ret_list = [self.p1.point, self.p2.point, self.p3.point, self.p4.point]
324            case "point":
325                ret_list = [self.p1.point, self.p2.point, self.p3.point, self.p4.point]
326            case "rank":
327                ret_list = [self.p1.rank, self.p2.rank, self.p3.rank, self.p4.rank]
328
329        return ret_list[: self.mode]
330
331    def calc(self, **kwargs: Any) -> None:
332        """獲得ポイント計算"""
333        if kwargs:
334            self.set(**kwargs)
335
336        match self.mode:
337            case 3:
338                if all([self.p1.has_valid_data(), self.p2.has_valid_data(), self.p3.has_valid_data()]):
339                    self.set(**self._calculation_point3())
340                    self.p4 = Score()
341            case 4:
342                if all([self.p1.has_valid_data(), self.p2.has_valid_data(), self.p3.has_valid_data(), self.p4.has_valid_data()]):
343                    self.set(**self._calculation_point4())
344            case _:
345                raise RuntimeError
346
347    def _normalized_expression(self, expr: str) -> int:
348        """
349        入力文字列を式として評価し、計算結果を返す
350
351        Args:
352            expr (str): 入力式
353
354        Returns:
355            int: 計算結果
356
357        """
358        normalized: list[str] = []
359
360        for token in re.findall(r"\d+|[+\-*/]", expr):
361            if isinstance(token, str):
362                if token.isnumeric():
363                    normalized.append(str(int(token)))
364                else:
365                    normalized.append(token)
366
367        return int(eval("".join(normalized)))
368
369    def _calculation_point3(self) -> dict[str, Any]:
370        """
371        獲得ポイントと順位を計算する(三人打ち)
372
373        Returns:
374            dict[str, Any]: 更新用辞書(順位と獲得ポイントのデータ)
375
376        """
377        # 計算用データフレーム
378        score_df = pd.DataFrame({"rpoint": [self._normalized_expression(str(x)) for x in self.to_list("str")]}, index=["p1", "p2", "p3"])
379
380        work_rank_point = self.rank_point.copy()  # ウマ
381        work_rank_point[0] += int((self.return_point - self.origin_point) / 10 * 3)  # オカ
382
383        # 席順
384        score_df["rank"] = score_df["rpoint"].rank(ascending=False, method="first").astype("int")
385
386        # 獲得ポイントの計算 (素点-配給原点)/10+順位点
387        score_df["position"] = score_df["rpoint"].rank(ascending=False, method="first").astype("int")  # 加算する順位点リストの位置
388        score_df["point"] = (score_df["rpoint"] - self.return_point) / 10 + score_df["position"].apply(lambda p: work_rank_point[p - 1])
389        score_df["point"] = score_df["point"].apply(lambda p: float(f"{p:.1f}"))  # 桁ブレ修正
390
391        # 返却値用辞書
392        ret_dict = {f"{k}_{x}": v for x in score_df.columns for k, v in score_df[x].to_dict().items()}
393        ret_dict.update(deposit=int(self.origin_point * 3 - score_df["rpoint"].sum()))
394
395        return ret_dict
396
397    def _calculation_point4(self) -> dict[str, Any]:
398        """
399        獲得ポイントと順位を計算する(四人打ち)
400
401        Returns:
402            dict[str, Any]: 更新用辞書(順位と獲得ポイントのデータ)
403
404        """
405
406        def point_split(point: list[int]) -> list[int]:
407            """
408            順位点を山分けする
409
410            Args:
411                point (list[int]): 山分けするポイントのリスト
412
413            Returns:
414                list[int]: 山分けした結果
415
416            """
417            new_point = [int(sum(point) / len(point))] * len(point)
418            if sum(point) % len(point):
419                new_point[0] += sum(point) % len(point)
420                if sum(point) < 0:
421                    new_point = list(map(lambda x: x - 1, new_point))
422
423            return new_point
424
425        # 計算用データフレーム
426        score_df = pd.DataFrame({"rpoint": [self._normalized_expression(str(x)) for x in self.to_list("str")]}, index=["p1", "p2", "p3", "p4"])
427
428        work_rank_point = self.rank_point.copy()  # ウマ
429        work_rank_point[0] += int((self.return_point - self.origin_point) / 10 * 4)  # オカ
430
431        if self.draw_split:  # 山分け
432            score_df["rank"] = score_df["rpoint"].rank(ascending=False, method="min").astype("int")
433
434            # 順位点リストの更新
435            match "".join(score_df["rank"].sort_values().to_string(index=False).split()):
436                case "1111":  # 2.5/2.5/2.5/2.5
437                    work_rank_point = point_split(work_rank_point)
438                case "1114":  # 2/2/2/4
439                    new_point = point_split(work_rank_point[0:3])
440                    work_rank_point[0] = new_point[0]
441                    work_rank_point[1] = new_point[1]
442                    work_rank_point[2] = new_point[2]
443                case "1134":  # 1.5/1.5/3/4
444                    new_point = point_split(work_rank_point[0:2])
445                    work_rank_point[0] = new_point[0]
446                    work_rank_point[1] = new_point[1]
447                case "1133":  # 1.5/1.5/3.5/3.5
448                    new_point = point_split(work_rank_point[0:2])
449                    work_rank_point[0] = new_point[0]
450                    work_rank_point[1] = new_point[1]
451                    new_point = point_split(work_rank_point[2:4])
452                    work_rank_point[2] = new_point[0]
453                    work_rank_point[3] = new_point[1]
454                case "1222":  # 1/3/3/3
455                    new_point = point_split(work_rank_point[1:4])
456                    work_rank_point[1] = new_point[0]
457                    work_rank_point[2] = new_point[1]
458                    work_rank_point[3] = new_point[2]
459                case "1224":  # 1/2.5/2.5/4
460                    new_point = point_split(work_rank_point[1:3])
461                    work_rank_point[1] = new_point[0]
462                    work_rank_point[2] = new_point[1]
463                case "1233":  # 1/2/3.5/3.5
464                    new_point = point_split(work_rank_point[2:4])
465                    work_rank_point[2] = new_point[0]
466                    work_rank_point[3] = new_point[1]
467                case _:
468                    pass
469
470        else:  # 席順
471            score_df["rank"] = score_df["rpoint"].rank(ascending=False, method="first").astype("int")
472
473        # 獲得ポイントの計算 (素点-配給原点)/10+順位点
474        score_df["position"] = score_df["rpoint"].rank(ascending=False, method="first").astype("int")  # 加算する順位点リストの位置
475        score_df["point"] = (score_df["rpoint"] - self.return_point) / 10 + score_df["position"].apply(lambda p: work_rank_point[p - 1])
476        score_df["point"] = score_df["point"].apply(lambda p: float(f"{p:.1f}"))  # 桁ブレ修正
477
478        # 返却値用辞書
479        ret_dict = {f"{k}_{x}": v for x in score_df.columns for k, v in score_df[x].to_dict().items()}
480        ret_dict.update(deposit=int(self.origin_point * 4 - score_df["rpoint"].sum()))
481
482        return ret_dict
483
484    @property
485    def rpoint_sum(self) -> int:
486        """
487        素点合計
488
489        Returns:
490            int: 素点合計
491
492        """
493        if not all(self.to_list("rank")):  # 順位が確定していない場合は先に計算
494            self.calc()
495
496        return sum(cast(list[int], self.to_list("rpoint")))