最近手痒用AI搓了个QQ机器人，主要用来在群里发发早安、查个天气、宿舍电费等，主要是学习使用。

QQ机器人官方开发文档在 https://bot.q.qq.com/wiki/develop/api-v2/ ，里面有python、go、nodejs的SDK，附带例程还是非常清晰易懂的。QQ机器人个人开发者只能在自己是群主的群里使用，官方限制比较大，不过我只是做给寝室群用的问题不大。
主程序如下，

# -*- coding: utf-8 -*-
import os
import re
import sys
from collections import defaultdict
from datetime import date

import botpy
from botpy import logging
from botpy.ext.command_util import Commands
from botpy.ext.cog_yaml import read
from botpy.errors import ServerError
from botpy.message import GroupMessage, C2CMessage

# 将项目根目录加入 sys.path，以便导入上级模块
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from main import ElecQueryError, query_electricity, format_result
import cookie_store

# 同目录模块

from panel_api import create_panel, get_panel, list_panels, update_panel
from panel_store import (
    DEFAULT_PANEL_ITEMS,
    SUPPORTED_SCOPES,
    get_panel as get_local_panel,
    set_panel as set_local_panel,
    compute_panel_store_hash,
)
from data_store import (
    get_elec_cache,
    set_elec_cache,
    add_elec_history,
    get_elec_history,
    get_elec_history_by_month,
    get_room_binding,
    set_room_binding,
    get_panel_hash,
    set_panel_hash,
    remove_panel_hash,
)


_log = logging.get_logger()

# ========== 消息回复工具 ==========

# msg_seq 计数器：同一 msg_id 的回复序号，避免 QQ API 去重错误
_msg_seq_counter: dict[str, int] = defaultdict(int)


async def safe_reply(message, **kwargs) -> bool:
    msg_id = message.id
    _msg_seq_counter[msg_id] += 1
    msg_seq = _msg_seq_counter[msg_id]

    try:
        if isinstance(message, GroupMessage):
            await message._api.post_group_message(
                group_openid=str(message.group_openid),
                msg_id=msg_id,
                msg_seq=msg_seq,
                **kwargs,
            )
        elif isinstance(message, C2CMessage):
            await message._api.post_c2c_message(
                openid=message.author.user_openid,
                msg_id=msg_id,
                msg_seq=str(msg_seq),
                **kwargs,
            )
        else:
            await message.reply(**kwargs)
        return True
    except ServerError as e:
        err_msg = str(e)
        if "去重" in err_msg or "msgseq" in err_msg.lower() or "40054005" in err_msg:
            _log.warning(
                f"[回复] 消息去重跳过: msg_id={msg_id}, msg_seq={msg_seq}, error={err_msg}"
            )
            return False
        raise
    finally:
        # 清理过旧的计数器，防止内存泄漏（保留最近 1000 条）
        if len(_msg_seq_counter) > 1000:
            # 简单策略：清空整个计数器，因为旧 msg_id 不会再被回复
            _msg_seq_counter.clear()


# 读取配置
config_path = os.path.join(os.path.dirname(__file__), "config.yaml")
bot_config = read(config_path)

# 默认寝室号（私聊场景兜底）
DEFAULT_ROOM = "xxxxx"

# 未绑定寝室号时的提示
ROOM_NOT_BOUND_TIP = (
    "本群尚未绑定寝室号，请先使用「设置寝室 <寝室号>」绑定\n例: 设置寝室 xxxxx"
)




# ========== 群寝室号获取 ==========


def _get_group_room(message) -> str | None:
    """获取群绑定的寝室号，未绑定返回 None"""
    if not isinstance(message, GroupMessage):
        return None
    return get_room_binding(str(message.group_openid))


def _get_room_or_reply(message) -> str | None:
    if isinstance(message, GroupMessage):
        room = get_room_binding(str(message.group_openid))
        if not room:
            import asyncio

            asyncio.get_event_loop().create_task(
                safe_reply(message, content=ROOM_NOT_BOUND_TIP)
            )
            return None
        return room
    # 私聊使用默认寝室号
    return DEFAULT_ROOM


# ========== 面板管理 ==========


