# identity_cmd.py — 统一身份指令：.my（兼容旧 .me / .bind / .unbind 别名） # 由 dispatch.py 前置路由调用。 # .my / .my help 帮助菜单 # .my account 查看身份信息 # .my account bind ... 绑定（状态查看 / QQ 侧发起 / TG 侧确认码） # .my account unbind [QQ号] 解绑（TG 侧任意 QQ；QQ 侧只能解绑自己） # .my qq-push QQ 侧私聊推送开关（默认开，作用于全部消息） # .my tg-push TG 侧私聊推送开关（默认开，作用于全部消息） import identity from channel import get_channel, send_message UNBOUND_TEXT = ( "你还没有绑定统一身份，暂时不能使用本 bot 的功能。\n" "绑定步骤：\n" "1. 在 Telegram 私聊我，发送 .my account 获取你的身份 id；\n" "2. 回到这里（QQ 私聊）发送 .my account bind <身份 id>；\n" "3. 在 Telegram 侧回复 .my account bind <确认码> 完成确认。" ) GROUP_BIND_TEXT = ( "该功能会保存你的个人配置，需要先绑定统一身份（请勿在群里操作）：\n" "1. Telegram 私聊我发送 .my account 获取身份 id；\n" "2. QQ 私聊我发送 .my account bind <身份 id>；\n" "3. 回 Telegram 侧用 .my account bind <确认码> 确认。\n" "群内其它功能（.setu / .pixiv / .jm / .live 等）无需绑定即可使用。" ) QQ_BIND_USAGE = ( "绑定步骤：\n" "1. 在 Telegram 私聊我，发送 .my account 获取你的身份 id；\n" "2. 在这里发送 .my account bind <身份 id>；\n" "3. 在 Telegram 侧回复 .my account bind <确认码> 完成确认。\n" "也可以先用 .my account bind status 查看绑定状态。" ) TG_BIND_USAGE = ( "用法：\n" "• .my account bind 查看绑定状态\n" "• .my account bind status 查看绑定状态\n" "• .my account bind <确认码> 确认一条 QQ 绑定请求（6 位数字）\n" "• .my account bind unbind [QQ号] 解绑（不写 QQ 号则列出已绑定的 QQ）\n" "绑定 QQ：让该 QQ 私聊我发送 .my account bind <你的身份 id>，再回到这里确认。" ) MY_HELP_TEXT = ( ".my 统一身份与个人设置\n" "━━━━━━━━━━━━\n" ".my 本菜单\n" ".my account 查看身份信息\n" ".my account bind 绑定 QQ（发起/确认/状态）\n" ".my account unbind 解绑 QQ\n" ".my qq-push QQ 侧私聊推送开关（默认开）\n" ".my tg-push TG 侧私聊推送开关（默认开）\n" "━━━━━━━━━━━━\n" "推送开关作用于你的全部消息（指令回复与主动通知），\n" "关闭某侧后该侧不再收到任何私聊消息。" ) STATUS_SUBCOMMANDS = ("status", "状态") UNBIND_SUBCOMMANDS = ("unbind", "解绑", "remove") async def _reply(channel, msg_type, target_id, content): # 身份/设置类指令的回复不跨平台镜像：设置指令在哪个平台发就回在哪个平台， # 避免 "关推送的确认语又弹到另一平台" 这类自我矛盾 await send_message(channel, msg_type, target_id, content, mirror=False) async def _notify(channel_name, target_id, content): channel = get_channel(channel_name) if channel is None or not target_id: return False # send_message returns False for a known delivery failure; never turn that # into a successful notification (notably the QQ binding request flow). return await send_message(channel, "private", target_id, content) is True def _pending_lines(info): lines = [] for item in info.get("pending") or []: lines.append(" QQ %s → 确认码 %s（剩余 %d 分钟）" % (item["qq"], item["token"], max(1, item["remain_seconds"] // 60))) return lines def _status_text(uid, qq=None): """绑定状态文案；qq 不为 None 时额外标注该 QQ 是否已绑定""" info = identity.summary(uid) if not info: return "身份信息读取失败，请联系管理员" lines = [ "统一身份绑定状态", "━━━━━━━━━━━━", "身份 id：%d（%s）" % (info["uid"], "管理员" if info["admin"] else "普通用户"), "TG 账号：%s" % (info["tg_raw"] or "未绑定"), "QQ 账号：%s" % (", ".join(str(x) for x in info["qq"]) if info["qq"] else "未绑定"), ] if qq is not None: bound = int(qq) in (info["qq"] or []) lines.append("本 QQ：%s" % ("已绑定" if bound else "未绑定")) pending = _pending_lines(info) if pending: lines.append("待确认：") lines.extend(pending) lines.append("━━━━━━━━━━━━") lines.append("绑定：QQ 私聊 .my account bind <身份 id> → TG 侧 .my account bind <确认码>") lines.append("解绑：.my account unbind [QQ号]") lines.append("推送：.my qq-push on|off / .my tg-push on|off") return "\n".join(lines) def _own_qq(channel, msg_type, target_id): """QQ 私聊里用户自己的 QQ 号（私聊时 target_id 即 QQ 号）；其它情况返回 None""" if channel.name == "qq" and msg_type == "private": try: return int(target_id) except (TypeError, ValueError): return None return None async def _do_unbind(channel, msg_type, target_id, uid, arg): """解绑：TG 侧可指定 QQ；QQ 侧只能解绑自己（必须私聊）""" info = identity.summary(uid) or {} qq_list = info.get("qq") or [] if channel.name == "qq": if msg_type != "private": await _reply(channel, msg_type, target_id, "解绑请在私聊进行：私聊我发送 .my account unbind") return own = _own_qq(channel, msg_type, target_id) if own is None: await _reply(channel, msg_type, target_id, "无法识别你的 QQ 号，请稍后重试") return if arg and str(arg) != str(own): await _reply(channel, msg_type, target_id, "QQ 侧只能解绑自己（本 QQ 为 %d）：如需解绑其它 QQ 请在 TG 侧操作" % own) return ok, message = identity.unbind_qq(uid, own) await _reply(channel, msg_type, target_id, message if ok else "解绑失败：%s" % message) return if not arg: if qq_list: await _reply(channel, msg_type, target_id, "已绑定的 QQ：%s\n用法：.my account unbind （不写则只列出）" % ", ".join(str(x) for x in qq_list)) else: await _reply(channel, msg_type, target_id, "你的身份还没有绑定任何 QQ") return ok, message = identity.unbind_qq(uid, arg) await _reply(channel, msg_type, target_id, message if ok else "解绑失败：%s" % message) if ok: await _notify("qq", arg, "你的 QQ 已与统一身份 %d 解绑，如需继续使用请重新绑定。" % uid) async def handle_me(channel, msg_type, target_id, uid): info = identity.summary(uid) if not info: await _reply(channel, msg_type, target_id, "身份信息读取失败，请联系管理员") return lines = [ "统一身份", "━━━━━━━━━━━━", "身份 id：%d" % info["uid"], "身份类型：%s" % ("管理员" if info["admin"] else "普通用户"), "TG 账号：%s" % (info["tg_raw"] or "未绑定"), "QQ 账号：%s" % (", ".join(str(x) for x in info["qq"]) if info["qq"] else "未绑定"), ] pending = _pending_lines(info) if pending: lines.append("待确认绑定：") lines.extend(pending) lines.append("━━━━━━━━━━━━") lines.append("绑定 QQ：QQ 私聊发 .my account bind %d，再回到这里发 .my account bind <确认码>" % info["uid"]) lines.append("推送：.my qq-push on|off / .my tg-push on|off") await _reply(channel, msg_type, target_id, "\n".join(lines)) async def handle_bind(channel, msg_type, target_id, uid, arg): """已绑定用户的 .bind：status / unbind /（TG 侧）确认码""" arg = (arg or "").strip() low = arg.lower() if low in STATUS_SUBCOMMANDS: await _reply(channel, msg_type, target_id, _status_text(uid, _own_qq(channel, msg_type, target_id))) return if low in UNBIND_SUBCOMMANDS: await _do_unbind(channel, msg_type, target_id, uid, "") return if channel.name != "tg": # QQ 侧已绑定：给出状态与解绑入口 info = identity.summary(uid) or {} pending = _pending_lines(info) text = "你已绑定身份 %d。\n" % uid if pending: text += "待确认的绑定请求：\n" + "\n".join(pending) + "\n请在 TG 侧回复 .bind <确认码>。\n" text += "查看状态：.my account bind status\n解绑本 QQ：.my account unbind\n为其它 QQ 绑定：让该 QQ 私聊我发送 .my account bind %d" % uid await _reply(channel, msg_type, target_id, text) return if not arg: info = identity.summary(uid) or {} pending = _pending_lines(info) if pending: await _reply(channel, msg_type, target_id, "待确认的 QQ 绑定请求：\n" + "\n".join(pending) + "\n回复 .my account bind <确认码> 确认。") else: await _reply(channel, msg_type, target_id, TG_BIND_USAGE) return token = low if token.isdigit() and len(token) == 6: ok, message, qq = identity.confirm_qq_bind(uid, token) if not ok: await _reply(channel, msg_type, target_id, "确认失败：%s" % message) return await _reply(channel, msg_type, target_id, "确认成功：QQ %d 已绑定到身份 %d，现在可以用该 QQ 使用全部功能。" % (qq, uid)) await _notify("qq", qq, "绑定成功：你的 QQ 已与统一身份 %d 关联。" % uid) return if token.isdigit(): await _reply(channel, msg_type, target_id, "这是身份 id 而不是确认码。绑定 QQ 请在 QQ 私聊发送 .my account bind %s，" "然后回到这里用 .my account bind <6 位确认码> 确认。" % token) return await _reply(channel, msg_type, target_id, TG_BIND_USAGE) async def handle_qq_bind(channel, msg_type, target_id, qq, arg): """QQ 侧未绑定用户的 .bind：status / unbind / <身份 id> 发起绑定""" arg = (arg or "").strip() low = arg.lower() if low in STATUS_SUBCOMMANDS: await _reply(channel, msg_type, target_id, "绑定状态\n━━━━━━━━━━━━\n本 QQ：%s\n尚未绑定任何统一身份。\n━━━━━━━━━━━━\n%s" % (qq, QQ_BIND_USAGE)) return if low in UNBIND_SUBCOMMANDS: await _reply(channel, msg_type, target_id, "本 QQ 尚未绑定任何统一身份，无需解绑。\n\n" + QQ_BIND_USAGE) return if msg_type != "private": await _reply(channel, msg_type, target_id, "绑定请在私聊进行（避免身份 id 泄露到群里）") return if not arg: await _reply(channel, msg_type, target_id, QQ_BIND_USAGE) return if not (arg.isdigit() and len(arg) >= 10): await _reply(channel, msg_type, target_id, "身份 id 应为 10 位数字（在 TG 私聊用 .me 查看），例如 .my account bind 9000000001\n\n" + QQ_BIND_USAGE) return uid = int(arg) # 先验身份与 TG 可达性，再创建 pending：避免告知失败但待确认请求已残留 info = identity.summary(uid) if not info: await _reply(channel, msg_type, target_id, "绑定失败：身份 id 不存在，请核对 TG 侧 .me 显示的 id") return if not info.get("tg_ns"): await _reply(channel, msg_type, target_id, "该身份没有可用的 TG 账号，无法发送确认请求") return ok, message, token = identity.request_qq_bind(qq, uid) if not ok: await _reply(channel, msg_type, target_id, "绑定失败：%s" % message) return text = ("【身份绑定请求】\n" "QQ %d 请求绑定到你的身份 %d。\n" "确认码：%s（10 分钟内有效）\n" "确认方式：在本对话回复 .my account bind %s\n" "如果这不是你本人的操作，请忽略本条消息。" % (qq, uid, token, token)) if not await _notify("tg", info["tg_ns"], text): # TG 通道不可用：pending 已创建（TG 恢复后仍可在有效期内确认）， # 文案如实告知，不谎称"已发送" await _reply(channel, msg_type, target_id, "TG 通道暂时不可用，确认请求未能送达。\n" "请稍后重发本指令（确认码 10 分钟内有效，重发会刷新有效期）。") return await _reply(channel, msg_type, target_id, "已向身份 %d 的 TG 账号发送确认请求（确认码 %s）。\n" "请在 Telegram 私聊中回复 .my account bind %s 完成绑定。" % (uid, token, token)) async def handle_unbind(channel, msg_type, target_id, uid, arg): await _do_unbind(channel, msg_type, target_id, uid, (arg or "").strip()) async def handle_my(channel, msg_type, target_id, user_id, uid, arg): """.my 统一入口。user_id=原始平台 id（未绑定 QQ 时为 QQ 号），uid=统一身份（未绑定为 None）。""" arg = (arg or "").strip() low = arg.lower() if not low or low in ("help", "-h", "帮助"): await _reply(channel, msg_type, target_id, MY_HELP_TEXT) return parts = arg.split(maxsplit=1) head = parts[0].lower() rest = parts[1].strip() if len(parts) > 1 else "" if head == "account": if not rest: if uid is None: await _reply(channel, msg_type, target_id, "尚未绑定统一身份。\n" + QQ_BIND_USAGE) return await handle_me(channel, msg_type, target_id, uid) return sub = rest.split(maxsplit=1) sub_head = sub[0].lower() sub_rest = sub[1].strip() if len(sub) > 1 else "" if sub_head == "bind": if uid is None: await handle_qq_bind(channel, msg_type, target_id, user_id, sub_rest) else: await handle_bind(channel, msg_type, target_id, uid, sub_rest) return if sub_head in UNBIND_SUBCOMMANDS: if uid is None: await _reply(channel, msg_type, target_id, "本 QQ 尚未绑定任何统一身份，无需解绑。\n\n" + QQ_BIND_USAGE) return await _do_unbind(channel, msg_type, target_id, uid, sub_rest) return await _reply(channel, msg_type, target_id, "未知子命令：%s\n\n" % sub_head + MY_HELP_TEXT) return if head in ("qq-push", "tg-push"): platform = head.split("-")[0] if uid is None: await _reply(channel, msg_type, target_id, "需要先绑定统一身份才能设置推送开关：\n" + QQ_BIND_USAGE) return flag = rest.lower() if flag not in ("on", "off"): cur = (identity.summary(uid) or {}).get("push") or {} state = "开" if cur.get(platform, True) else "关" await _reply(channel, msg_type, target_id, "用法：.my %s on|off（当前：%s）" % (head, state)) return ok, message = identity.set_push(uid, platform, flag == "on") if not ok: await _reply(channel, msg_type, target_id, "设置失败：%s" % message) return effect = "该侧将不再收到任何私聊消息" if flag == "off" else "该侧将收到你的全部私聊消息" await _reply(channel, msg_type, target_id, "%s 侧私聊推送已%s。%s。" % (platform.upper(), message, effect)) return await _reply(channel, msg_type, target_id, MY_HELP_TEXT)