integrations.base.interface

  1"""
  2integrations/base/interface.py
  3"""
  4
  5from abc import ABC, abstractmethod
  6from configparser import ConfigParser
  7from dataclasses import dataclass, field
  8from types import NoneType
  9from typing import TYPE_CHECKING, Any, Generic, Literal, Optional, Type, TypeVar
 10
 11import pandas as pd
 12
 13if TYPE_CHECKING:
 14    from pathlib import Path
 15
 16    from integrations.protocols import MessageParserProtocol, MsgData, PostData, StatusData
 17    from libs.types import MessageType, StyleOptions
 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    _parser: Optional[ConfigParser] = field(default=None)
 46
 47    # ディスパッチテーブル用
 48    _command_dispatcher: dict[str, Any] = field(default_factory=dict)
 49    _keyword_dispatcher: dict[str, Any] = field(default_factory=dict)
 50
 51    # 共通設定
 52    main_conf: Optional[ConfigParser] = field(default=None)
 53    """設定ファイル"""
 54    channel_config: Optional["Path"] = field(default=None)
 55    """チャンネル個別設定状況
 56    - *Path*: チャンネル個別設定ファイルパス
 57    - *None*: 個別設定を利用していない
 58    """
 59
 60    slash_command: str = field(default="")
 61    """スラッシュコマンド名"""
 62    badge_degree: bool = field(default=False)
 63    """プレイしたゲーム数に対して表示される称号
 64    - *True*: 表示する
 65    - *False*: 表示しない
 66    """
 67    badge_status: bool = field(default=False)
 68    """勝率に対して付く調子バッジ
 69    - *True*: 表示する
 70    - *False*: 表示しない
 71    """
 72    badge_grade: bool = field(default=False)
 73    """段位表示
 74    - *True*: 表示する
 75    - *False*: 表示しない
 76    """
 77
 78    separate: bool = field(default=False)
 79    """スコア入力元識別子別集計フラグ
 80    - *True*: 識別子別に集計
 81    - *False*: すべて集計
 82    """
 83    channel_id: Optional[str] = field(default=None)
 84    """チャンネルIDを上書きする"""
 85
 86    plotting_backend: Literal["matplotlib", "plotly"] = field(default="matplotlib")
 87    """グラフ描写ライブラリ"""
 88
 89    @property
 90    def command_dispatcher(self) -> dict[str, Any]:
 91        """
 92        コマンドディスパッチテーブルを辞書で取得
 93
 94        Returns:
 95            dict[str, Any]: コマンドディスパッチテーブル
 96
 97        """
 98        return self._command_dispatcher
 99