async def ensure_panels(client: botpy.Client) -> None:
    http = client.api._http
    current_hash = compute_panel_store_hash()

    for scope in SUPPORTED_SCOPES:
        local = get_local_panel(scope)
        panel_id = local.get("panel_id") if local else None
        stored_hash = get_panel_hash(scope)

        # 本地缓存存在
        if local and panel_id:
            if current_hash and current_hash != stored_hash:
                # hash 不一致 → 以本地为准更新远程
                _log.info(
                    f"[面板] {scope} 本地缓存 hash 变化 (stored={stored_hash}, current={current_hash})，更新远程面板"
                )
                try:
                    await update_panel(http, panel_id=panel_id, items=local.get("items", []))
                    set_panel_hash(scope, current_hash)
                    _log.info(f"[面板] {scope} 远程面板已同步，hash 已更新 (id={panel_id})")
                except Exception as e:
                    _log.error(f"[面板] {scope} 远程面板更新失败: {e}")
            else:
                _log.info(f"[面板] {scope} 本地缓存 hash 一致，跳过 (id={panel_id})")
        else:
            # 本地缓存不存在 → 删除数据库 hash 记录
            if stored_hash:
                remove_panel_hash(scope)
                _log.info(f"[面板] {scope} 本地缓存不存在，已清除数据库 hash 记录")


async def refresh_panel_for_scope(http, scope: str) -> str:
    local = get_local_panel(scope)
    panel_id = local.get("panel_id") if local else None
    current_hash = compute_panel_store_hash()
    stored_hash = get_panel_hash(scope)

    # 本地有缓存
    if local and panel_id:
        if current_hash and current_hash != stored_hash:
            # hash 不一致 → 以本地为准更新远程
            try:
                await update_panel(http, panel_id=panel_id, items=local.get("items", []))
                set_panel_hash(scope, current_hash)
                return f"✅ {scope}: 已将本地缓存同步到远程面板 (id={panel_id})"
            except Exception as e:
                _log.warning(f"[刷新菜单] {scope} 本地数据更新失败: {e}")
                return f"❌ {scope}: 本地缓存同步失败 - {e}"
        else:
            return f"⏭️ {scope}: 本地缓存无变化，无需更新 (id={panel_id})"

    # 本地无缓存 → 从远程拉取
    try:
        panel_list = await list_panels(http, scope=scope)
        if isinstance(panel_list, dict):
            panels = panel_list.get("data", panel_list.get("panels", []))
        elif isinstance(panel_list, list):
            panels = panel_list
        else:
            panels = []

        if not panels:
            return f"⚠️ {scope}: 远程无面板，请重启机器人自动创建"

        remote_id = (
            panels[0].get("panel_id", panels[0].get("id", ""))
            if isinstance(panels[0], dict)
            else ""
        )
        if not remote_id:
            return f"❌ {scope}: 远程面板数据异常"

        detail = await get_panel(http, remote_id)
        remote_items = (
            detail.get("panel", {}).get("items", []) if isinstance(detail, dict) else []
        )
        set_local_panel(
            scope, {"panel_id": remote_id, "scope": scope, "items": remote_items}
        )
        # 重新计算 hash 并存入数据库
        new_hash = compute_panel_store_hash()
        if new_hash:
            set_panel_hash(scope, new_hash)
        return f"✅ {scope}: 已从远程拉取面板到本地缓存 (id={remote_id})"
    except Exception as e:
        _log.error(f"[刷新菜单] {scope} 远程拉取失败: {e}")
        return f"❌ {scope}: 远程拉取失败 - {e}"


# ========== Cookie 等待状态 ==========
# 查电费失败后，下一条消息自动作为 Cookie 缓存
_waiting_for_cookie: bool = False


# ========== 电量查询（带缓存）==========


def _parse_elec_remain(result: dict) -> float | None:
    """从查询结果提取剩余电量数值"""
    remain_str = result.get("elecRemain", "")
    if not remain_str:
        return None
    match = re.search(r"[\d.]+", str(remain_str))
    return float(match.group()) if match else None


# 电量查询结果必需字段（缺失则视为缓存无效）
_CACHE_REQUIRED_FIELDS = ("returncode", "elecroominfo", "elecRemain")


def _is_valid_cache(data: dict) -> bool:
    return all(f in data for f in _CACHE_REQUIRED_FIELDS)


