libs.data.lookup.db

  1"""
  2lib/data/lookup/db.py
  3"""
  4
  5from contextlib import closing
  6from datetime import datetime
  7
  8import pandas as pd
  9
 10import libs.global_value as g
 11from cls.score import GameResult
 12from cls.types import TeamDataDict
 13from libs.data import loader
 14from libs.utils import dbutil
 15
 16
 17def get_member_id(name: str | None = None) -> dict:
 18    """メンバーのIDを返す
 19
 20    Args:
 21        name (str | None, optional): 指定メンバーのみ. Defaults to None.
 22
 23    Returns:
 24        dict: メンバー名とIDのペア
 25    """
 26
 27    with closing(dbutil.get_connection()) as conn:
 28        rows = conn.execute("select name, id from member;")
 29        id_list = dict(rows.fetchall())
 30
 31    if name in id_list:
 32        return {name: id_list[name]}
 33    return id_list
 34
 35
 36def member_info(name: str) -> dict:
 37    """指定メンバーの記録情報を返す
 38
 39    Args:
 40        name (str): 対象メンバー
 41
 42    Returns:
 43        dict: 記録情報
 44    """
 45
 46    ret: dict = {}
 47    sql = loader.query_modification("""
 48        select
 49            count() as game_count,
 50            min(ts) as first_game,
 51            max(ts) as last_game,
 52            max(rpoint) as max_rpoint,
 53            min(rpoint) as min_rpoint
 54        from
 55            --[individual] individual_results as results
 56            --[team] team_results as results
 57        where
 58            rule_version = ?
 59            and name = ?
 60    """)
 61
 62    with closing(dbutil.get_connection()) as conn:
 63        rows = conn.execute(sql, (g.cfg.mahjong.rule_version, name))
 64        ret = dict(rows.fetchone())
 65
 66    return ret
 67
 68
 69def get_guest() -> str:
 70    """ゲスト名取得
 71
 72    Returns:
 73        str: ゲスト名
 74    """
 75
 76    guest_name: str = ""
 77    with closing(dbutil.get_connection()) as conn:
 78        rows = conn.execute("select name from member where id=0")
 79        guest_name = str(rows.fetchone()[0])
 80
 81    return guest_name
 82
 83
 84def get_member_list() -> dict[str, str]:
 85    """メンバー情報取得
 86
 87    Returns:
 88        dict[str, str]: 別名, 表示名
 89    """
 90
 91    with closing(dbutil.get_connection()) as conn:
 92        rows = conn.execute("select name, member from alias")
 93        member_list = dict(rows.fetchall())
 94
 95    return member_list
 96
 97
 98def get_team_list() -> list[TeamDataDict]:
 99    """チーム情報取得
100
101    Returns:
102        list[TeamDataDict]: チーム情報
103    """
104
105    ret: list[TeamDataDict] = []
106    with closing(dbutil.get_connection()) as conn:
107        rows = conn.execute(
108            """
109                select
110                    team.id as id,
111                    team.name as team,
112                    group_concat(member.name) as member
113                from
114                    team
115                left join member on
116                    team.id == member.team_id
117                group by
118                    team.id
119            """)
120
121        for row in rows.fetchall():
122            ret.append({
123                "id": int(row["id"]),
124                "team": str(row["team"]),
125                "member": str(row["member"]).split(",")
126            })
127
128    return ret
129
130
131def rule_version_range() -> dict:
132    """DBに記録されているルールバージョン毎の範囲を取得する
133
134    Returns:
135        dict: 取得結果
136    """
137
138    rule: dict = {}
139    with closing(dbutil.get_connection()) as conn:
140        ret = conn.execute(
141            """
142            select
143                rule_version,
144                strftime("%Y/%m/%d %H:%M:%S", min(playtime)) as min,
145                strftime("%Y/%m/%d %H:%M:%S", max(playtime)) as max
146            from
147                result
148            group by
149                rule_version
150            """
151        )
152
153        for version, first_time, last_time in ret.fetchall():
154            rule[version] = {
155                "first_time": first_time,
156                "last_time": last_time,
157            }
158
159    return rule
160
161
162def regulation_list(word_type: int = 0) -> list:
163    """登録済みワードリストを取得する
164
165    Args:
166        word_type (int, optional): 取得するタイプ. Defaults to 0.
167
168    Returns:
169        list: 取得結果
170    """
171
172    with closing(dbutil.get_connection()) as cur:
173        ret = cur.execute(
174            """
175            select
176                word,
177                ex_point
178            from
179                words
180            where
181                type=?
182            """, (word_type,)
183        ).fetchall()
184
185    return ret
186
187
188def exsist_record(ts: str) -> GameResult:
189    """記録されているゲーム結果を返す
190
191    Args:
192        ts (str): 検索するタイムスタンプ
193
194    Returns:
195        GameResult: スコアデータ
196    """
197
198    result = GameResult()
199    with closing(dbutil.get_connection()) as conn:
200        row = conn.execute(g.sql["SELECT_GAME_RESULTS"], {"ts": ts}).fetchone()
201
202    if row:
203        result.calc(**dict(row))
204
205    return result
206
207
208def first_record() -> datetime:
209    """最初のゲーム記録時間を返す
210
211    Returns:
212        datetime: 最初のゲーム記録時間
213    """
214
215    ret = datetime.now()
216    try:
217        with closing(dbutil.get_connection()) as conn:
218            table_count = conn.execute(
219                "select count() from sqlite_master where type='view' and name='game_results';",
220            ).fetchall()[0][0]
221
222            if table_count:
223                record = conn.execute(
224                    "select min(playtime) from game_results where rule_version=?;",
225                    (g.params.get("rule_version", g.cfg.mahjong.rule_version), )
226                ).fetchall()[0][0]
227                if record:
228                    ret = datetime.fromisoformat(record)
229    except AttributeError:
230        ret = datetime.now()
231
232    return ret
233
234
235def get_results_list(name: str, rule_version: str = "") -> pd.DataFrame:
236    """段位集計用順位リスト生成
237
238    Args:
239        name (str): 集計対象メンバー名
240        rule_version (str, optional): 集計ルールバージョン. Defaults to 空欄.
241
242    Returns:
243        pd.DataFrame: 順位, 素点
244    """
245
246    ret_data = pd.read_sql(
247        sql=g.sql["SELECT_ALL_RESULTS"],
248        con=dbutil.get_connection(),
249        params={
250            "rule_version": rule_version if rule_version else g.cfg.mahjong.rule_version,
251            "player_name": name,
252        }
253    )
254
255    return ret_data