db_utils.py

# 云端Space代码/db_utils.py
# ==========================================
# 🔧 P2代码质量优化：数据库工具函数
# ==========================================
# 作用：封装 JSON 数据库常用操作，减少重复代码
# 关联文件：
#   - 数据库连接.py (基础读写)
#   - router_tasks.py (任务操作)
#   - router_posts.py (帖子操作)
#   - router_items.py (商品操作)
# ==========================================

from typing import Any, Dict, List, Optional, Callable, Union
import 数据库连接 as db


# ==========================================
# 📖 查询工具函数
# ==========================================

def get_by_id(file_name: str, item_id: str, id_field: str = "id") -> Optional[Dict]:
    """

    根据 ID 获取单个记录

    

    参数：

        file_name: JSON 文件名（如 tasks.json）

        item_id: 要查找的 ID

        id_field: ID 字段名（默认 "id"）

    

    返回：

        找到的记录，或 None

    

    示例：

        task = get_by_id("tasks.json", "task_123")

        user = get_by_id("users.json", "user@example.com", id_field="account")

    """
    data = db.load_data(file_name, default_data=[])
    
    if isinstance(data, dict):
        return data.get(item_id)
    
    return next((item for item in data if item.get(id_field) == item_id), None)


def get_by_field(file_name: str, field: str, value: Any) -> Optional[Dict]:
    """

    根据指定字段获取单个记录

    

    参数：

        file_name: JSON 文件名

        field: 字段名

        value: 字段值

    

    返回：

        找到的记录，或 None

    """
    data = db.load_data(file_name, default_data=[])
    
    if isinstance(data, dict):
        for item in data.values():
            if isinstance(item, dict) and item.get(field) == value:
                return item
        return None
    
    return next((item for item in data if item.get(field) == value), None)


def filter_by(file_name: str, **conditions) -> List[Dict]:
    """

    根据条件筛选记录

    

    参数：

        file_name: JSON 文件名

        **conditions: 筛选条件（键值对）

    

    返回：

        符合条件的记录列表

    

    示例：

        open_tasks = filter_by("tasks.json", status="open")

        user_posts = filter_by("posts.json", author="user123", deleted=False)

    """
    data = db.load_data(file_name, default_data=[])
    
    if isinstance(data, dict):
        data = list(data.values())
    
    result = []
    for item in data:
        if all(item.get(key) == value for key, value in conditions.items()):
            result.append(item)
    
    return result


def count_by(file_name: str, **conditions) -> int:
    """

    统计符合条件的记录数量

    

    参数：

        file_name: JSON 文件名

        **conditions: 筛选条件

    

    返回：

        符合条件的记录数量

    """
    return len(filter_by(file_name, **conditions))


# ==========================================
# ✏️ 更新工具函数
# ==========================================

def update_by_id(file_name: str, item_id: str, updates: Dict, id_field: str = "id") -> bool:
    """

    根据 ID 更新记录

    

    参数：

        file_name: JSON 文件名

        item_id: 要更新的 ID

        updates: 要更新的字段（键值对）

        id_field: ID 字段名

    

    返回：

        True 更新成功 / False 记录不存在

    

    示例：

        update_by_id("tasks.json", "task_123", {"status": "completed"})

    """
    data = db.load_data(file_name, default_data=[])
    
    if isinstance(data, dict):
        if item_id in data:
            data[item_id].update(updates)
            db.save_data(file_name, data)
            return True
        return False
    
    for item in data:
        if item.get(id_field) == item_id:
            item.update(updates)
            db.save_data(file_name, data)
            return True
    
    return False


def update_with_fn(file_name: str, item_id: str, update_fn: Callable[[Dict], None], id_field: str = "id") -> bool:
    """

    使用函数更新记录（支持复杂更新逻辑）

    

    参数：

        file_name: JSON 文件名

        item_id: 要更新的 ID

        update_fn: 更新函数，接收记录 dict，直接修改

        id_field: ID 字段名

    

    返回：

        True 更新成功 / False 记录不存在

    

    示例：

        def increment_views(item):

            item["views"] = item.get("views", 0) + 1

        

        update_with_fn("items.json", "item_123", increment_views)

    """
    data = db.load_data(file_name, default_data=[])
    
    if isinstance(data, dict):
        if item_id in data:
            update_fn(data[item_id])
            db.save_data(file_name, data)
            return True
        return False
    
    for item in data:
        if item.get(id_field) == item_id:
            update_fn(item)
            db.save_data(file_name, data)
            return True
    
    return False