async def _query_with_cache(room_no: str, cookie: str = "", force_refresh: bool = False) -> tuple[str, dict | None, bool]:
    # 检查缓存（需包含必需字段才算有效），有外部 cookie 或强制刷新时跳过缓存
    if not cookie and not force_refresh:
        cache = get_elec_cache(room_no)
        if cache and _is_valid_cache(cache["data"]):
            _log.info(f"[查询] 使用缓存: {room_no}")
            return format_result(cache["data"]), cache["data"], True

    # 调用接口
    try:
        result = query_electricity(room_no, cookie=cookie)
    except ElecQueryError as e:
        err_msg = str(e)
        if "Cookie 已失效" in err_msg:
            return (
                f"查询失败: Cookie 已失效\n"
                f"请在浏览器登录 pay.neu.edu.cn 后，将 JSESSIONID 发送给机器人\n"
                f"格式: 设置Cookie <JSESSIONID值>",
                None,
                False,
            )
        return f"查询失败: {e}", None, False
    except Exception as e:
        _log.error(f"电量查询异常: {e}")
        return "查询时发生未知错误，请稍后重试", None, False

    # 缓存 + 记录历史
    if result.get("returncode") == "SUCCESS":
        set_elec_cache(room_no, result)
        elec_remain = _parse_elec_remain(result)
        if elec_remain is not None:
            add_elec_history(room_no, elec_remain)

    return format_result(result), result, False


# ========== 命令处理函数 ==========


@Commands("帮助")
async def cmd_help(api, message, params=None):
    """显示帮助信息，列出所有可用命令及用法"""
    help_lines = [
        "📖 NEU 寝室电量查询机器人 - 帮助",
        "",
        "🔹 设置寝室 <寝室号>",
        "   绑定群与寝室号（仅群聊）",
        "   例: 设置寝室 xxxxx",
        "",
        "🔹 查电费 [寝室号] [刷新]",
        "   查询寝室剩余电量",
        "   群聊默认使用绑定寝室号，加\"刷新\"强制从服务器获取",
        "   例: 查电费 / 查电费 1120294 / 查电费 刷新",
        "",
        "🔹 看周报",
        "   查看最近7天电量变化趋势（仅群聊）",
        "",
        "🔹 看月报 [月份]",
        "   查看月度电量变化趋势（仅群聊，默认当月）",
        "   例: 看月报 / 看月报 9 / 看月报 2026-09",
        "",
        "🔹 刷新菜单",
        "   管理员专用，同步指令面板",
        "",
        "💡 Cookie 失效时，发送 JSESSIONID 即可自动更新",
    ]
    await safe_reply(message, content="\n".join(help_lines))
    return True


@Commands("设置寝室")
async def cmd_set_room(api, message, params=None):
    if not isinstance(message, GroupMessage):
        await safe_reply(message, content="设置寝室仅支持群聊操作")
        return True

    if not params or not params.strip():
        await safe_reply(
            message, content="用法: 设置寝室 <寝室号>\n例: 设置寝室 1120294"
        )
        return True

    room_no = params.strip().split()[0]
    group_openid = str(message.group_openid)
    member_openid = message.author.member_openid if message.author else ""

    set_room_binding(group_openid, room_no, member_openid)
    await safe_reply(
        message,
        content=f"已绑定寝室号: {room_no}\n后续查电费、周报、月报均使用此寝室号",
    )
    return True


@Commands("查电费")
async def cmd_elec_query(api, message, params=None):
    global _waiting_for_cookie
    force_refresh = False

    # 解析参数：提取寝室号和"刷新"关键字
    if params and params.strip():
        parts = params.strip().split()
        # 检查是否包含"刷新"关键字
        if "刷新" in parts:
            force_refresh = True
            parts = [p for p in parts if p != "刷新"]
        room_no = parts[0] if parts else None
    else:
        room_no = None

    # 群聊：优先用参数中的寝室号，否则用绑定
    if isinstance(message, GroupMessage):
        if not room_no:
            room_no = get_room_binding(str(message.group_openid))
            if not room_no:
                await safe_reply(message, content=ROOM_NOT_BOUND_TIP)
                return True
    else:
        # 私聊：需要参数或使用默认
        if not room_no:
            room_no = DEFAULT_ROOM

    result_text, _, from_cache = await _query_with_cache(room_no, force_refresh=force_refresh)
    source_tag = "📡 数据来源：本机缓存" if from_cache else "📡 数据来源：服务器查询"
    if force_refresh and not from_cache:
        source_tag = "📡 数据来源：服务器查询（强制刷新）"
    await safe_reply(message, content=f"{result_text}\n{source_tag}")

    # Cookie 失效时，设置等待标志，下一条消息作为 Cookie
    if "Cookie 已失效" in result_text:
        _waiting_for_cookie = True
        _log.info("[Cookie] 查询失败，等待用户发送新 Cookie")

    return True





@Commands("刷新菜单")
async def cmd_refresh_panel(api, message, params=None):
    """处理刷新菜单命令（管理员专用）"""
    http = api._http
    results = []
    for scope in SUPPORTED_SCOPES:
        result = await refresh_panel_for_scope(http, scope)
        results.append(result)
    await safe_reply(message, content="\n".join(results))
    return True


