libs.domain.datamodels

  1"""
  2libs/domain/datamodels.py
  3"""
  4
  5import logging
  6from dataclasses import MISSING, dataclass, field, fields
  7from math import ceil
  8from typing import TYPE_CHECKING, Literal, Optional
  9
 10import pandas as pd
 11
 12import libs.global_value as g
 13from libs.utils.timekit import ExtendedDatetime as ExtDt
 14
 15if TYPE_CHECKING:
 16    from pathlib import Path
 17
 18    from integrations.protocols import MessageParserProtocol
 19    from libs.domain.score import GameResult
 20    from libs.types import RemarkDict
 21
 22
 23@dataclass
 24class Args:
 25    """コマンドラインオプション"""
 26
 27    service: str
 28    config: "Path"
 29    """設定ファイルパス"""
 30    no_cleanup: bool
 31    """作業ディレクトリの内容を削除"""
 32
 33    debug: int
 34    """デバッグ出力フラグ"""
 35    verbose: int
 36    """詳細出力フラグ"""
 37
 38    moderate: bool
 39    """INFO以下のログレベル出力を抑止"""
 40    notime: bool
 41    """ログに日付を付与しない"""
 42
 43    # Only allowed when --service=standard_io
 44    text: str
 45
 46    # Only allowed when --service=web
 47    host: str
 48    port: int
 49
 50    # dbtools
 51    compar: bool
 52    unification: "Path"
 53    recalculation: bool
 54    export_data: str
 55    import_data: str
 56    vacuum: bool
 57    gen_test_data: int
 58    testcase: Optional["Path"]
 59
 60
 61@dataclass
 62class GameInfo:
 63    """ゲーム集計情報"""
 64
 65    count: int = field(default=0)
 66    """集計範囲のゲーム数"""
 67    first_game: Optional[ExtDt] = field(default=None)
 68    """集計範囲の最初のゲーム時間"""
 69    last_game: Optional[ExtDt] = field(default=None)
 70    """集計範囲の最後のゲーム時間"""
 71    first_comment: Optional[str] = field(default=None)
 72    """集計範囲の最初のゲームコメント"""
 73    last_comment: Optional[str] = field(default=None)
 74    """集計範囲の最後のゲームコメント"""
 75
 76    def __post_init__(self) -> None:
 77        self.get()
 78
 79    def get(self) -> None:
 80        """指定条件を満たすゲーム数のカウント、最初と最後の時刻とコメントを取得"""
 81        # グローバルパラメータチェック
 82        if not g.params.rule_version:
 83            g.params.rule_version = g.cfg.setting.default_rule
 84        if not g.params.starttime:
 85            g.params.starttime = ExtDt().range("全部").start
 86        if not g.params.endtime:
 87            g.params.endtime = ExtDt().range("全部").end
 88
 89        # データ収集
 90        df = g.params.read_data("GAME_INFO")
 91        if df.empty:
 92            self.count = 0
 93            self.first_game = ExtDt()
 94            self.last_game = ExtDt()
 95            self.first_comment = None
 96            self.last_comment = None
 97        else:
 98            self.count = int(df["count"].iloc[0])
 99            self.first_game = ExtDt(str(df["first_game"].iloc[0]))