# ==========================================
# ➕ 添加工具函数
# ==========================================

def insert(file_name: str, item: Dict, prepend: bool = True) -> bool:
    """

    插入新记录

    

    参数：

        file_name: JSON 文件名

        item: 要插入的记录

        prepend: True 插入到开头 / False 插入到末尾

    

    返回：

        True 插入成功

    

    示例：

        insert("tasks.json", {"id": "task_123", "title": "新任务"})

    """
    data = db.load_data(file_name, default_data=[])
    
    if isinstance(data, dict):
        item_id = item.get("id") or item.get("account")
        if item_id:
            data[item_id] = item
            db.save_data(file_name, data)
            return True
        return False
    
    if prepend:
        data.insert(0, item)
    else:
        data.append(item)
    
    db.save_data(file_name, data)
    return True


def insert_if_not_exists(file_name: str, item: Dict, id_field: str = "id") -> bool:
    """

    如果不存在则插入

    

    参数：

        file_name: JSON 文件名

        item: 要插入的记录

        id_field: ID 字段名

    

    返回：

        True 插入成功 / False 已存在

    """
    item_id = item.get(id_field)
    if not item_id:
        return False
    
    existing = get_by_id(file_name, item_id, id_field)
    if existing:
        return False
    
    return insert(file_name, item)


# ==========================================
# ❌ 删除工具函数
# ==========================================

def delete_by_id(file_name: str, item_id: str, id_field: str = "id") -> bool:
    """

    根据 ID 删除记录

    

    参数：

        file_name: JSON 文件名

        item_id: 要删除的 ID

        id_field: ID 字段名

    

    返回：

        True 删除成功 / False 记录不存在

    """
    data = db.load_data(file_name, default_data=[])
    
    if isinstance(data, dict):
        if item_id in data:
            del data[item_id]
            db.save_data(file_name, data)
            return True
        return False
    
    original_len = len(data)
    data = [item for item in data if item.get(id_field) != item_id]
    
    if len(data) < original_len:
        db.save_data(file_name, data)
        return True
    
    return False


def soft_delete_by_id(file_name: str, item_id: str, id_field: str = "id") -> bool:
    """

    软删除（标记为已删除，不物理删除）

    

    参数：

        file_name: JSON 文件名

        item_id: 要删除的 ID

        id_field: ID 字段名

    

    返回：

        True 成功 / False 记录不存在

    """
    import time
    return update_by_id(file_name, item_id, {
        "deleted": True,
        "deleted_at": int(time.time())
    }, id_field)


# ==========================================
# 🔍 分页工具函数
# ==========================================

def paginate(

    file_name: str, 

    page: int = 1, 

    limit: int = 20, 

    sort_by: str = None,

    sort_desc: bool = True,

    **filters

) -> Dict:
    """

    分页查询

    

    参数：

        file_name: JSON 文件名

        page: 页码（从 1 开始）

        limit: 每页数量

        sort_by: 排序字段

        sort_desc: 是否降序

        **filters: 筛选条件

    

    返回：

        {

            "data": [...],      # 当前页数据

            "total": 100,       # 总数

            "page": 1,          # 当前页

            "limit": 20,        # 每页数量

            "pages": 5          # 总页数

        }

    """
    # 获取并筛选数据
    if filters:
        data = filter_by(file_name, **filters)
    else:
        data = db.load_data(file_name, default_data=[])
        if isinstance(data, dict):
            data = list(data.values())
    
    # 排序
    if sort_by:
        try:
            data = sorted(data, key=lambda x: x.get(sort_by, 0), reverse=sort_desc)
        except TypeError:
            pass  # 排序失败时忽略
    
    # 计算分页
    total = len(data)
    pages = (total + limit - 1) // limit  # 向上取整
    start = (page - 1) * limit
    end = start + limit
    
    return {
        "data": data[start:end],
        "total": total,
        "page": page,
        "limit": limit,
        "pages": pages
    }