@Commands("看周报")
async def cmd_weekly_report(api, message, params=None):
    if not isinstance(message, GroupMessage):
        await safe_reply(message, content="看周报仅支持群聊查看")
        return True

    room_no = get_room_binding(str(message.group_openid))
    if not room_no:
        await safe_reply(message, content=ROOM_NOT_BOUND_TIP)
        return True

    history = get_elec_history(room_no, days=7)
    if not history:
        await safe_reply(
            message, content="暂无电量历史数据，请先使用「查电费」积累数据"
        )
        return True

    # 计算统计
    stats_lines = []
    if len(history) >= 2:
        first = history[0].get("elec_remain", 0)
        last = history[-1].get("elec_remain", 0)
        days = len(history) - 1
        daily_avg = (first - last) / days if days > 0 else 0
        stats_lines.append(f"日均消耗 **{daily_avg:.1f}** 度/天")
        if daily_avg > 0 and last > 0:
            est_days = int(last / daily_avg)
            stats_lines.append(f"预计可用约 **{est_days}** 天")

    await safe_reply(
        message,
        msg_type=2,
        markdown={"content": _build_markdown_report("电费周报", room_no, history, stats_lines)},
    )

    return True


@Commands("看月报")
async def cmd_monthly_report(api, message, params=None):
    if not isinstance(message, GroupMessage):
        await safe_reply(message, content="看月报仅支持群聊查看")
        return True

    room_no = get_room_binding(str(message.group_openid))
    if not room_no:
        await safe_reply(message, content=ROOM_NOT_BOUND_TIP)
        return True

    # 解析月份参数
    year, month = None, None
    if params and params.strip():
        param = params.strip()
        if re.match(r"^\d{4}-\d{1,2}$", param):
            # 格式: 2026-09
            parts = param.split("-")
            year, month = int(parts[0]), int(parts[1])
        elif re.match(r"^\d{1,2}$", param):
            # 格式: 9 或 09
            month = int(param)
        else:
            await safe_reply(
                message,
                content="月份格式错误，请使用: 看月报 / 看月报 9 / 看月报 2026-09",
            )
            return True

    history = get_elec_history_by_month(room_no, year=year, month=month)
    if not history:
        target = (
            f"{year}-{month:02d}"
            if year and month
            else f"{date.today().year}-{date.today().month:02d}"
        )
        await safe_reply(
            message, content=f"{target} 暂无电量历史数据，请先使用「查电费」积累数据"
        )
        return True

    # 标题显示目标月份
    target_year = year or date.today().year
    target_month = month or date.today().month
    title_month = f"{target_year}年{target_month}月"

    # 计算统计
    stats_lines = []
    if len(history) >= 2:
        first = history[0].get("elec_remain", 0)
        last = history[-1].get("elec_remain", 0)
        days = len(history) - 1
        daily_avg = (first - last) / days if days > 0 else 0
        monthly_total = first - last
        stats_lines.append(f"本月消耗 **{monthly_total:.1f}** 度")
        stats_lines.append(f"日均消耗 **{daily_avg:.1f}** 度/天")
        if daily_avg > 0 and last > 0:
            est_days = int(last / daily_avg)
            stats_lines.append(f"预计可用约 **{est_days}** 天")

    await safe_reply(
        message,
        msg_type=2,
        markdown={"content": _build_markdown_report(
            f"电费月报 · {title_month}", room_no, history, stats_lines
        )},
    )

    return True


def _build_markdown_report(
    title: str, room_no: str, history: list, stats_lines: list
) -> str:
    lines = [f"## {title} · 寝室 {room_no}"]
    lines.append("")
    lines.append("***")
    lines.append("")

    for entry in history:
        date_str = entry.get("date", "未知")
        # 缩短日期格式: 2026-09-01 → 09-01
        short_date = date_str[5:] if len(date_str) == 10 else date_str
        elec = entry.get("elec_remain", "未知")
        if isinstance(elec, (int, float)):
            lines.append(f"**{short_date}**　{elec:.1f} 度")
        else:
            lines.append(f"**{short_date}**　{elec} 度")

    if stats_lines:
        lines.append("")
        lines.append("***")
        lines.append("")
        for s in stats_lines:
            lines.append(f"> {s}")

    return "\n".join(lines)


# 命令处理列表
COMMAND_HANDLERS = [
    cmd_help,
    cmd_set_room,
    cmd_elec_query,
    cmd_refresh_panel,
    cmd_weekly_report,
    cmd_monthly_report,
]


