libs.domain.rule

  1"""
  2libs/domain/rule.py
  3"""
  4
  5import logging
  6import sys
  7from configparser import ConfigParser
  8from dataclasses import dataclass, field
  9from typing import TYPE_CHECKING, Any, Literal, Mapping
 10
 11from table2ascii import Alignment, PresetStyle, table2ascii
 12
 13from libs.domain.command import CommandParser
 14from libs.utils import dbutil
 15from libs.utils.timekit import ExtendedDatetime as ExtDt
 16
 17if TYPE_CHECKING:
 18    from pathlib import Path
 19
 20
 21@dataclass
 22class RuleData:
 23    """ルールデータ"""
 24
 25    # ルール
 26    rule_version: str = ""
 27    """ルール識別子"""
 28    mode: Literal[3, 4] = 4
 29    """ 集計モード切替(四人打ち/三人打ち)"""
 30    origin_point: int = 250
 31    """配給原点"""
 32    return_point: int = 300
 33    """返し点"""
 34    rank_point: list[int] = field(default_factory=list)
 35    """順位点"""
 36    ignore_flying: bool = False
 37    """トビカウント
 38    - *True*: なし
 39    - *False*: あり
 40    """
 41    draw_split: bool = False
 42    """同点時の順位点
 43    - *True*: 山分けにする
 44    - *False*: 席順で決める
 45    """
 46    undefined_word: int = 1
 47    """未定義ワードタイプ"""
 48    keywords: list[str] = field(default_factory=list)
 49    """成績記録キーワード"""
 50    remarks: list[str] = field(default_factory=list)
 51    """メモ記録用ワード"""
 52    dropitems: list[str] = field(default_factory=list)
 53    """非表示にする項目"""
 54
 55    # ステータス
 56    first_time: ExtDt = field(default=ExtDt("1900-01-01 00:00:00"))
 57    """記録開始日時"""
 58    last_time: ExtDt = field(default=ExtDt("1900-01-01 00:00:00"))
 59    """最終記録日時"""
 60    count: int = 0
 61    """記録回数"""
 62
 63    def update(self, rule_data: Mapping[str, Any]) -> None:
 64        """
 65        ルール更新
 66
 67        Args:
 68            rule_data (Mapping): 更新データ
 69
 70        """
 71        if "rule_version" in rule_data:
 72            self.rule_version = str(rule_data["rule_version"])
 73        if "origin_point" in rule_data:
 74            self.origin_point = int(rule_data["origin_point"])
 75        if "return_point" in rule_data:
 76            self.return_point = int(rule_data["return_point"])
 77
 78        if rank_point := rule_data.get("rank_point"):
 79            if isinstance(rank_point, str):
 80                rank_point = rank_point.split(",")
 81                self.rank_point = list(map(int, map(float, rank_point[: self.mode])))
 82            elif isinstance(rank_point, list):
 83                self.rank_point = list(map(int, map(float, rank_point[: self.mode])))
 84
 85        if undefined_word := rule_data.get("undefined_word"):
 86            self.undefined_word = int(undefined_word)
 87        else:
 88            self.undefined_word = 1
 89
 90        for item in ["ignore_flying", "draw_split"]:
 91            if flag := rule_data.get(item):
 92                if isinstance(flag, bool):
 93                    setattr(self, item, flag)
 94                else:
 95                    setattr(self, item, str(flag).lower() in {"1", "true", "yes", "on"})
 96
 97        for item in ["keywords", "remarks", "dropitems"]:
 98            if item_list := rule_data.get(item):
 99                if isinstance(item_list, str):