# ==========================================
# 👁️ 访问量记录工具函数
# ==========================================

def record_view(data_file: str, item_id: str, user_account: str) -> Optional[Dict]:
    """

    记录访问量（原子操作，并发安全）

    

    参数：

        data_file: JSON 文件名（如 Items.json）

        item_id: 要记录访问的 item/task ID

        user_account: 访问用户账号

    

    返回：

        {"views": N, "daily_views": N} 成功

        None 记录不存在

    

    逻辑：

        - 初始化字段: views=0, viewed_by=[], daily_views=0, daily_views_date=""

        - daily_views 每次调用都增加

        - views 只在用户首次访问时增加（user_account 不在 viewed_by 中）

        - 如果 daily_views_date 不是今天，重置 daily_views=0 并更新日期

    

    并发安全：

        - 使用 atomic_update 确保读-改-写在同一把锁内完成

        - 高并发下不会丢失访问量

    """
    from datetime import datetime, timedelta
    
    # 用于在闭包中存储结果
    result_container = [None]
    
    def updater(data):
        # 查找目标记录
        target_item = None
        
        if isinstance(data, dict):
            if item_id in data:
                target_item = data[item_id]
        else:
            for item in data:
                if item.get("id") == item_id:
                    target_item = item
                    break
        
        if target_item is None:
            result_container[0] = None
            return
        
        # 初始化字段（如果不存在）
        if "views" not in target_item:
            target_item["views"] = 0
        if "viewed_by" not in target_item:
            target_item["viewed_by"] = []
        if "daily_views" not in target_item:
            target_item["daily_views"] = 0
        if "daily_views_date" not in target_item:
            target_item["daily_views_date"] = ""
        
        # 获取今天的日期字符串（使用 UTC+8 时区）
        now_utc8 = datetime.utcnow() + timedelta(hours=8)
        today_str = now_utc8.date().isoformat()  # "2026-04-02"
        
        # 检查是否需要重置日访问量
        if target_item["daily_views_date"] != today_str:
            target_item["daily_views"] = 0
            target_item["daily_views_date"] = today_str
        
        # 增加日访问量（每次调用都增加）
        target_item["daily_views"] += 1
        
        # 检查用户是否已访问过
        if user_account not in target_item["viewed_by"]:
            target_item["viewed_by"].append(user_account)
            target_item["views"] += 1
        
        # 保存结果到闭包容器
        result_container[0] = {
            "views": target_item["views"],
            "daily_views": target_item["daily_views"]
        }
    
    # 使用原子更新，整个读-改-写过程在同一把锁内完成
    db.atomic_update(data_file, updater, default_data=[])
    
    return result_container[0]


# ==========================================
# 🗂️ 排序结果缓存工具类
# ==========================================

import time
from typing import Callable, List, Dict, Any

class SortCache:
    """

    排序结果缓存类 - 用于缓存列表排序结果，减少重复排序开销

    

    设计原则：

    1. 缓存排序后的 ID 顺序，而非完整数据，避免内存浪费

    2. 使用 TTL (默认5分钟) 自动过期，确保数据新鲜度

    3. 写操作时清除相关缓存，确保数据一致性

    4. 在 load_data 返回的最新数据之上应用缓存的顺序

    

    使用示例：

        # 在列表接口中

        def get_items(sort="time"):

            items = db.load_data("items.json", default_data=[])

            cache_key = f"items:{sort}"

            

            def sort_fn(data):

                if sort == "likes":

                    data.sort(key=lambda x: x.get("likes", 0), reverse=True)

                else:

                    data.sort(key=lambda x: x.get("created_at", 0), reverse=True)

            

            return sort_cache.get_sorted(cache_key, items, sort_fn)

        

        # 在写操作后

        def create_item(...):

            ...

            sort_cache.invalidate("items")

    """
    
    def __init__(self, ttl: int = 300):
        """

        初始化排序缓存

        

        参数：

            ttl: 缓存过期时间（秒），默认 300 秒（5分钟）

        """
        self._cache: Dict[str, tuple] = {}  # {cache_key: (sorted_ids, timestamp)}
        self._ttl = ttl
    
    def get_sorted(self, cache_key: str, items: List[Dict], sort_fn: Callable[[List[Dict]], None]) -> List[Dict]:
        """

        获取排序后的数据（带缓存）

        

        参数：

            cache_key: 缓存键，应包含数据文件和排序参数

            items: 原始数据列表（来自 load_data 的最新数据）

            sort_fn: 排序函数，接收 items 列表并原地排序

        

        返回：

            排序后的 items 列表

        """
        now = time.time()
        
        # 检查缓存是否有效
        if cache_key in self._cache:
            sorted_ids, cached_time = self._cache[cache_key]
            if now - cached_time < self._ttl:
                # 缓存有效，用缓存的顺序重排当前数据
                id_order = {id_: idx for idx, id_ in enumerate(sorted_ids)}
                # 按缓存的顺序排序，新数据（不在缓存中的）放在最后
                return sorted(items, key=lambda x: id_order.get(x.get("id"), float('inf')))
        
        # 缓存无效或不存在，执行排序
        sort_fn(items)
        
        # 缓存排序后的 ID 列表
        sorted_ids = [item.get("id") for item in items]
        self._cache[cache_key] = (sorted_ids, now)
        
        return items
    
    def invalidate(self, prefix: str = ""):
        """

        清除缓存

        

        参数：

            prefix: 缓存键前缀，如果指定则只清除匹配的缓存，否则清除所有缓存

        """
        if prefix:
            keys_to_remove = [k for k in self._cache if k.startswith(prefix)]
            for k in keys_to_remove:
                del self._cache[k]
        else:
            self._cache.clear()
    
    def get_stats(self) -> Dict[str, Any]:
        """

        获取缓存统计信息（用于调试）

        """
        now = time.time()
        valid_count = sum(1 for _, ts in self._cache.values() if now - ts < self._ttl)
        return {
            "total_cached": len(self._cache),
            "valid": valid_count,
            "expired": len(self._cache) - valid_count,
            "ttl": self._ttl
        }


# 全局排序缓存实例（TTL 5分钟）
sort_cache = SortCache(ttl=300)


# ==========================================
# 🔄 批量操作工具函数
# ==========================================

def batch_update(file_name: str, item_ids: List[str], updates: Dict, id_field: str = "id") -> int:
    """

    批量更新记录

    

    参数：

        file_name: JSON 文件名

        item_ids: 要更新的 ID 列表

        updates: 要更新的字段

        id_field: ID 字段名

    

    返回：

        更新的记录数量

    """
    data = db.load_data(file_name, default_data=[])
    updated_count = 0
    
    if isinstance(data, dict):
        for item_id in item_ids:
            if item_id in data:
                data[item_id].update(updates)
                updated_count += 1
    else:
        id_set = set(item_ids)
        for item in data:
            if item.get(id_field) in id_set:
                item.update(updates)
                updated_count += 1
    
    if updated_count > 0:
        db.save_data(file_name, data)
    
    return updated_count


def batch_delete(file_name: str, item_ids: List[str], id_field: str = "id") -> int:
    """

    批量删除记录

    

    参数：

        file_name: JSON 文件名

        item_ids: 要删除的 ID 列表

        id_field: ID 字段名

    

    返回：

        删除的记录数量

    """
    data = db.load_data(file_name, default_data=[])
    original_count = len(data) if isinstance(data, list) else len(data)
    
    if isinstance(data, dict):
        for item_id in item_ids:
            data.pop(item_id, None)
    else:
        id_set = set(item_ids)
        data = [item for item in data if item.get(id_field) not in id_set]
    
    new_count = len(data) if isinstance(data, list) else len(data)
    deleted_count = original_count - new_count
    
    if deleted_count > 0:
        db.save_data(file_name, data)
    
    return deleted_count