class ElecQueryClient(botpy.Client):
    async def on_ready(self):
        robot_name = self.robot.name if self.robot else "未知"
        _log.info(f"机器人「{robot_name}」已上线！")
        # 启动时确保面板存在
        await ensure_panels(self)

    # ========== 群聊 @消息 ==========

    # 消息去重：记录已处理的 msg_id，防止重复处理
    _processed_msg_ids: set[str] = set()

    async def on_group_at_message_create(self, message: GroupMessage):
        """处理群聊中 @机器人 的消息"""
        global _waiting_for_cookie
        content = message.content.strip()
        msg_id = message.id
        _log.info(
            f"[群聊] 收到消息: {content} (group_openid={message.group_openid}, msg_id={msg_id})"
        )

        # 消息去重检查
        if msg_id in self._processed_msg_ids:
            _log.warning(
                f"[群聊] 检测到重复消息，跳过处理: msg_id={msg_id}, content={content}"
            )
            return
        self._processed_msg_ids.add(msg_id)

        # 防止内存泄漏：保留最近 2000 条 msg_id
        if len(self._processed_msg_ids) > 2000:
            self._processed_msg_ids.clear()

        # 查电费失败后，下一条消息自动作为 Cookie 缓存
        if _waiting_for_cookie:
            _waiting_for_cookie = False
            jsessionid = content.strip()
            _log.info(f"[Cookie] 收到用户提供的 Cookie: {jsessionid[:8]}...")
            if re.match(r"^[A-Za-z0-9]{16,}$", jsessionid):
                cookie_store.save_cookie(jsessionid)
                await safe_reply(message, content=f"Cookie 已更新: {jsessionid[:8]}...\n正在验证...")
                # 用新 Cookie 重新查询验证
                room_no = get_room_binding(str(message.group_openid))
                if room_no:
                    result_text, result, from_cache = await _query_with_cache(room_no, cookie=jsessionid)
                    source_tag = "📡 数据来源：本机缓存" if from_cache else "📡 数据来源：服务器查询"
                    if result and result.get("returncode") == "SUCCESS":
                        await safe_reply(message, content=f"Cookie 验证成功！\n{result_text}\n{source_tag}")
                    else:
                        await safe_reply(message, content=f"Cookie 已保存，但验证未成功:\n{result_text}")
                else:
                    await safe_reply(message, content="Cookie 已保存，发送「查电费」验证是否有效")
            else:
                await safe_reply(
                    message,
                    content=f"Cookie 格式不正确: {jsessionid[:8]}...\n"
                    "JSESSIONID 通常为32位十六进制字符串，请重新发送",
                )
                # 格式不对，继续等待
                _waiting_for_cookie = True
            return

        for handler in COMMAND_HANDLERS:
            if await handler(api=self.api, message=message):
                return

    # ========== 私聊消息 ==========

    async def on_c2c_message_create(self, message: C2CMessage):
        """处理私聊消息"""
        content = message.content.strip()
        msg_id = message.id
        _log.info(f"[私聊] 收到消息: {content} (msg_id={msg_id})")

        # 消息去重检查
        if msg_id in self._processed_msg_ids:
            _log.warning(
                f"[私聊] 检测到重复消息，跳过处理: msg_id={msg_id}, content={content}"
            )
            return
        self._processed_msg_ids.add(msg_id)

        for handler in COMMAND_HANDLERS:
            if await handler(api=self.api, message=message):
                return


if __name__ == "__main__":
    intents = botpy.Intents(public_messages=True)
    client = ElecQueryClient(intents=intents)
    client.run(appid=bot_config["appid"], secret=bot_config["secret"])

一开始挂在自己电脑上，结果出门一关机，Bot就失联，而且24小时开机实在太费电。去买个商用云服务器吧，按量付费或者包月，对于这种纯粹自娱自乐、随时可能吃灰的个人项目来说，又觉得有点亏。

于是开始在网上找免费的羊毛。转了一圈，最后决定试试阿贝云(https://www.abeiyun.com)，服务器配置是1核1G，对于学习来说够用。

过程没啥好说的，常规操作。在控制台开了个Ubuntu，拿到root密码后直接用SSH连上去。装了Python环境，配好依赖，把Bot脚本扔进去，用 systemd 开机自启，断开ssh连接也能继续跑。

机器人负载如下：top

服务器挂个轻量级的Bot完全够用，消息收发也没遇到什么延迟。相见恨晚啊

开通很简单

折腾完了，继续去改Bot的bug了。