libs.domain.command
libs/domain/command.py
1""" 2libs/domain/command.py 3""" 4 5import re 6from collections.abc import Mapping 7from dataclasses import dataclass 8from typing import Any, Callable, Literal, TypedDict, Union 9 10import pandas as pd 11 12from libs.utils import formatter, textutil 13from libs.utils.timekit import ExtendedDatetime as ExtDt 14 15CommandResult = Mapping[str, Union[str, int, bool, tuple[str, ...]]] 16"""コマンド処理結果の型（パラメータ名とその値のマッピング）""" 17CommandAction = Callable[[Union[str, tuple[str, ...]]], CommandResult] 18"""コマンド処理関数の型（入力文字列を受け取り、結果辞書を返す）""" 19 20 21class CommandSpec(TypedDict, total=False): 22 """コマンドマッピングテーブル""" 23 24 match: list[str] 25 action: CommandAction 26 type: Literal["int", "str", "sql", "filename"] 27 28 29CommandsDict = dict[str, CommandSpec] 30COMMANDS: CommandsDict = { 31 # --- ゲスト処理 32 "guest_off": { 33 "match": [r"^ゲストナシ$"], 34 "action": lambda _: {"guest_skip": False, "guest_skip2": False, "unregistered_replace": True}, 35 }, 36 "guest_on": { 37 "match": [r"^ゲストアリ$"], 38 "action": lambda _: {"guest_skip": True, "guest_skip2": True, "unregistered_replace": True}, 39 }, 40 # --- 個人戦/チーム戦 41 "individual": { 42 "match": [r"^個人$", "^個人成績$"], 43 "action": lambda _: {"individual": True}, 44 }, 45 "team": { 46 "match": [r"^チーム$", "^チーム成績$", "^team$"], 47 "action": lambda _: {"individual": False}, 48 }, 49 "all_player": { 50 "match": [r"^全員$", r"^all$"], 51 "action": lambda _: {"all_player": True}, 52 }, 53 "a": { 54 "match": [r"^(チーム同卓アリ|コンビアリ|同士討チ)$"], 55 "action": lambda _: {"friendly_fire": True}, 56 }, 57 "b": { 58 "match": [r"^(チーム同卓ナシ|コンビナシ)$"], 59 "action": lambda _: {"friendly_fire": False}, 60 }, 61 # --- プレイヤー名変換処理 62 "guest_disable": { 63 "match": [r"^ゲスト無効$"], 64 "action": lambda _: {"unregistered_replace": False}, 65 }, 66 "anonymous": { 67 "match": [r"^匿名$", r"^anonymous$"], 68 "action": lambda _: {"anonymous": True}, 69 }, 70 # --- 動作変更フラグ 71 "score_comparisons": { # 比較 72 "match": [r"^比較$", r"^点差$", r"^差分$"], 73 "action": lambda _: {"score_comparisons": True}, 74 }, 75 "order": { # 順位出力 76 "match": [r"^順位$"], 77 "action": lambda _: {"order": True}, 78 }, 79 "results": { # 戦績 80 "match": [r"^戦績$"], 81 "action": lambda _: {"game_results": True}, 82 }, 83 "versus": { # 対戦結果 84 "match": [r"^対戦結果$", r"^対戦$"], 85 "action": lambda _: {"versus_matrix": True}, 86 }, 87 "statistics": { # 統計 88 "match": [r"^統計$"], 89 "action": lambda _: {"statistics": True}, 90 }, 91 "rating": { # レーティング 92 "match": [r"^レート$", r"^レーティング$", r"^rate$", r"^ratings?$"], 93 "action": lambda _: {"rating": True}, 94 }, 95 "verbose": { # 詳細 96 "match": [r"^詳細$", r"^verbose$"], 97 "action": lambda _: {"verbose": True}, 98 }, 99 # --- 集計条件 100 "ranked": { 101 "match": [r"^(トップ|上位|top)(\d*)$"], 102 "action": lambda w: {"ranked": w}, 103 }, 104 "stipulated": { 105 "match": [r"^(規定数|規定打数)(\d*)$"], 106 "action": lambda w: {"stipulated": w}, 107 }, 108 "interval": { 109 "match": [r"^(期間|区間|区切リ?|interval)(\d*)$"], 110 "action": lambda w: {"interval": w}, 111 }, 112 # --- 集約 / 検索条件 113 "daily": { 114 "match": [r"^daily$", r"^日次$", r"^デイリー$"], 115 "action": lambda _: {"collection": "daily"}, 116 }, 117 "weekly": { 118 "match": [r"^weekly$", r"^週次$", r"^ウイークリー$"], 119 "action": lambda _: {"collection": "weekly"}, 120 }, 121 "monthly": { 122 "match": [r"^monthly$", r"^月次$", r"^マンスリー$"], 123 "action": lambda _: {"collection": "monthly"}, 124 }, 125 "yearly": { 126 "match": [r"^yearly$", r"^年次$", r"^イヤーリー$"], 127 "action": lambda _: {"collection": "yearly"}, 128 }, 129 "collection": { 130 "match": [r"^全体$"], 131 "action": lambda _: {"collection": "all"}, 132 }, 133 "comment": { 134 "match": [r"^(コメント|comment)(.*)$"], 135 "action": lambda w: {"search_word": w}, 136 "type": "sql", 137 }, 138 "grouping": { 139 "match": [r"^(集約)(\d*)$"], 140 "action": lambda w: {"group_length": w}, 141 }, 142 "mode3": { 143 "match": [r"^三人打ち$", r"^三人打$", r"^三麻$", r"^サンマ$"], 144 "action": lambda _: {"target_mode": 3}, 145 }, 146 "mode4": { 147 "match": [r"^四人打ち$", r"^四人打$", r"^四麻$", r"^ヨンマ$"], 148 "action": lambda _: {"target_mode": 4}, 149 }, 150 "most_recent": { 151 "match": [r"^(直近)(\d*)$"], 152 "action": lambda w: {"target_count": w}, 153 }, 154 "mixed": { 155 "match": [r"^横断$", r"^mix$", r"^mixed$"], 156 "action": lambda _: {"mixed": True}, 157 }, 158 # --- 出力オプション 159 "format": { 160 "match": [r"^(csv|text|txt)$"], 161 "action": lambda w: {"format": w if w != "text" else "txt"}, 162 "type": "str", 163 }, 164 "filename": { 165 "match": [r"^(filename:|ファイル名)(.*)$"], 166 "action": lambda w: {"filename": w}, 167 "type": "filename", 168 }, 169} 170 171 172@dataclass 173class ParsedCommand: 174 """コマンド解析結果""" 175 176 flags: dict[str, Any] 177 """真偽値、引数を持つオプションを格納""" 178 arguments: list[str] 179 """単独オプションを格納""" 180 unknown: list[str] 181 """オプションと認識されない文字列を格納（プレイヤー名候補）""" 182 search_range: list[ExtDt] 183 """検索範囲の日時を格納""" 184 185 186class CommandParser: 187 """引数解析クラス""" 188 189 def __init__(self) -> None: 190 self.day_format = re.compile(r"^([0-9]{8}|[0-9/.-]{8,10})$") 191 """日付文字列判定用正規表現 192 - *yyyymmdd* 193 - *yyyy/mm/dd*, *yyyy/m/d* 194 - *yyyy-mm-dd*, *yyyy-m-d* 195 - *yyyy.mm.dd*, *yyyy.m.d* 196 """ 197 198 @classmethod 199 def is_valid_command(cls, word: str) -> bool: 200 """ 201 引数がコマンド名と一致するか判定する 202 203 Args: 204 word (str): チェック文字列 205 206 Returns: 207 bool: 真偽 208 209 """ 210 for cmd in COMMANDS.values(): 211 for pattern in cmd["match"]: 212 m = re.match(pattern, word) 213 if m: 214 return True 215 m = re.match(pattern, textutil.str_conv(word.lower(), textutil.ConversionType.HtoK)) 216 if m: 217 return True 218 219 return False 220 221 def analysis_argument(self, argument: list[str]) -> ParsedCommand: 222 """ 223 コマンドライン引数を解析する 224 225 Args: 226 argument (list[str]): 引数 227 228 Returns: 229 ParsedCommand: 結果 230 231 """ 232 ret: dict[str, Any] = {} 233 unknown: list[str] = [] 234 args: list[str] = [] 235 search_range: list[Any] = [] 236 237 for keyword in argument: 238 check_word = textutil.str_conv(keyword.lower(), textutil.ConversionType.HtoK) 239 check_word = check_word.replace("無シ", "ナシ").replace("有リ", "アリ") 240 241 if re.match(r"^([0-9]{8}|[0-9/.-]{8,10})$", check_word): 242 try_day = pd.to_datetime(check_word, errors="coerce").to_pydatetime() 243 if not pd.isna(try_day): 244 search_range.append(ExtDt(try_day)) 245 search_range.append(ExtDt(try_day) + {"hour": 23, "minute": 59, "second": 59, "microsecond": 999999}) 246 continue 247 248 if check_word in ExtDt.valid_keywords(): 249 search_range.append(check_word) 250 continue 251 252 for cmd in COMMANDS.values(): 253 for pattern in cmd["match"]: 254 m = re.match(pattern, keyword) 255 if m: 256 ret.update(self._parse_match(cmd, m)) 257 break 258 m = re.match(pattern, check_word) 259 if m: 260 ret.update(self._parse_match(cmd, m)) 261 break 262 else: 263 continue 264 break 265 else: 266 unknown.append(formatter.name_replace(keyword, add_mark=False, not_replace=True)) 267 268 return ParsedCommand(flags=ret, arguments=args, unknown=unknown, search_range=search_range) 269 270 def _parse_match(self, cmd: CommandSpec, obj: re.Match[str]) -> dict[str, Any]: 271 """ 272 コマンド名に一致したときの処理 273 274 Args: 275 cmd (CommandSpec): コマンドマップ 276 obj (re.Match[str]): Matchオブジェクト 277 278 Returns: 279 dict[str, Any]: 更新用辞書 280 281 """ 282 ret: dict[str, Any] = {} 283 284 def with_arguments(tmp: dict[str, Any]) -> None: 285 key = str(next(iter(tmp.keys()))) 286 val = str(tmp[key][1]) 287 if "" != val: 288 match cmd.get("type"): 289 case "str": 290 ret.update({key: val}) 291 case "sql": 292 ret.update({key: f"%{val}%"}) 293 case "filename": 294 if re.search(r"^[\w()\-\.]+$", val): 295 ret.update({key: val}) 296 case "int": 297 ret.update({key: int(val)}) 298 case _: 299 ret.update({key: int(val) if val.isdigit() else val}) 300 301 match len(obj.groups()): 302 case 0: # 完全一致: ^command$ 303 ret.update(cmd["action"](obj.group())) 304 case 1: # 選択: ^(command1|command2|...)$ 305 ret.update(cmd["action"](obj.groups()[0])) 306 case 2: # 引数あり: ^(command)(\d*)$ 307 tmp = cmd["action"](obj.groups()) 308 if isinstance(tmp, dict): 309 for k, v in tmp.items(): 310 if isinstance(v, tuple): # 引数取り出し&セット 311 with_arguments(tmp) 312 if isinstance(v, bool): # フラグ上書き 313 ret.update({k: v}) 314 315 return ret
CommandResult =
collections.abc.Mapping[str, str | int | bool | tuple[str, ...]]
コマンド処理結果の型（パラメータ名とその値のマッピング）
CommandAction =
typing.Callable[[str | tuple[str, ...]], collections.abc.Mapping[str, str | int | bool | tuple[str, ...]]]
コマンド処理関数の型（入力文字列を受け取り、結果辞書を返す）
class
CommandSpec(typing.TypedDict):
22class CommandSpec(TypedDict, total=False): 23 """コマンドマッピングテーブル""" 24 25 match: list[str] 26 action: CommandAction 27 type: Literal["int", "str", "sql", "filename"]
コマンドマッピングテーブル
CommandsDict =
dict[str, CommandSpec]
COMMANDS: dict[str, CommandSpec] =
{'guest_off': {'match': ['^ゲストナシ$'], 'action': <function <lambda>>}, 'guest_on': {'match': ['^ゲストアリ$'], 'action': <function <lambda>>}, 'individual': {'match': ['^個人$', '^個人成績$'], 'action': <function <lambda>>}, 'team': {'match': ['^チーム$', '^チーム成績$', '^team$'], 'action': <function <lambda>>}, 'all_player': {'match': ['^全員$', '^all$'], 'action': <function <lambda>>}, 'a': {'match': ['^(チーム同卓アリ|コンビアリ|同士討チ)$'], 'action': <function <lambda>>}, 'b': {'match': ['^(チーム同卓ナシ|コンビナシ)$'], 'action': <function <lambda>>}, 'guest_disable': {'match': ['^ゲスト無効$'], 'action': <function <lambda>>}, 'anonymous': {'match': ['^匿名$', '^anonymous$'], 'action': <function <lambda>>}, 'score_comparisons': {'match': ['^比較$', '^点差$', '^差分$'], 'action': <function <lambda>>}, 'order': {'match': ['^順位$'], 'action': <function <lambda>>}, 'results': {'match': ['^戦績$'], 'action': <function <lambda>>}, 'versus': {'match': ['^対戦結果$', '^対戦$'], 'action': <function <lambda>>}, 'statistics': {'match': ['^統計$'], 'action': <function <lambda>>}, 'rating': {'match': ['^レート$', '^レーティング$', '^rate$', '^ratings?$'], 'action': <function <lambda>>}, 'verbose': {'match': ['^詳細$', '^verbose$'], 'action': <function <lambda>>}, 'ranked': {'match': ['^(トップ|上位|top)(\\d*)$'], 'action': <function <lambda>>}, 'stipulated': {'match': ['^(規定数|規定打数)(\\d*)$'], 'action': <function <lambda>>}, 'interval': {'match': ['^(期間|区間|区切リ?|interval)(\\d*)$'], 'action': <function <lambda>>}, 'daily': {'match': ['^daily$', '^日次$', '^デイリー$'], 'action': <function <lambda>>}, 'weekly': {'match': ['^weekly$', '^週次$', '^ウイークリー$'], 'action': <function <lambda>>}, 'monthly': {'match': ['^monthly$', '^月次$', '^マンスリー$'], 'action': <function <lambda>>}, 'yearly': {'match': ['^yearly$', '^年次$', '^イヤーリー$'], 'action': <function <lambda>>}, 'collection': {'match': ['^全体$'], 'action': <function <lambda>>}, 'comment': {'match': ['^(コメント|comment)(.*)$'], 'action': <function <lambda>>, 'type': 'sql'}, 'grouping': {'match': ['^(集約)(\\d*)$'], 'action': <function <lambda>>}, 'mode3': {'match': ['^三人打ち$', '^三人打$', '^三麻$', '^サンマ$'], 'action': <function <lambda>>}, 'mode4': {'match': ['^四人打ち$', '^四人打$', '^四麻$', '^ヨンマ$'], 'action': <function <lambda>>}, 'most_recent': {'match': ['^(直近)(\\d*)$'], 'action': <function <lambda>>}, 'mixed': {'match': ['^横断$', '^mix$', '^mixed$'], 'action': <function <lambda>>}, 'format': {'match': ['^(csv|text|txt)$'], 'action': <function <lambda>>, 'type': 'str'}, 'filename': {'match': ['^(filename:|ファイル名)(.*)$'], 'action': <function <lambda>>, 'type': 'filename'}}
@dataclass
class
ParsedCommand:
173@dataclass 174class ParsedCommand: 175 """コマンド解析結果""" 176 177 flags: dict[str, Any] 178 """真偽値、引数を持つオプションを格納""" 179 arguments: list[str] 180 """単独オプションを格納""" 181 unknown: list[str] 182 """オプションと認識されない文字列を格納（プレイヤー名候補）""" 183 search_range: list[ExtDt] 184 """検索範囲の日時を格納"""
コマンド解析結果
ParsedCommand( flags: dict[str, typing.Any], arguments: list[str], unknown: list[str], search_range: list[libs.utils.timekit.ExtendedDatetime])
class
CommandParser:
187class CommandParser: 188 """引数解析クラス""" 189 190 def __init__(self) -> None: 191 self.day_format = re.compile(r"^([0-9]{8}|[0-9/.-]{8,10})$") 192 """日付文字列判定用正規表現 193 - *yyyymmdd* 194 - *yyyy/mm/dd*, *yyyy/m/d* 195 - *yyyy-mm-dd*, *yyyy-m-d* 196 - *yyyy.mm.dd*, *yyyy.m.d* 197 """ 198 199 @classmethod 200 def is_valid_command(cls, word: str) -> bool: 201 """ 202 引数がコマンド名と一致するか判定する 203 204 Args: 205 word (str): チェック文字列 206 207 Returns: 208 bool: 真偽 209 210 """ 211 for cmd in COMMANDS.values(): 212 for pattern in cmd["match"]: 213 m = re.match(pattern, word) 214 if m: 215 return True 216 m = re.match(pattern, textutil.str_conv(word.lower(), textutil.ConversionType.HtoK)) 217 if m: 218 return True 219 220 return False 221 222 def analysis_argument(self, argument: list[str]) -> ParsedCommand: 223 """ 224 コマンドライン引数を解析する 225 226 Args: 227 argument (list[str]): 引数 228 229 Returns: 230 ParsedCommand: 結果 231 232 """ 233 ret: dict[str, Any] = {} 234 unknown: list[str] = [] 235 args: list[str] = [] 236 search_range: list[Any] = [] 237 238 for keyword in argument: 239 check_word = textutil.str_conv(keyword.lower(), textutil.ConversionType.HtoK) 240 check_word = check_word.replace("無シ", "ナシ").replace("有リ", "アリ") 241 242 if re.match(r"^([0-9]{8}|[0-9/.-]{8,10})$", check_word): 243 try_day = pd.to_datetime(check_word, errors="coerce").to_pydatetime() 244 if not pd.isna(try_day): 245 search_range.append(ExtDt(try_day)) 246 search_range.append(ExtDt(try_day) + {"hour": 23, "minute": 59, "second": 59, "microsecond": 999999}) 247 continue 248 249 if check_word in ExtDt.valid_keywords(): 250 search_range.append(check_word) 251 continue 252 253 for cmd in COMMANDS.values(): 254 for pattern in cmd["match"]: 255 m = re.match(pattern, keyword) 256 if m: 257 ret.update(self._parse_match(cmd, m)) 258 break 259 m = re.match(pattern, check_word) 260 if m: 261 ret.update(self._parse_match(cmd, m)) 262 break 263 else: 264 continue 265 break 266 else: 267 unknown.append(formatter.name_replace(keyword, add_mark=False, not_replace=True)) 268 269 return ParsedCommand(flags=ret, arguments=args, unknown=unknown, search_range=search_range) 270 271 def _parse_match(self, cmd: CommandSpec, obj: re.Match[str]) -> dict[str, Any]: 272 """ 273 コマンド名に一致したときの処理 274 275 Args: 276 cmd (CommandSpec): コマンドマップ 277 obj (re.Match[str]): Matchオブジェクト 278 279 Returns: 280 dict[str, Any]: 更新用辞書 281 282 """ 283 ret: dict[str, Any] = {} 284 285 def with_arguments(tmp: dict[str, Any]) -> None: 286 key = str(next(iter(tmp.keys()))) 287 val = str(tmp[key][1]) 288 if "" != val: 289 match cmd.get("type"): 290 case "str": 291 ret.update({key: val}) 292 case "sql": 293 ret.update({key: f"%{val}%"}) 294 case "filename": 295 if re.search(r"^[\w()\-\.]+$", val): 296 ret.update({key: val}) 297 case "int": 298 ret.update({key: int(val)}) 299 case _: 300 ret.update({key: int(val) if val.isdigit() else val}) 301 302 match len(obj.groups()): 303 case 0: # 完全一致: ^command$ 304 ret.update(cmd["action"](obj.group())) 305 case 1: # 選択: ^(command1|command2|...)$ 306 ret.update(cmd["action"](obj.groups()[0])) 307 case 2: # 引数あり: ^(command)(\d*)$ 308 tmp = cmd["action"](obj.groups()) 309 if isinstance(tmp, dict): 310 for k, v in tmp.items(): 311 if isinstance(v, tuple): # 引数取り出し&セット 312 with_arguments(tmp) 313 if isinstance(v, bool): # フラグ上書き 314 ret.update({k: v}) 315 316 return ret
引数解析クラス
@classmethod
def
is_valid_command(cls, word: str) -> bool:
199 @classmethod 200 def is_valid_command(cls, word: str) -> bool: 201 """ 202 引数がコマンド名と一致するか判定する 203 204 Args: 205 word (str): チェック文字列 206 207 Returns: 208 bool: 真偽 209 210 """ 211 for cmd in COMMANDS.values(): 212 for pattern in cmd["match"]: 213 m = re.match(pattern, word) 214 if m: 215 return True 216 m = re.match(pattern, textutil.str_conv(word.lower(), textutil.ConversionType.HtoK)) 217 if m: 218 return True 219 220 return False
引数がコマンド名と一致するか判定する
Arguments:
- word (str): チェック文字列
Returns:
bool: 真偽
222 def analysis_argument(self, argument: list[str]) -> ParsedCommand: 223 """ 224 コマンドライン引数を解析する 225 226 Args: 227 argument (list[str]): 引数 228 229 Returns: 230 ParsedCommand: 結果 231 232 """ 233 ret: dict[str, Any] = {} 234 unknown: list[str] = [] 235 args: list[str] = [] 236 search_range: list[Any] = [] 237 238 for keyword in argument: 239 check_word = textutil.str_conv(keyword.lower(), textutil.ConversionType.HtoK) 240 check_word = check_word.replace("無シ", "ナシ").replace("有リ", "アリ") 241 242 if re.match(r"^([0-9]{8}|[0-9/.-]{8,10})$", check_word): 243 try_day = pd.to_datetime(check_word, errors="coerce").to_pydatetime() 244 if not pd.isna(try_day): 245 search_range.append(ExtDt(try_day)) 246 search_range.append(ExtDt(try_day) + {"hour": 23, "minute": 59, "second": 59, "microsecond": 999999}) 247 continue 248 249 if check_word in ExtDt.valid_keywords(): 250 search_range.append(check_word) 251 continue 252 253 for cmd in COMMANDS.values(): 254 for pattern in cmd["match"]: 255 m = re.match(pattern, keyword) 256 if m: 257 ret.update(self._parse_match(cmd, m)) 258 break 259 m = re.match(pattern, check_word) 260 if m: 261 ret.update(self._parse_match(cmd, m)) 262 break 263 else: 264 continue 265 break 266 else: 267 unknown.append(formatter.name_replace(keyword, add_mark=False, not_replace=True)) 268 269 return ParsedCommand(flags=ret, arguments=args, unknown=unknown, search_range=search_range)
コマンドライン引数を解析する
Arguments:
- argument (list[str]): 引数
Returns:
ParsedCommand: 結果