100            self.last_game = ExtDt(str(df["last_game"].iloc[0]))
101            first_comment_val = df["first_comment"].iloc[0]
102            self.first_comment = str(first_comment_val) if pd.notna(first_comment_val) else None
103            last_comment_val = df["last_comment"].iloc[0]
104            self.last_comment = str(last_comment_val) if pd.notna(last_comment_val) else None
105
106        # 規定打数更新
107        if not g.params.stipulated:  # 規定打数0はレートから計算
108            match g.params.command:
109                case "results":
110                    g.params.stipulated = g.cfg.results.stipulated_calculation(self.count)
111                case "graph":
112                    g.params.stipulated = g.cfg.graph.stipulated_calculation(self.count)
113                case "ranking":
114                    g.params.stipulated = g.cfg.ranking.stipulated_calculation(self.count)
115                case "report":
116                    g.params.stipulated = g.cfg.report.stipulated_calculation(self.count)
117                case _:
118                    pass
119
120        logging.debug(self)
121
122    def clear(self) -> None:
123        """情報削除"""
124        self.count = 0
125        self.first_game = None
126        self.first_comment = None
127        self.last_game = None
128        self.last_comment = None
129
130
131@dataclass
132class ComparisonResults:
133    """突合結果"""
134
135    search_after: int = field(default=-7)
136    """突合範囲(日数)"""
137    score_list: dict[str, "MessageParserProtocol"] = field(default_factory=dict)
138    """スコアリスト(一時保管用)"""
139
140    mismatch: list[dict[str, "GameResult"]] = field(default_factory=list)
141    """スコア差分"""
142    missing: list["GameResult"] = field(default_factory=list)
143    """スコア追加"""
144    delete: list["GameResult"] = field(default_factory=list)
145    """スコア削除"""
146    remark_mod: list["RemarkDict"] = field(default_factory=list)
147    """メモ変更"""
148    remark_del: list["RemarkDict"] = field(default_factory=list)
149    """メモ削除"""
150    invalid_score: list["GameResult"] = field(default_factory=list)
151    """素点合計不一致"""
152    pending: list["GameResult"] = field(default_factory=list)
153    """処理保留データ"""
154
155    @property
156    def after(self) -> ExtDt:
157        """突合開始日時"""
158        return ExtDt(days=self.search_after, hours=g.cfg.setting.time_adjust)
159
160    @property
161    def before(self) -> ExtDt:
162        """突合終了日時"""
163        return ExtDt()
164
165    def output(
166        self,
167        kind: Literal[
168            "summary",
169            "headline",
170            "pending",
171            "mismatch",
172            "missing",
173            "delete",
174            "remark_mod",
175            "remark_del",
176            "invalid_score",
177        ],
178    ) -> str:
179        """
180        出力メッセージ生成
181
182        Args:
183            kind (Literal[summary, headline, pending, mismatch, missing, delete, remark_mod, remark_del, invalid_score]): 種類
184
185        Returns:
186            str: 生成文字列
187
188        """  # noqa: E501
189        ret: str = ""
190        match kind:
191            case "summary":
192                ret += f"pending:{len(self.pending)} "
193                ret += f"mismatch:{len(self.mismatch)} "
194                ret += f"missing:{len(self.missing)} "
195                ret += f"delete:{len(self.delete)} "
196                ret += f"remark_mod:{len(self.remark_mod)} "
197                ret += f"remark_del:{len(self.remark_del)} "
198                ret += f"invalid_score:{len(self.invalid_score)} "
199            case "headline":
200                ret = f"突合範囲：{self.after.format(ExtDt.FMT.YMDHMS)} - {self.before.format(ExtDt.FMT.YMDHMS)}"
201            case "pending":
202                ret += f"＊ 保留：{len(self.pending)}件\n"
203                for score in self.pending:
204                    ret += f"{ExtDt(float(score.ts)).format(ExtDt.FMT.YMDHMS)} {score.to_text()}\n"
205            case "mismatch":
206                ret += f"＊ 不一致：{len(self.mismatch)}件\n"
207                for score_dict in self.mismatch:
208                    ret += f"{ExtDt(float(score_dict['before'].ts)).format(ExtDt.FMT.YMDHMS)}\n"
209                    ret += f"\t修正前：{score_dict['before'].to_text()}\n"
210                    ret += f"\t修正後：{score_dict['after'].to_text()}\n"
211            case "missing":
212                ret += f"＊ 取りこぼし：{len(self.missing)}件\n"
213                for score in self.missing:
214                    ret += f"{ExtDt(float(score.ts)).format(ExtDt.FMT.YMDHMS)} {score.to_text()}\n"
215            case "delete":
216                ret += f"＊ 削除漏れ：{len(self.delete)}件\n"
217                for score in self.delete:
218                    ret += f"{ExtDt(float(score.ts)).format(ExtDt.FMT.YMDHMS)} {score.to_text()}\n"
219            case "remark_mod":
220                ret += f"＊ メモ更新：{len(self.remark_mod)}件\n"
221                for remark in self.remark_mod:
222                    ret += f"{ExtDt(float(remark['thread_ts'])).format(ExtDt.FMT.YMDHMS)} "
223                    ret += f"{remark['name']} {remark['matter']}\n"
224            case "remark_del":
225                ret += f"＊ メモ削除：{len(self.remark_del)}件\n"
226            case "invalid_score":
227                ret += f"＊ 素点合計不一致：{len(self.invalid_score)}件\n"
228                for score in self.invalid_score:
229                    ret += f"{ExtDt(float(score.ts)).format(ExtDt.FMT.YMDHMS)} {score.to_text()}\n"
230
231        return ret
232
233
234@dataclass
235class ParameterData:
236    """動作パラメータ"""
237
238    # 検索条件変更フラグ
239    individual: bool = field(default=True)
240    """個人/チーム集計切替フラグ
241    - *True*: 個人集計
242    - *False*: チーム集計
243    """
244    guest_skip: bool = field(default=True)
245    """ゲストアリ/ナシフラグ(サマリ集計用)"""
246    guest_skip2: bool = field(default=True)
247    """ゲストアリ/ナシフラグ(詳細集計用)"""
248    unregistered_replace: bool = field(default=True)
249    """メンバー未登録プレイヤー名をゲストに置き換えるかフラグ
250    - *True*: 置き換える
251    - *False*: 置き換えない
252    """
253    friendly_fire: bool = field(default=False)
254    """チーム戦集計時のチーム同卓ゲームの扱い
255    - *True*: チーム同卓ゲームを集計(同じチームのポイントは合算される)
256    - *False*: チーム同卓ゲームを集計対象外にする
257    """
258
259    # 動作変更フラグ
260    statistics: bool = field(default=False)
261    """統計情報表示"""
262
263    # 表示情報変更フラグ
264    ranked: int = field(default=3)
265    """ランキング/レーティングで表示する順位"""
266    stipulated: int = field(default=0)
267    """規定打数指定"""
268    stipulated_rate: float = field(default=0.05)
269    """規定打数計算レート"""
270
271    # 集約条件変更フラグ
272    interval: int = field(default=80)
273    """移動平均算出ゲーム数指定"""
274
275    # コメント検索
276    search_word: str = field(default="")
277    """コメント検索文字列"""
278    group_length: int = field(default=0)
279    """コメント検索時に指定文字数でグループ化する"""
280
281    def default_reset(self) -> None:
282        """デフォルト値にリセット"""
283        for f in fields(self):
284            if f.default is not MISSING:
285                setattr(self, f.name, f.default)
286            elif f.default_factory is not MISSING:
287                setattr(self, f.name, f.default_factory())
288
289
290@dataclass
291class CommandAttrs(ParameterData):
292    """サブコマンド設定パラメータ"""
293
294    commandword: list[str] = field(default_factory=list)
295    """呼び出しキーワード"""
296    command_suffix: list[str] = field(default_factory=list)
297    """コマンド接尾辞(登録キーワード+接尾辞を呼び出しキーワードとして扱う)"""
298
299    aggregation_range: str = field(default="当日")
300    """検索範囲未指定時に使用される範囲"""
301    always_argument: list[str] = field(default_factory=list)
302    """オプションとして常に付与される文字列"""
303    dropitems: list[str] = field(default_factory=list)
304    """非表示にする項目"""
305
306    def stipulated_calculation(self, game_count: int) -> int:
307        """
308        規定打数をゲーム数から計算
309
310        Args:
311            game_count (int): 指定ゲーム数
312
313        Returns:
314            int: 規定ゲーム数
315
316        """
317        return int(ceil(game_count * self.stipulated_rate) + 1)