integrations.base.interface

  1"""
  2integrations/base/interface.py
  3"""
  4
  5import re
  6from abc import ABC, abstractmethod
  7from configparser import ConfigParser
  8from dataclasses import dataclass, field, fields
  9from typing import (TYPE_CHECKING, Any, Generic, Literal, Optional, Type,
 10                    TypeVar, Union)
 11
 12from integrations.protocols import MsgData, PostData, StatusData
 13from libs.types import MessageTypeDict, StyleOptions
 14
 15if TYPE_CHECKING:
 16    from integrations.protocols import MessageParserProtocol
 17    from libs.types import MessageType
 18
 19ConfigT = TypeVar("ConfigT", bound="IntegrationsConfig")
 20ApiT = TypeVar("ApiT", bound="APIInterface")
 21FunctionsT = TypeVar("FunctionsT", bound="FunctionsInterface")
 22ParserT = TypeVar("ParserT", bound="MessageParserInterface")
 23
 24
 25class AdapterInterface(ABC, Generic[ConfigT, ApiT, FunctionsT, ParserT]):
 26    """アダプタインターフェース"""
 27
 28    interface_type: str
 29    """サービス識別子"""
 30
 31    conf: ConfigT
 32    """個別設定データクラス"""
 33    api: ApiT
 34    """インターフェース操作APIインスタンス"""
 35    functions: FunctionsT
 36    """サービス専用関数インスタンス"""
 37    parser: Type[ParserT]
 38    """メッセージパーサクラス"""
 39
 40
 41@dataclass
 42class IntegrationsConfig(ABC):
 43    """個別設定値"""
 44
 45    # ディスパッチテーブル用
 46    _command_dispatcher: dict = field(default_factory=dict)
 47    _keyword_dispatcher: dict = field(default_factory=dict)
 48
 49    config_file: Optional[ConfigParser] = field(default=None)
 50    """設定ファイル"""
 51
 52    # 共通設定
 53    slash_command: str = field(default="")
 54    """スラッシュコマンド名"""
 55
 56    badge_degree: bool = field(default=False)
 57    """プレイしたゲーム数に対して表示される称号
 58    - *True*: 表示する
 59    - *False*: 表示しない
 60    """
 61    badge_status: bool = field(default=False)
 62    """勝率に対して付く調子バッジ
 63    - *True*: 表示する
 64    - *False*: 表示しない
 65    """
 66    badge_grade: bool = field(default=False)
 67    """段位表示
 68    - *True*: 表示する
 69    - *False*: 表示しない
 70    """
 71
 72    plotting_backend: Literal["matplotlib", "plotly"] = field(default="matplotlib")
 73    """グラフ描写ライブラリ"""
 74
 75    @property
 76    def command_dispatcher(self) -> dict:
 77        """コマンドディスパッチテーブルを辞書で取得
 78
 79        Returns:
 80            dict: コマンドディスパッチテーブル
 81        """
 82
 83        return self._command_dispatcher
 84
 85    @property
 86    def keyword_dispatcher(self) -> dict:
 87        """キーワードディスパッチテーブルを辞書で取得
 88
 89        Returns:
 90            dict: キーワードディスパッチテーブル
 91        """
 92
 93        return self._keyword_dispatcher
 94
 95    def read_file(self, selected_service: str):
 96        """設定値取り込み
 97
 98        Args:
 99            selected_service (str): セクション
100
101        Raises:
102            TypeError: 無効な型が指定されている場合
103        """
104
105        if self.config_file is None:
106            raise TypeError("Configuration file not specified.")
107
108        value: Union[int, float, bool, str, list]
109        if self.config_file.has_section(selected_service):
110            for f in fields(self):
111                if f.name.startswith("_"):
112                    continue
113                if self.config_file.has_option(selected_service, f.name):
114                    if f.type is int:
115                        value = self.config_file.getint(selected_service, f.name)
116                    elif f.type is float:
117                        value = self.config_file.getfloat(selected_service, f.name)
118                    elif f.type is bool:
119                        value = self.config_file.getboolean(selected_service, f.name)
120                    elif f.type is str:
121                        value = self.config_file.get(selected_service, f.name)
122                    elif f.type is list:
123                        value = [x.strip() for x in self.config_file.get(selected_service, f.name).split(",")]
124                    else:
125                        raise TypeError(f"Unsupported type: {f.type}")
126                    setattr(self, f.name, value)
127
128
129class FunctionsInterface(ABC):
130    """個別関数インターフェース"""
131
132    @abstractmethod
133    def post_processing(self, m: "MessageParserProtocol"):
134        """後処理
135
136        Args:
137            m (MessageParserProtocol): メッセージデータ
138        """
139
140    @abstractmethod
141    def get_conversations(self, m: "MessageParserProtocol") -> dict:
142        """スレッド情報の取得
143
144        Args:
145            m (MessageParserProtocol): メッセージデータ
146
147        Returns:
148            dict: API response
149        """
150        return {}
151
152
153class APIInterface(ABC):
154    """アダプタAPIインターフェース"""
155
156    @abstractmethod
157    def post(self, m: "MessageParserProtocol"):
158        """メッセージを出力する
159
160        Args:
161            m (MessageParserProtocol): メッセージデータ
162        """
163
164
165class MessageParserDataMixin:
166    """メッセージ解析共通処理"""
167
168    data: "MsgData"
169    post: "PostData"
170    status: "StatusData"
171
172    def reset(self) -> None:
173        """初期化"""
174
175        self.data.reset()
176        self.post.reset()
177        self.status.reset()
178
179    def set_data(
180        self,
181        title: str,
182        data: "MessageType",
183        options: StyleOptions,
184    ):
185        """メッセージデータをセットshow_index
186
187        Args:
188            title (str): データ識別子
189            data (MessageType): 内容
190            options (StyleOptions): 表示オプション
191        """
192
193        msg = MessageTypeDict(
194            data=data,
195            options=options,
196        )
197        self.post.message.append({title: msg})
198
199    def get_score(self, keyword: str) -> dict:
200        """textからスコアを抽出する
201
202        Args:
203            keyword (str): 成績報告キーワード
204
205        Returns:
206            dict: 結果
207        """
208
209        text = self.data.text
210        ret: dict = {}
211
212        # 記号を置換
213        replace_chr = [
214            (chr(0xff0b), "+"),  # 全角プラス符号
215            (chr(0x2212), "-"),  # 全角マイナス符号
216            (chr(0xff08), "("),  # 全角丸括弧
217            (chr(0xff09), ")"),  # 全角丸括弧
218            (chr(0x2017), "_"),  # DOUBLE LOW LINE(半角)
219        ]
220        for z, h in replace_chr:
221            text = text.replace(z, h)
222
223        text = "".join(text.split())  # 改行削除
224
225        # パターンマッチング
226        pattern1 = re.compile(
227            rf"^({keyword})" + r"([^0-9()+-]+)([0-9+-]+)" * 4 + r"$"
228        )
229        pattern2 = re.compile(
230            r"^" + r"([^0-9()+-]+)([0-9+-]+)" * 4 + rf"({keyword})$"
231        )
232        pattern3 = re.compile(
233            rf"^({keyword})\((.+?)\)" + r"([^0-9()+-]+)([0-9+-]+)" * 4 + r"$"
234        )
235        pattern4 = re.compile(
236            r"^" + r"([^0-9()+-]+)([0-9+-]+)" * 4 + rf"({keyword})\((.+?)\)$"
237        )
238
239        # 情報取り出し
240        position: dict[str, int] = {}
241        match text:
242            case text if pattern1.findall(text):
243                msg = pattern1.findall(text)[0]
244                position = {
245                    "p1_name": 1, "p1_str": 2,
246                    "p2_name": 3, "p2_str": 4,
247                    "p3_name": 5, "p3_str": 6,
248                    "p4_name": 7, "p4_str": 8,
249                }
250                comment = None
251            case text if pattern2.findall(text):
252                msg = pattern2.findall(text)[0]
253                position = {
254                    "p1_name": 0, "p1_str": 1,
255                    "p2_name": 2, "p2_str": 3,
256                    "p3_name": 4, "p3_str": 5,
257                    "p4_name": 6, "p4_str": 7,
258                }
259                comment = None
260            case text if pattern3.findall(text):
261                msg = pattern3.findall(text)[0]
262                position = {
263                    "p1_name": 2, "p1_str": 3,
264                    "p2_name": 4, "p2_str": 5,
265                    "p3_name": 6, "p3_str": 7,
266                    "p4_name": 8, "p4_str": 9,
267                }
268                comment = str(msg[1])
269            case text if pattern4.findall(text):
270                msg = pattern4.findall(text)[0]
271                position = {
272                    "p1_name": 0, "p1_str": 1,
273                    "p2_name": 2, "p2_str": 3,
274                    "p3_name": 4, "p3_str": 5,
275                    "p4_name": 6, "p4_str": 7,
276                }
277                comment = str(msg[9])
278            case _:
279                return ret
280
281        for k, p in position.items():
282            ret.update({k: str(msg[p])})
283
284        ret.update(comment=comment)
285        ret.update(source=self.status.source)
286        ret.update(ts=self.data.event_ts)
287
288        return ret
289
290    def get_remarks(self, keyword: str) -> list:
291        """textからメモを抽出する
292
293        Args:
294            keyword (str): メモ記録キーワード
295
296        Returns:
297            list: 結果
298        """
299
300        ret: list = []
301        if re.match(rf"^{keyword}", self.data.text):  # キーワードが先頭に存在するかチェック
302            text = self.data.text.replace(keyword, "").strip().split()
303            for name, matter in zip(text[0::2], text[1::2]):
304                ret.append([name, matter])
305
306        return ret
307
308
309class MessageParserInterface(ABC):
310    """メッセージ解析インターフェース"""
311
312    data: "MsgData"
313    post: "PostData"
314    status: "StatusData"
315
316    @abstractmethod
317    def parser(self, body: Any):
318        """メッセージ解析
319
320        Args:
321            body (Any): 解析データ
322        """
323
324    @property
325    @abstractmethod
326    def in_thread(self) -> bool:
327        """元メッセージへのリプライとなっているか
328
329        Returns:
330            bool: 真偽値
331            - *True* : リプライの形（リプライ／スレッドなど）
332            - *False* : 通常メッセージ
333        """
334
335    @property
336    @abstractmethod
337    def is_bot(self) -> bool:
338        """botのポストかチェック
339
340        Returns:
341            bool: 真偽値
342            - *True* : botのポスト
343            - *False* : ユーザのポスト
344        """
345
346    @property
347    @abstractmethod
348    def check_updatable(self) -> bool:
349        """DB操作の許可チェック
350
351        Returns:
352            bool: 真偽値
353            - *True* : 許可
354            - *False* : 禁止
355        """
356
357    @property
358    @abstractmethod
359    def ignore_user(self) -> bool:
360        """ignore_useridに存在するユーザかチェック
361
362        Returns:
363            bool: 真偽値
364            - *True* : 存在する(操作禁止ユーザ)
365            - *False* : 存在しない
366        """
367
368    @property
369    def is_command(self) -> bool:
370        """コマンドで実行されているかチェック
371
372        Returns:
373            bool: 真偽値
374            - *True* : コマンド実行
375            - *False* : 非コマンド(キーワード呼び出し)
376        """
377
378        return self.status.command_flg
379
380    @property
381    def keyword(self) -> str:
382        """コマンドとして認識している文字列を返す
383
384        Returns:
385            str: コマンド名
386        """
387
388        if (ret := self.data.text.split()):
389            return ret[0]
390        return self.data.text
391
392    @property
393    def argument(self) -> list:
394        """コマンド引数として認識している文字列をリストで返す
395
396        Returns:
397            list: 引数リスト
398        """
399
400        if (ret := self.data.text.split()):
401            return ret[1:]
402        return ret
403
404    @property
405    def reply_ts(self) -> str:
406        """リプライ先のタイムスタンプを取得する
407
408        Returns:
409            str: タイムスタンプ
410        """
411
412        ret_ts: str = "0"
413
414        # tsが指定されていれば最優先
415        if self.post.ts != "undetermined":
416            return self.post.ts
417
418        # スレッドに返すか
419        if self.post.thread and self.in_thread:  # スレッド内
420            if self.data.thread_ts != "undetermined":
421                ret_ts = self.data.thread_ts
422        elif self.post.thread and not self.in_thread:  # スレッド外
423            ret_ts = self.data.event_ts
424
425        return ret_ts