cls.parser

  1"""
  2cls/parser.py
  3"""
  4
  5import logging
  6import re
  7from dataclasses import dataclass
  8from typing import Any
  9
 10import pandas as pd
 11from slack_sdk import WebClient
 12
 13import libs.global_value as g
 14from cls.timekit import ExtendedDatetime as ExtDt
 15from cls.types import CommandSpec
 16from libs.data import lookup
 17from libs.utils import textutil
 18
 19
 20@dataclass
 21class ParsedCommand:
 22    """コマンド解析結果"""
 23    flags: dict[str, Any]
 24    arguments: list[str]
 25    unknown: list[str]
 26    search_range: list["ExtDt"]
 27
 28
 29CommandsDict = dict[str, CommandSpec]
 30COMMANDS: CommandsDict = {
 31    "guest": {
 32        "match": [r"^ゲストナシ$", r"^ゲストアリ$", r"^ゲスト無効$"],
 33        "action": lambda w: {
 34            "ゲストナシ": {"guest_skip": False, "guest_skip2": False, "unregistered_replace": True},
 35            "ゲストアリ": {"guest_skip": True, "guest_skip2": True, "unregistered_replace": True},
 36            "ゲスト無効": {"unregistered_replace": False},
 37        }[w[0] if isinstance(w, tuple) else w],
 38    },
 39    "anonymous": {
 40        "match": [r"^匿名$", r"^anonymous$"],
 41        "action": lambda w: {"anonymous": True},
 42    },
 43
 44    "individual": {
 45        "match": [r"^個人$", "^個人成績$"],
 46        "action": lambda w: {"individual": True},
 47    },
 48    "team": {
 49        "match": [r"^チーム$", "^チーム成績$", "^team$"],
 50        "action": lambda w: {"individual": False},
 51    },
 52
 53    "all_player": {
 54        "match": [r"^全員$", r"^all$"],
 55        "action": lambda w: {"all_player": True},
 56    },
 57
 58    "a": {
 59        "match": [r"^(チーム同卓アリ|コンビアリ|同士討チ)$"],
 60        "action": lambda w: {"friendly_fire": True},
 61    },
 62    "b": {
 63        "match": [r"^(チーム同卓ナシ|コンビナシ)$"],
 64        "action": lambda w: {"friendly_fire": False},
 65    },
 66    # --- 動作変更フラグ
 67    "score_comparisons": {  # 比較
 68        "match": [r"^比較$", r"^点差$", r"^差分$"],
 69        "action": lambda w: {"score_comparisons": True},
 70    },
 71    "order": {  # 順位出力
 72        "match": [r"^順位$"],
 73        "action": lambda w: {"order": True},
 74    },
 75    "results": {  # 戦績
 76        "match": [r"^戦績$"],
 77        "action": lambda w: {"game_results": True},
 78    },
 79    "versus": {  # 対戦結果
 80        "match": [r"^対戦結果$", r"^対戦$"],
 81        "action": lambda w: {"versus_matrix": True},
 82    },
 83    "statistics": {  # 統計
 84        "match": [r"^統計$"],
 85        "action": lambda w: {"statistics": True},
 86    },
 87    "rating": {  # レーティング
 88        "match": [r"^レート$", r"^レーティング$", r"^rate$", r"^ratings?$"],
 89        "action": lambda w: {"rating": True},
 90    },
 91    "verbose": {  # 詳細
 92        "match": [r"^詳細$", r"^verbose$"],
 93        "action": lambda w: {"verbose": True},
 94    },
 95    # --- 集計条件
 96    "ranked": {
 97        "match": [r"^(トップ|上位|top)(\d*)$"],
 98        "action": lambda w: {"ranked": w},
 99    },
100    "stipulated": {
101        "match": [r"^(規定数|規定打数)(\d*)$"],
102        "action": lambda w: {"stipulated": w}
103    },
104    "interval": {
105        "match": [r"^(期間|区間|区切リ?|interval)(\d*)$"],
106        "action": lambda w: {"interval": w}
107    },
108    # --- 集約 / 検索条件
109    "daily": {
110        "match": [r"^daily$", r"^日次$", r"^デイリー$"],
111        "action": lambda w: {"collection": "daily"},
112    },
113    "monthly": {
114        "match": [r"^monthly$", r"^月次$", r"^マンスリー$"],
115        "action": lambda w: {"collection": "monthly"},
116    },
117    "yearly": {
118        "match": [r"^yearly$", r"^年次$", r"^イヤーリー$"],
119        "action": lambda w: {"collection": "yearly"},
120    },
121    "collection": {
122        "match": [r"^全体$"],
123        "action": lambda w: {"collection": "all"}
124    },
125    "comment": {
126        "match": [r"^(コメント|comment)(.*)$"],
127        "action": lambda w: {"search_word": w},
128        "type": "sql",
129    },
130    "grouping": {
131        "match": [r"^(集約)(\d*)$"],
132        "action": lambda w: {"group_length": w}
133    },
134    "rule_version": {
135        "match": [r"^(ルール|rule)(.*)$"],
136        "action": lambda w: {"rule_version": w},
137        "type": "str",
138    },
139    "most_recent": {
140        "match": [r"^(直近)(\d*)$"],
141        "action": lambda w: {"target_count": w}
142    },
143    # --- 出力オプション
144    "format": {
145        "match": [r"^(csv|text|txt)$"],
146        "action": lambda w: {"format": w},
147        "type": "str",
148    },
149    "filename": {
150        "match": [r"^(filename:|ファイル名)(.*)$"],
151        "action": lambda w: {"filename": w},
152        "type": "filename",
153    },
154}
155
156
157class CommandParser:
158    """引数解析クラス"""
159
160    def __init__(self):
161        self.day_format = re.compile(r"^([0-9]{8}|[0-9/.-]{8,10})$")
162        """日付文字列判定用正規表現
163        - *yyyymmdd*
164        - *yyyy/mm/dd*, *yyyy/m/d*
165        - *yyyy-mm-dd*, *yyyy-m-d*
166        - *yyyy.mm.dd*, *yyyy.m.d*
167        """
168
169    @classmethod
170    def is_valid_command(cls, word: str) -> bool:
171        """引数がコマンド名と一致するか判定する
172
173        Args:
174            word (str): チェック文字列
175
176        Returns:
177            bool: 真偽
178        """
179
180        for cmd in COMMANDS.values():
181            for pattern in cmd["match"]:
182                m = re.match(pattern, word)
183                if m:
184                    return True
185                m = re.match(pattern, textutil.str_conv(word.lower(), "h2k"))
186                if m:
187                    return True
188
189        return False
190
191    def analysis_argument(self, argument: list[str]) -> ParsedCommand:
192        """コマンドライン引数を解析する
193
194        Args:
195            argument (list[str]): 引数
196
197        Returns:
198            ParsedCommand: 結果
199        """
200
201        ret: dict = {}
202        unknown: list = []
203        args: list = []
204        search_range: list = []
205
206        for keyword in argument:
207            check_word = textutil.str_conv(keyword.lower(), "h2k")
208            check_word = check_word.replace("無シ", "ナシ").replace("有リ", "アリ")
209
210            if re.match(r"^([0-9]{8}|[0-9/.-]{8,10})$", check_word):
211                try_day = pd.to_datetime(check_word, errors="coerce").to_pydatetime()
212                if not pd.isna(try_day):
213                    search_range.append(ExtDt(try_day))
214                    search_range.append(ExtDt(try_day) + {"hour": 23, "minute": 59, "second": 59, "microsecond": 999999})
215                continue
216
217            if check_word in ExtDt.valid_keywords():
218                search_range.append(check_word)
219                continue
220
221            for cmd in COMMANDS.values():
222                for pattern in cmd["match"]:
223                    m = re.match(pattern, keyword)
224                    if m:
225                        ret.update(self._parse_match(cmd, m))
226                        break
227                    m = re.match(pattern, check_word)
228                    if m:
229                        ret.update(self._parse_match(cmd, m))
230                        break
231                else:
232                    continue
233                break
234            else:
235                unknown.append(keyword)
236
237        return ParsedCommand(flags=ret, arguments=args, unknown=unknown, search_range=search_range)
238
239    def _parse_match(self, cmd: CommandSpec, m: re.Match) -> dict:
240        """コマンド名に一致したときの処理
241
242        Args:
243            cmd (CommandSpec): コマンドマップ
244            m (re.Match): Matchオブジェクト
245
246        Returns:
247            dict: 更新用辞書
248        """
249        ret: dict = {}
250
251        match len(m.groups()):
252            case 0:  # 完全一致: ^command$
253                ret.update(cmd["action"](m.group()))
254            case 1:  # 選択: ^(command1|command2|...)$
255                ret.update(cmd["action"](m.groups()[0]))
256            case 2:  # 引数あり: ^(command)(\d*)$
257                tmp = cmd["action"](m.groups())
258                if isinstance(tmp, dict):
259                    key = next(iter(tmp.keys()))
260                    val = str(tmp[key][1])
261                    if "" != val:
262                        match cmd.get("type"):
263                            case "str":
264                                ret.update({key: val})
265                            case "sql":
266                                ret.update({key: f"%{val}%"})
267                            case "filename":
268                                if re.search(r"^[\w\-\.]+$", val):
269                                    ret.update({key: val})
270                            case "int":
271                                ret.update({key: int(val)})
272                            case _:
273                                ret.update({key: int(val) if val.isdigit() else val})
274
275        return ret
276
277
278class MessageParser:
279    """メッセージ解析クラス"""
280    client: WebClient = WebClient()
281    """slack WebClient オブジェクト"""
282
283    def __init__(self, body: dict | None = None):
284        self.channel_id: str | None = str()
285        """ポストされたチャンネルのID"""
286        self.channel_type: str | None = str()
287        """チャンネルタイプ
288        - *channel*: 通常チャンネル
289        - *group*: プライベートチャンネル
290        - *im*: ダイレクトメッセージ
291        - *search_messages*: 検索API
292        """
293        self.user_id: str = str()
294        """ポストしたユーザのID"""
295        self.text: str | None = str()
296        """ポストされた文字列"""
297        self.event_ts: str = str()  # テキストのまま処理する
298        """タイムスタンプ"""
299        self.thread_ts: str = str()  # テキストのまま処理する
300        """スレッドになっている場合のスレッド元のタイムスタンプ"""
301        self.status: str = str()  # event subtype
302        """イベントステータス
303        - *message_append*: 新規ポスト
304        - *message_changed*: 編集
305        - *message_deleted*: 削除
306        """
307        self.keyword: str = str()
308        self.argument: list = []
309        self.updatable: bool = bool()
310        self.in_thread: bool = bool()
311
312        if isinstance(body, dict):
313            self.parser(body)
314
315    def parser(self, _body: dict):
316        """postされたメッセージをパースする
317
318        Args:
319            _body (dict): postされたデータ
320        """
321
322        logging.trace(_body)  # type: ignore
323
324        # 初期値
325        self.text = ""
326        self.channel_id = ""
327        self.user_id = ""
328        self.thread_ts = "0"
329        self.keyword = ""
330        self.argument = []
331
332        if _body.get("command") == g.cfg.setting.slash_command:  # スラッシュコマンド
333            _event = _body
334            if not self.channel_id:
335                if _body.get("channel_name") == "directmessage":
336                    self.channel_id = _body.get("channel_id", None)
337                else:
338                    self.channel_id = lookup.api.get_dm_channel_id(_body.get("user_id", ""))
339
340        if _body.get("container"):  # Homeタブ
341            self.user_id = _body["user"].get("id")
342            self.channel_id = lookup.api.get_dm_channel_id(self.user_id)
343            self.text = "dummy"
344
345        _event = self.get_event_attribute(_body)
346        self.user_id = _event.get("user", self.user_id)
347        self.event_ts = _event.get("ts", self.event_ts)
348        self.thread_ts = _event.get("thread_ts", self.thread_ts)
349        self.channel_type = self.get_channel_type(_body)
350
351        # スレッド内のポストか判定
352        if float(self.thread_ts):
353            self.in_thread = self.event_ts != self.thread_ts
354        else:
355            self.in_thread = False
356
357        if "text" in _event:
358            self.text = _event.get("text")
359            if self.text:  # 空文字以外はキーワードと引数に分割
360                self.keyword = self.text.split()[0]
361                self.argument = self.text.split()[1:]
362        else:  # text属性が見つからないときはログに出力
363            if not _event.get("text") and not _body.get("type") == "block_actions":
364                logging.error("text not found: %s", _body)
365
366        self.check_updatable()
367        logging.info("channel_id=%s, channel_type=%s", self.channel_id, self.channel_type)
368
369    def get_event_attribute(self, _body: dict) -> dict:
370        """レスポンスからevent属性を探索して返す
371
372        Args:
373            _body (dict): レスポンス内容
374
375        Returns:
376            dict: event属性
377        """
378
379        _event: dict = {}
380
381        if _body.get("command") == g.cfg.setting.slash_command:
382            _event = _body
383
384        if _body.get("event"):
385            if not self.channel_id:
386                if _body.get("channel_name") != "directmessage":
387                    self.channel_id = _body["event"].get("channel")
388                else:
389                    self.channel_id = lookup.api.get_dm_channel_id(_body.get("user_id", ""))
390
391            match _body["event"].get("subtype"):
392                case "message_changed":
393                    self.status = "message_changed"
394                    _event = _body["event"]["message"]
395                case "message_deleted":
396                    self.status = "message_deleted"
397                    _event = _body["event"]["previous_message"]
398                case "file_share":
399                    self.status = "message_append"
400                    _event = _body["event"]
401                case None:
402                    self.status = "message_append"
403                    _event = _body["event"]
404                case _:
405                    self.status = "message_append"
406                    _event = _body["event"]
407                    logging.info("unknown subtype: %s", _body)
408
409        return _event
410
411    def get_channel_type(self, _body: dict) -> str | None:
412        """レスポンスからchannel_typeを探索して返す
413
414        Args:
415            _body (dict): レスポンス内容
416
417        Returns:
418            str | None: channel_type
419        """
420
421        _channel_type: str | None = None
422
423        if _body.get("command") == g.cfg.setting.slash_command:
424            if _body.get("channel_name") == "directmessage":
425                _channel_type = "im"
426            else:
427                _channel_type = "channel"
428        else:
429            if _body.get("event"):
430                _channel_type = _body["event"].get("channel_type")
431
432        return _channel_type
433
434    def check_updatable(self):
435        """DB更新可能チャンネルのポストかチェックする"""
436        self.updatable = False
437
438        if g.cfg.db.channel_limitations:
439            if self.channel_id in g.cfg.db.channel_limitations:
440                self.updatable = True
441        else:  # リストが空なら全チャンネルが対象
442            match self.channel_type:
443                case "channel":  # public channel
444                    self.updatable = True
445                case "group":  # private channel
446                    self.updatable = True
447                case "im":  # direct message
448                    self.updatable = False
449                case "search_messages":
450                    self.updatable = True
451                case _:
452                    self.updatable = True