从零做一个完整的机器人

import sqlite3from contextlib import contextmanagerfrom pathlib import PathDB_PATH = Path(__file__).parent / "data.db"@contextmanagerdef connect():    conn = sqlite3.connect(DB_PATH)    try:        yield conn        conn.commit()    finally:        conn.close()def init_db() -> None:    with connect() as conn:        conn.execute("""            CREATE TABLE IF NOT EXISTS checkins (                user_id INTEGER NOT NULL,                day     TEXT    NOT NULL,                PRIMARY KEY (user_id, day)            )        """)        conn.execute("""            CREATE TABLE IF NOT EXISTS users (                user_id INTEGER PRIMARY KEY,                name    TEXT    NOT NULL,                points  INTEGER NOT NULL DEFAULT 0            )        """)def check_in(user_id: int, name: str, day: str, reward: int) -> int | None:    """签到。成功返回签到后的总分；今天已经签过返回 None。"""    with connect() as conn:        try:            conn.execute(                "INSERT INTO checkins (user_id, day) VALUES (?, ?)",                (user_id, day),            )        except sqlite3.IntegrityError:            return None          # 主键撞了 —— 今天签过了        conn.execute(            """            INSERT INTO users (user_id, name, points) VALUES (?, ?, ?)            ON CONFLICT(user_id) DO UPDATE SET                points = points + excluded.points,                name   = excluded.name            """,            (user_id, name, reward),        )        row = conn.execute(            "SELECT points FROM users WHERE user_id = ?", (user_id,)        ).fetchone()        return row[0]def get_user(user_id: int):    with connect() as conn:        return conn.execute(            "SELECT name, points FROM users WHERE user_id = ?", (user_id,)        ).fetchone()def count_days(user_id: int) -> int:    with connect() as conn:        return conn.execute(            "SELECT COUNT(*) FROM checkins WHERE user_id = ?", (user_id,)        ).fetchone()[0]def top_users(limit: int = 10):    with connect() as conn:        return conn.execute(            "SELECT name, points FROM users "            "ORDER BY points DESC, user_id ASC LIMIT ?",            (limit,),        ).fetchall()

from datetime import datefrom aiogram import Routerfrom aiogram.filters import Commandfrom aiogram.types import Messageimport storageREWARD = 10router = Router()@router.message(Command("start"))async def cmd_start(message: Message) -> None:    await message.answer(        "签到机器人\n\n"        "/checkin  每天签到一次，拿 10 分\n"        "/me       看我的积分\n"        "/rank     看排行榜"    )@router.message(Command("checkin"))async def cmd_checkin(message: Message) -> None:    if message.from_user is None:        return    points = storage.check_in(        user_id=message.from_user.id,        name=message.from_user.full_name,        day=date.today().isoformat(),        reward=REWARD,    )    if points is None:        await message.answer("今天已经签过了，明天再来")        return    days = storage.count_days(message.from_user.id)    await message.answer(        f"签到成功 +{REWARD}\n累计 {days} 天，当前 {points} 分"    )@router.message(Command("me"))async def cmd_me(message: Message) -> None:    if message.from_user is None:        return    row = storage.get_user(message.from_user.id)    if row is None:        await message.answer("你还没签过到，发 /checkin 试试")        return    name, points = row    days = storage.count_days(message.from_user.id)    await message.answer(f"{name}\n积分 {points}\n签到 {days} 天")@router.message(Command("rank"))async def cmd_rank(message: Message) -> None:    rows = storage.top_users(10)    if not rows:        await message.answer("还没有人签到")        return    lines = ["积分榜", ""]    for i, (name, points) in enumerate(rows, start=1):        lines.append(f"{i}. {name} — {points}")    await message.answer("\n".join(lines))

import asyncioimport loggingimport osfrom aiogram import Bot, Dispatcherfrom aiogram.client.session.aiohttp import AiohttpSessionfrom dotenv import load_dotenvimport storagefrom handlers import routerload_dotenv()logging.basicConfig(level=logging.INFO)async def main() -> None:    token = os.getenv("BOT_TOKEN")    if not token:        raise SystemExit("没读到 BOT_TOKEN，检查 .env 文件")    storage.init_db()          # 表不存在就建，已存在就跳过    proxy = os.getenv("PROXY_URL")    session = AiohttpSession(proxy=proxy) if proxy else None    bot = Bot(token=token, session=session)    dp = Dispatcher()    dp.include_router(router)    me = await bot.get_me()    logging.info("启动成功：@%s", me.username)    await dp.start_polling(bot)if __name__ == "__main__":    asyncio.run(main())