配置管理

import pandas as pd
import json
import time
from pathlib import Path
from functools import wraps
from sqlmodel import create_engine

engine_config = create_engine('sqlite:///volume/dbs/config.db')

class ConfigTable:

    def __init__(self):
        self.path = Path('volume/dbs/update.lock')
        if not self.path.exists():
            self.path.touch()
        self.update_time = self.path.stat().st_mtime
        self.master_table = self._load_master_table()
        # 启动时一次性加载所有配置表到内存
        self.tables = {
            tn: pd.read_sql(tn, engine_config, index_col=index_col)
            if index_col else pd.read_sql(tn, engine_config)
            for tn, index_col in self.master_table['index_col'].items()
        }

    @property
    def config(self):
        """通过 tables.config['表名', '列名'] 访问"""
        return self.master_table.loc

    @staticmethod
    def _load_master_table():
        mt = pd.read_sql_table('table_master', engine_config, index_col='table_name')
        mt['index_col'] = mt['index_col'].apply(json.loads)
        mt['bool_col'] = mt['bool_col'].apply(json.loads)
        return mt

    @staticmethod
    def ensure_new(func):
        """装饰器：访问前检查文件锁时间戳，有变化才重新加载"""
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            if self.path.stat().st_mtime != self.update_time:
                new_master = self._load_master_table()
                # 只重新加载发生变化的表（对比 update_time 列）
                old_time = self.master_table['update_time']
                new_time = new_master['update_time']
                changed = new_time[new_time != old_time.reindex(new_time.index)].index
                for tn in changed:
                    self.tables[tn] = pd.read_sql(
                        tn, engine_config,
                        index_col=new_master.loc[tn, 'index_col']
                    )
                self.master_table = new_master
                self.update_time = self.path.stat().st_mtime
            return func(self, *args, **kwargs)
        return wrapper

    @ensure_new
    def __getitem__(self, table_name: str) -> pd.DataFrame:
        if table_name not in self.tables:
            raise KeyError(f'Table {table_name} not found')
        return self.tables[table_name]

tables = ConfigTable()

async def update_table(self, table_name: str):
    # 1. 从外部源拉取数据
    df = pd.read_csv(f'{source_url}/csv/{table_name}', ...)

    # 2. 写入 SQLite（持久化）
    df.to_sql(table_name, engine_config, index=True, if_exists='replace')

    # 3. 更新 master_table 的时间戳
    self.master_table.loc[table_name, 'update_time'] = int(time.time() * 1000)
    self.master_table.to_sql('master_table', engine_config, ...)

    # 4. 更新内存中的表
    self.tables[table_name] = df

    # 5. 触摸文件锁，通知其他进程
    self.path.touch()
    self.update_time = self.path.stat().st_mtime

@register_dataframe_accessor("safe_at")
class SafeLocAccessor:
    # 回退规则：索引找不到时，尝试这些默认值
    fallback_rules = {
        'game_cd': 0,      # 找不到特定游戏配置 → 用默认配置
        'lang': 'en',      # 找不到特定语言 → 用英语
    }

    def __getitem__(self, key: tuple):
        k0, k1 = key  # (索引值, 列名)
        if k0 in self._df.index:
            return self._df.at[key]
        # 多级索引：用笛卡尔积遍历所有回退组合
        ...

# 直接取值（可能 KeyError）
value = tables['prompt_config'].at['game_a', 'system_prompt']

# 安全取值（自动回退）
value = tables['prompt_config'].safe_at['game_a', 'system_prompt']
# 如果 game_a 不存在，自动尝试 game_cd=0 的默认配置

# 全局缓存
game_expire_dict = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时从API加载
    response = await client.get(f'{api_url}/config')
    game_expire_dict.update(response.json())
    yield