功能:通过 HTTP API 实现 Danding_QqPush 插件,用于 QQ 群通知

- 增加了通过外部 HTTP API 向 QQ 群组发送消息的核心功能。
- 实现了对长文本消息的图片渲染,以避免被认定为垃圾信息。
- 支持在消息中提及特定的 QQ 用户。
- 创建了用于 API 令牌和图片渲染设置的配置选项。
- 开发了一个测试脚本以验证 API 功能。
- 对现有代码进行了重构,以提高组织性和可维护性。
This commit is contained in:
2026-01-20 21:19:05 +08:00
parent 4a944316fe
commit dedc872f1b
14 changed files with 1065 additions and 48 deletions

View File

@@ -74,55 +74,70 @@ async def query_qq_binding(qq: str) -> Tuple[bool, Optional[str], Optional[str]]
logger.error(f"查询QQ绑定状态异常: {str(e)}")
return False, None, None
async def add_user_viptime(username: str, time_class: str = "Day") -> Tuple[bool, str]:
async def add_user_viptime(username: str, time_class: str = "Day", count: int = 1) -> Tuple[bool, str]:
"""
为用户添加VIP时间
Args:
username: 蛋定用户名
time_class: 时间类型 (Hour/Day/Week/Month/Season/Year)
count: 添加次数默认为1
Returns:
Tuple[是否成功, 响应消息]
"""
try:
url = f"{DD_API_HOST}bot_add_user_viptime"
data = {
"user": BOT_USER_ID,
"token": BOT_TOKEN,
"username": username,
"classes": time_class
}
response = requests.post(url=url, json=data)
logger.debug(f"添加VIP时间响应: {response}")
# 如果count大于1需要多次调用API
success_count = 0
last_message = ""
if response.status_code != 200:
error_msg = f"添加VIP时间失败状态码: {response.status_code}"
logger.error(error_msg)
return False, error_msg
for i in range(count):
data = {
"user": BOT_USER_ID,
"token": BOT_TOKEN,
"username": username,
"classes": time_class
}
result = response.json()
logger.debug(f"添加VIP时间结果: {result}")
response = requests.post(url=url, json=data)
logger.debug(f"添加VIP时间响应({i+1}/{count}): {response}")
if response.status_code != 200:
error_msg = f"添加VIP时间失败({i+1}/{count}),状态码: {response.status_code}"
logger.error(error_msg)
continue
result = response.json()
logger.debug(f"添加VIP时间结果({i+1}/{count}): {result}")
if result.get("code") == 200:
success_count += 1
last_message = result.get("msg", "添加VIP时间成功")
else:
error_msg = result.get("msg", "添加VIP时间失败")
logger.error(f"添加VIP时间失败({i+1}/{count}): {error_msg}")
if result.get("code") == 200:
return True, result.get("msg", "添加VIP时间成功")
if success_count == count:
return True, f"成功添加{count}{time_class}时长。{last_message}"
elif success_count > 0:
return False, f"仅成功添加{success_count}/{count}{time_class}时长。{last_message}"
else:
error_msg = result.get("msg", "添加VIP时间失败")
logger.error(f"添加VIP时间失败: {error_msg}")
return False, error_msg
return False, f"添加{count}{time_class}时长全部失败"
except Exception as e:
error_msg = f"添加VIP时间异常: {str(e)}"
logger.error(error_msg)
return False, error_msg
async def process_ssr_sp_reward(user_id: str) -> Tuple[bool, str]:
async def process_ssr_sp_reward(user_id: str, count: int = 1) -> Tuple[bool, str]:
"""
处理SSR/SP奖励发放
Args:
user_id: QQ用户ID
count: 奖励数量默认为1
Returns:
Tuple[是否自动发放成功, 消息内容]
@@ -132,26 +147,42 @@ async def process_ssr_sp_reward(user_id: str) -> Tuple[bool, str]:
if not is_bound:
# 用户未绑定,返回提示信息
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神🎉\n"
f"获得奖励:蛋定助手天卡一张\n"
f"奖励请联系管理员或前往蛋定云服务中绑定QQ号即可体验自动加时")
if count == 1:
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神🎉\n"
f"奖励:蛋定助手天卡一张\n"
f"获取奖励请联系管理员或前往蛋定云服务中绑定QQ号即可体验自动加时")
else:
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神🎉\n"
f"获得奖励:蛋定助手天卡{count}\n"
f"获取奖励请联系管理员或前往蛋定云服务中绑定QQ号即可体验自动加时")
return False, msg
else:
# 用户已绑定,自动加时
success, message = await add_user_viptime(username, "Day")
success, message = await add_user_viptime(username, "Day", count)
if success:
masked_username = mask_username(username)
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神🎉\n"
f"🎁已自动为您的蛋定账号({masked_username})添加天卡时长!\n"
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
if count == 1:
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神🎉\n"
f"🎁已自动为您的蛋定账号({masked_username})添加天卡时长!\n"
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
else:
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神🎉\n"
f"🎁已自动为您的蛋定账号({masked_username})添加{count}天卡时长!\n"
f"📅到期时间: {message.split('到期时间为:')[-1] if '到期时间为:' in message else '请查看您的账号详情'}")
return True, msg
else:
# 自动加时失败,返回错误信息和手动领取提示
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神🎉\n"
f"获得奖励:蛋定助手天卡一张\n"
f"⚠️自动加时失败: {message}\n"
f"请联系管理员手动领取奖励!")
if count == 1:
msg = (f"🎉恭喜您抽中了SSR/SP稀有度式神🎉\n"
f"获得奖励:蛋定助手天卡一张\n"
f"⚠️自动加时失败: {message}\n"
f"请联系管理员手动领取奖励!")
else:
msg = (f"🎉恭喜您抽中了{count}张SSR/SP稀有度式神🎉\n"
f"获得奖励:蛋定助手天卡{count}\n"
f"⚠️自动加时失败: {message}\n"
f"请联系管理员手动领取奖励!")
return False, msg
async def process_achievement_reward(user_id: str, achievement_id: str) -> Tuple[bool, str]: