Telegram钓鱼小游戏Bot能否用.txt作用户数据库？是否需SQL？

是否必须使用SQL？ #

不是必须。存储方式的选择完全取决于你的Bot需求复杂度：

• 如果用户量少、数据结构简单（仅需记录用户ID、鱼种/数量），用文件存储（JSON/CSV）就足够，完全不需要数据库。
• 如果需要支持多用户并发访问、复杂查询（比如统计用户总钓鱼数、按鱼种排序展示），SQL类数据库（比如轻量的SQLite）或轻量NoSQL会更合适。

Python数据存储实操方案 #

方案1：JSON文件存储（入门首选） #

适合小体量数据，无需额外依赖，直接用Python内置json模块就能实现。

示例代码：

import json
from pathlib import Path

# 定义存储文件路径
DATA_FILE = Path("fishing_data.json")

# 初始化数据文件（不存在则创建）
if not DATA_FILE.exists():
    with open(DATA_FILE, "w", encoding="utf-8") as f:
        json.dump({}, f)

# 保存用户钓鱼记录
def save_catch(user_id, fish_type):
    with open(DATA_FILE, "r+", encoding="utf-8") as f:
        data = json.load(f)
        # 初始化用户记录（如果不存在）
        user_key = str(user_id)
        if user_key not in data:
            data[user_key] = {"total_fish": 0, "fish_types": {}}
        # 更新记录
        data[user_key]["total_fish"] += 1
        data[user_key]["fish_types"][fish_type] = data[user_key]["fish_types"].get(fish_type, 0) + 1
        # 写入文件
        f.seek(0)
        json.dump(data, f, indent=2, ensure_ascii=False)
        f.truncate()

# 查询用户钓鱼记录
def get_user_record(user_id):
    with open(DATA_FILE, "r", encoding="utf-8") as f:
        data = json.load(f)
        return data.get(str(user_id), {"total_fish": 0, "fish_types": {}})

# 测试调用
save_catch(123456, "鲫鱼")
print(get_user_record(123456))

方案2：SQLite数据库（轻量SQL方案） #

SQLite是Python内置的数据库，无需额外安装服务，适合需要结构化查询的场景，学习成本低。

示例代码：

import sqlite3
from pathlib import Path

DB_FILE = Path("fishing_bot.db")

# 初始化数据库表
def init_db():
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    # 创建用户表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            user_id INTEGER PRIMARY KEY,
            total_fish INTEGER DEFAULT 0
        )
    """)
    # 创建钓鱼记录表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS catches (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER,
            fish_type TEXT,
            catch_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (user_id) REFERENCES users(user_id)
        )
    """)
    conn.commit()
    conn.close()

# 保存钓鱼记录
def save_catch(user_id, fish_type):
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    # 插入/忽略用户记录（避免重复创建）
    cursor.execute("INSERT OR IGNORE INTO users (user_id) VALUES (?)", (user_id,))
    # 更新用户总钓鱼数
    cursor.execute("UPDATE users SET total_fish = total_fish + 1 WHERE user_id = ?", (user_id,))
    # 添加单条钓鱼记录
    cursor.execute("INSERT INTO catches (user_id, fish_type) VALUES (?, ?)", (user_id, fish_type))
    conn.commit()
    conn.close()

# 查询用户钓鱼记录
def get_user_record(user_id):
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    # 获取总钓鱼数
    cursor.execute("SELECT total_fish FROM users WHERE user_id = ?", (user_id,))
    total = cursor.fetchone()
    total_fish = total[0] if total else 0
    # 获取各鱼种数量
    cursor.execute("""
        SELECT fish_type, COUNT(*) as count 
        FROM catches 
        WHERE user_id = ? 
        GROUP BY fish_type
    """, (user_id,))
    fish_types = {row[0]: row[1] for row in cursor.fetchall()}
    conn.close()
    return {"total_fish": total_fish, "fish_types": fish_types}

# 初始化数据库
init_db()
# 测试调用
save_catch(123456, "鲤鱼")
print(get_user_record(123456))

方案3：Redis（缓存型NoSQL，高并发场景） #

如果Bot用户量较大，需要快速读写，可以用Redis，需先安装依赖：pip install redis。

示例代码：

import redis
from datetime import datetime

# 连接本地Redis服务
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 保存钓鱼记录
def save_catch(user_id, fish_type):
    user_key = f"user:{user_id}:catches"
    # 增加总钓鱼数
    r.hincrby(user_key, "total_fish", 1)
    # 增加对应鱼种数量
    r.hincrby(user_key, f"fish:{fish_type}", 1)
    # 记录钓鱼时间（可选）
    r.rpush(f"user:{user_id}:catch_log", f"{datetime.now().isoformat()} - {fish_type}")

# 查询用户钓鱼记录
def get_user_record(user_id):
    user_key = f"user:{user_id}:catches"
    data = r.hgetall(user_key)
    if not data:
        return {"total_fish": 0, "fish_types": {}}
    total_fish = int(data.get("total_fish", 0))
    fish_types = {}
    for key, value in data.items():
        if key.startswith("fish:"):
            fish_types[key.split(":", 1)[1]] = int(value)
    return {"total_fish": total_fish, "fish_types": fish_types}

# 测试调用
save_catch(123456, "草鱼")
print(get_user_record(123456))

选择建议 #

• 新手入门：先从JSON文件开始，理解数据存储逻辑后再过渡到数据库。
• 需要查询统计：选SQLite，内置无需额外配置，适合中小型Bot。
• 高并发场景：选Redis，读写速度快，还能兼顾用户状态缓存。

内容的提问来源于stack exchange，提问作者Ravi