100                    setattr(self, item, [x.strip() for x in item_list.split(",")])
101                elif isinstance(item_list, list):
102                    setattr(self, item, item_list)
103
104
105class RuleSet:
106    """ルールセット"""
107
108    def __init__(self) -> None:
109        self.config: ConfigParser = ConfigParser()
110        """ルール設定ファイル"""
111        self.data: dict[str, RuleData] = {}
112        """ルール情報格納辞書"""
113        self.keyword_mapping: dict[str, str] = {}
114        """登録キーワードとルール識別子のマッピング"""
115        self.remarks_words: list[str] = []
116        """メモ記録用ワードリスト"""
117
118    def data_set(self, section_name: str, rule_data: Mapping[str, Any]) -> None:
119        """
120        ルール登録
121
122        Args:
123            section_name (str): セクション名
124            rule_data (Mapping): 更新データ情報
125
126        """
127        rule = RuleData()
128
129        # 初期値セット
130        match int(rule_data.get("mode", 4)):
131            case 3:
132                rule.mode = 3
133                rule.origin_point = 350
134                rule.return_point = 400
135                rule.rank_point = [30, 0, -30]
136            case 4:
137                rule.mode = 4
138                rule.origin_point = 250
139                rule.return_point = 300
140                rule.rank_point = [30, 10, -10, -30]
141            case _:
142                logging.warning("Do not register: %s (invalid mode: %s)", section_name, rule_data.get("mode"))
143                return
144
145        # 設定値取り込み
146        rule.update(rule_data)
147        if not rule.rule_version:
148            rule.rule_version = section_name
149        self.data.update({rule.rule_version: rule})
150
151    def read_config(self, config: "Path") -> None:
152        """
153        設定ファイル読み込み
154
155        Args:
156            config (Path): ルール設定ファイル
157
158        """
159        self.config.read(config, encoding="utf-8")
160        for section_name in map(str, self.config.sections()):
161            if section_name.startswith("regulations_") or section_name.endswith("_regulations"):
162                continue
163            if section_name.startswith("regulations_team_") or section_name.endswith("_regulations_team"):
164                continue
165
166            self.data_set(section_name, dict(self.config[section_name]))
167
168    def status_update(self, params: dict[str, Any]) -> None:
169        """
170        ステータス更新
171
172        Args:
173            params (dict[str, Any]): プレースホルダ
174
175        """
176        # ステータスリセット
177        for rule_version in self.rule_list:
178            self.data[rule_version].count = 0
179            self.data[rule_version].first_time.set("1900-01-01 00:00:00")
180            self.data[rule_version].last_time.set("1900-01-01 00:00:00")
181
182        status = dbutil.execute(
183            """
184            select
185                rule_version,
186                min(ts) as first_time,
187                max(ts) as last_time,
188                count() as count
189            from
190                result
191            --[separate] where source = :source
192            group by
193                rule_version
194            ;
195            """,
196            params,
197        )
198
199        # ステータス更新
200        for status_data in status:
201            if (rule_version := str(status_data.get("rule_version", ""))) and self.data.get(rule_version):
202                if "count" in status_data:
203                    self.data[rule_version].count = int(status_data["count"])
204                if "first_time" in status_data:
205                    self.data[rule_version].first_time = ExtDt(float(status_data["first_time"]))
206                if "last_time" in status_data:
207                    self.data[rule_version].last_time = ExtDt(float(status_data["last_time"]))
208
209    def remarks_words_update(self, suffix: list[str]) -> None:
210        """
211        メモ記録ワードリストを更新する
212
213        Args:
214            suffix (list[str]): メモに追加するサフィックス
215
216        """
217        ret: list[str] = []
218
219        for rule in self.rule_list:
220            if rule_data := self.data.get(rule):
221                if rule_data.remarks:
222                    ret.extend(rule_data.remarks)
223                else:
224                    ret.extend(
225                        [f"{pre}{suf}" for pre in self.keywords(rule) for suf in suffix],
226                    )
227
228        self.remarks_words = list(set(ret))
229
230    def to_dict(self, rule_version: str) -> dict[str, Any]:
231        """
232        指定ルール識別子の情報を辞書で返す
233
234        Args:
235            rule_version (str): ルール識別子
236
237        Returns:
238            dict[str, Any]: ルール情報
239
240        """
241        if rule := self.data.get(rule_version):
242            return {
243                "rule_version": rule.rule_version,
244                "mode": rule.mode,
245                "origin_point": rule.origin_point,
246                "return_point": rule.return_point,
247                "rank_point": rule.rank_point,
248                "ignore_flying": rule.ignore_flying,
249                "draw_split": rule.draw_split,
250                "undefined_word": rule.undefined_word,
251            }
252
253        return {}
254
255    def keywords(self, rule_version: str) -> set[str]:
256        """
257        成績記録キーワードの取得
258
259        Args:
260            rule_version (str): ルール識別子
261
262        Returns:
263            set[str]: 成績記録キーワード
264        """
265        if items := self.data.get(rule_version):
266            return set(items.keywords)
267        else:
268            return set([])
269
270    def dropitems(self, rule_version: str) -> set[str]:
271        """
272        非表示項目の取得
273
274        Args:
275            rule_version (str): ルール識別子
276
277        Returns:
278            set[str]: 非表示項目
279        """
280        if items := self.data.get(rule_version):
281            return set(items.dropitems)
282        else:
283            return set([])
284
285    def get_version(self, mode: int, mapping: bool = True) -> list[str]:
286        """
287        指定した条件のルール識別子をリストで返す
288
289        Args:
290            mode (int): 集計モード
291            mapping (bool, optional): Defaults to True.
292                - *True*: キーワードマッピングに登録されているルールのみ
293                - *False*: ルールとして定義されているものすべて
294
295        Returns:
296            list[str]: ルール識別子
297
298        """
299        ret: list[str] = []
300
301        for keyword, rule in self.data.items():
302            if rule.mode == mode:
303                if mapping:
304                    if keyword in self.keyword_mapping.values():
305                        ret.append(rule.rule_version)
306                else:
307                    ret.append(rule.rule_version)
308
309        return ret
310
311    def get_mode(self, rule_version: str) -> int:
312        """
313        指定ルール識別子の集計モードを返す
314
315        Args:
316            rule_version (str): ルール識別子
317
318        Returns:
319            int: 集計モード
320
321        """
322        return int(self.to_dict(rule_version).get("mode", 0))
323
324    def get_ignore_flying(self, rule_version: str) -> bool:
325        """
326        指定ルール識別子のトビカウントフラグを返す
327
328        Args:
329            rule_version (str): ルール識別子
330
331        Returns:
332            bool: トビカウントフラグ
333
334        """
335        return bool(self.to_dict(rule_version).get("ignore_flying", False))
336
337    def get_undefined_word(self, rule_version: str) -> int:
338        """
339        指定ルール識別子の未定義ワードタイプを返す
340
341        Args:
342            rule_version (str): ルール識別子
343
344        Returns:
345            int: 未定義ワードタイプ
346
347        """
348        return int(self.to_dict(rule_version).get("undefined_word", 1))
349
350    def print(self, rule_version: str) -> str:
351        """
352        指定ルール識別子の内容を出力する
353
354        Args:
355            rule_version (str): ルール識別子
356
357        Returns:
358            str: 内容
359
360        """
361        ret: str = ""
362        body_data: list[list[str]] = []
363
364        if rule := self.data.get(rule_version):
365            body_data.append(["ルール識別子", rule.rule_version])
366
367            # 集計モード
368            match rule.mode:
369                case 3:
370                    body_data.append(["集計モード", "三人打ち"])
371                case 4:
372                    body_data.append(["集計モード", "四人打ち"])
373                case _:
374                    body_data.append(["集計モード", "未定義"])
375
376            body_data.extend(
377                [
378                    ["素点", f"{rule.origin_point * 100}点持ち / {rule.return_point * 100}点返し"],
379                    ["順位点", " / ".join([f"{pt}pt".replace("-", "▲") for pt in rule.rank_point])],
380                    ["同点時", "順位点山分け" if rule.draw_split else "席順"],
381                ]
382            )
383
384            # マッピング情報
385            if keyword := [word for word, mapping_rule in self.keyword_mapping.items() if mapping_rule == rule_version]:
386                body_data.append(["成績登録ワード", "、".join(keyword)])
387            else:
388                body_data.append(["成績登録ワード", "---"])
389
390            # 記録時間
391            body_data.append(["記録数", f"{rule.count} ゲーム"])
392            if rule.count:
393                body_data.extend(
394                    [
395                        ["記録開始日時", rule.first_time.format(ExtDt.FMT.YMDHMS)],
396                        ["最終記録日時", rule.last_time.format(ExtDt.FMT.YMDHMS)],
397                    ]
398                )
399
400            ret = table2ascii(
401                body=body_data,
402                alignments=[Alignment.LEFT, Alignment.LEFT],
403                style=PresetStyle.plain,
404            )
405
406        return ret
407
408    def info(self) -> None:
409        """定義ルールをログに出力する"""
410        for rule in self.data.values():
411            logging.info(
412                "%s: mode=%s, origin_point=%s, return_point=%s, rank_point=%s, draw_split=%s, ignore_flying=%s, undefined_word=%s",
413                rule.rule_version,
414                rule.mode,
415                rule.origin_point,
416                rule.return_point,
417                rule.rank_point,
418                rule.draw_split,
419                rule.ignore_flying,
420                rule.undefined_word,
421            )
422        if self.keyword_mapping:
423            logging.info("keyword_mapping: %s", self.keyword_mapping)
424        else:
425            logging.warning("keyword_mapping: empty")
426        if self.remarks_words:
427            logging.info("remarks_words: %s", self.remarks_words)
428        else:
429            logging.warning("remarks_words: empty")
430
431    def check(self, chk_commands: set[str], chk_members: set[str], default_rule: str) -> None:
432        """
433        キーワード重複チェック
434
435        Args:
436            chk_commands (set[str]): チェック対象コマンド名
437            chk_members (set[str]): チェック対象メンバー名/チーム名
438            default_rule (str): デフォルトルールバージョン
439
440        Raises:
441            RuntimeError: 重複あり
442
443        """
444        chk_word: str | RuleData
445
446        try:
447            # ルール識別子チェック
448            for chk_word in self.data.values():
449                if CommandParser().is_valid_command(chk_word.rule_version):
450                    raise RuntimeError(f"ルール識別子にオプションに使用される単語が使用されています。({chk_word.rule_version})")
451                if chk_word.rule_version in ExtDt.valid_keywords():
452                    raise RuntimeError(f"ルール識別子に検索範囲指定に使用される単語が使用されています。({chk_word.rule_version})")
453                if chk_word.rule_version in chk_commands:
454                    raise RuntimeError(f"ルール識別子と定義済みコマンドに重複があります。({chk_word.rule_version})")
455                if chk_word.rule_version in chk_members:
456                    raise RuntimeError(f"ルール識別子と登録メンバー(チーム)に重複があります。({chk_word.rule_version})")
457            # 成績登録ワードチェック
458            for chk_word in self.keyword_mapping.keys():
459                if CommandParser().is_valid_command(chk_word):
460                    raise RuntimeError(f"成績登録ワードにオプションに使用される単語が使用されています。({chk_word})")
461                if chk_word in ExtDt.valid_keywords():
462                    raise RuntimeError(f"成績登録ワードに検索範囲指定に使用される単語が使用されています。({chk_word})")
463                if chk_word in chk_commands:
464                    raise RuntimeError(f"成績登録ワードと定義済みコマンドに重複があります。({chk_word})")
465                if chk_word in chk_members:
466                    raise RuntimeError(f"成績登録ワードと登録メンバー(チーム)に重複があります。({chk_word})")
467            # デフォルトルールバージョンチェック
468            if default_rule not in self.rule_list:
469                raise RuntimeError(f"デフォルトルールバージョンに指定されているルールセットが見つかりません。({default_rule})")
470        except RuntimeError as err:
471            logging.critical("%s", err)
472            sys.exit(1)
473
474    def register_to_database(self) -> None:
475        """ルールセット情報をDBに登録する"""
476        dbutil.execute("delete from rule;")
477        for rule in self.rule_list:
478            params = self.to_dict(rule)
479            params.update(rank_point=" ".join(map(str, params["rank_point"])))
480            dbutil.execute(
481                """
482                insert into
483                rule (
484                    rule_version, mode, origin_point, return_point, rank_point, ignore_flying, draw_split, undefined_word
485                ) values (
486                    :rule_version, :mode, :origin_point, :return_point, :rank_point, :ignore_flying, :draw_split, :undefined_word
487                );
488                """,
489                params,
490            )
491
492    @property
493    def rule_list(self) -> list[str]:
494        """
495        定義済みルール識別子の列挙
496
497        Returns:
498            list[str]: ルール識別子
499
500        """
501        return [x.rule_version for x in self.data.values()]