RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
362 lines
16 KiB
Python
362 lines
16 KiB
Python
# 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 <on|off> QQ 侧私聊推送开关（默认开，作用于全部消息）
|
||
# .my tg-push <on|off> 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 <on|off> QQ 侧私聊推送开关（默认开）\n"
|
||
".my tg-push <on|off> 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 <QQ号>（不写则只列出）"
|
||
% ", ".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)
|
||
|
||
|