100    @property
101    def keyword_dispatcher(self) -> dict[str, Any]:
102        """
103        キーワードディスパッチテーブルを辞書で取得
104
105        Returns:
106            dict[str, Any]: キーワードディスパッチテーブル
107
108        """
109        return self._keyword_dispatcher
110
111
112class FunctionsInterface(ABC):
113    """個別関数インターフェース"""
114
115    @abstractmethod
116    def post_processing(self, m: "MessageParserProtocol") -> None:
117        """
118        後処理
119
120        Args:
121            m (MessageParserProtocol): メッセージデータ
122
123        """
124
125    @abstractmethod
126    def get_conversations(self, m: "MessageParserProtocol") -> dict[str, Any]:
127        """
128        スレッド情報の取得
129
130        Args:
131            m (MessageParserProtocol): メッセージデータ
132
133        Returns:
134            dict: API response
135
136        """
137        return {}
138
139
140class APIInterface(ABC):
141    """アダプタAPIインターフェース"""
142
143    @abstractmethod
144    def post(self, m: "MessageParserProtocol") -> None:
145        """
146        メッセージを出力する
147
148        Args:
149            m (MessageParserProtocol): メッセージデータ
150
151        """
152
153
154class MessageParserDataMixin:
155    """メッセージ解析共通処理"""
156
157    data: "MsgData"
158    post: "PostData"
159    status: "StatusData"
160
161    def reset(self) -> None:
162        """初期化"""
163        self.data.reset()
164        self.post.reset()
165        self.status.reset()
166
167    def set_headline(
168        self,
169        data: "MessageType",
170        options: "StyleOptions",
171    ) -> None:
172        """
173        ヘッドラインメッセージをセット
174
175        Args:
176            data (MessageType): 内容
177            options (StyleOptions): 表示オプション
178
179        """
180        # 空データは登録しない
181        if isinstance(data, NoneType) or (isinstance(data, pd.DataFrame) and data.empty):
182            self.post.headline = None
183        else:
184            self.post.headline = (data, options)
185
186    def set_message(
187        self,
188        data: "MessageType",
189        options: "StyleOptions",
190    ) -> None:
191        """
192        本文メッセージをセット
193
194        Args:
195            data (MessageType): 内容
196            options (StyleOptions): 表示オプション
197
198        """
199        # 空データは登録しない
200        if isinstance(data, NoneType) or (isinstance(data, pd.DataFrame) and data.empty):
201            return
202
203        self.post.message.append((data, options))
204
205
206class MessageParserInterface(ABC):
207    """メッセージ解析インターフェース"""
208
209    data: "MsgData"
210    post: "PostData"
211    status: "StatusData"
212
213    @abstractmethod
214    def parser(self, body: Any) -> None:
215        """
216        メッセージ解析
217
218        Args:
219            body (Any): 解析データ
220
221        """
222
223    @property
224    @abstractmethod
225    def in_thread(self) -> bool:
226        """
227        元メッセージへのリプライとなっているか
228
229        Returns:
230            bool: 真偽値
231            - *True*: リプライの形（リプライ／スレッドなど）
232            - *False*: 通常メッセージ
233
234        """
235
236    @property
237    @abstractmethod
238    def is_bot(self) -> bool:
239        """
240        botのポストかチェック
241
242        Returns:
243            bool: 真偽値
244            - *True*: botのポスト
245            - *False*: ユーザのポスト
246
247        """
248
249    @property
250    @abstractmethod
251    def check_updatable(self) -> bool:
252        """
253        DB操作の許可チェック
254
255        Returns:
256            bool: 真偽値
257            - *True*: 許可
258            - *False*: 禁止
259
260        """
261
262    @property
263    @abstractmethod
264    def ignore_user(self) -> bool:
265        """
266        ignore_useridに存在するユーザかチェック
267
268        Returns:
269            bool: 真偽値
270            - *True*: 存在する(操作禁止ユーザ)
271            - *False*: 存在しない
272
273        """
274
275    @property
276    def is_command(self) -> bool:
277        """
278        コマンドで実行されているかチェック
279
280        Returns:
281            bool: 真偽値
282            - *True*: コマンド実行
283            - *False*: 非コマンド(キーワード呼び出し)
284
285        """
286        return self.status.command_flg
287
288    @property
289    def keyword(self) -> str:
290        """
291        コマンドとして認識している文字列を返す
292
293        Returns:
294            str: コマンド名
295
296        """
297        if ret := self.data.text.split():
298            return ret[0]
299        return self.data.text
300
301    @property
302    def argument(self) -> list[str]:
303        """
304        コマンド引数として認識している文字列をリストで返す
305
306        Returns:
307            list[str]: 引数リスト
308
309        """
310        if ret := self.data.text.split():
311            return ret[1:]
312        return ret
313
314    @property
315    def reply_ts(self) -> str:
316        """
317        リプライ先のタイムスタンプを取得する
318
319        Returns:
320            str: タイムスタンプ
321
322        """
323        ret_ts: str = "0"
324
325        # tsが指定されていれば最優先
326        if self.post.ts != "undetermined":
327            return self.post.ts
328
329        # スレッドに返すか
330        if self.post.thread and self.in_thread:  # スレッド内
331            if self.data.thread_ts != "undetermined":
332                ret_ts = self.data.thread_ts
333        elif self.post.thread and not self.in_thread:  # スレッド外
334            ret_ts = self.data.event_ts
335
336        return ret_ts