WPSBot/.tasks/2025-11-03_1_user-webhook-url.md

背景

文件名：2025-11-03_1_user-webhook-url.md 创建于：2025-11-03_09:38:30 创建者：admin 主分支：main 任务分支：task/user-webhook-url_2025-11-03_1 Yolo模式：Off

任务描述

在WPS Bot Game项目中添加用户专属webhook URL功能，允许每个用户注册自己的个人webhook URL作为私聊途径。

核心需求

1. 用户可以通过 .register url <url> 指令注册个人webhook URL
2. 私聊消息发送功能将被封装为API接口，供其他系统调用
3. 提供检测用户是否具有个人URL的接口，用于系统运行时确保参与用户都能被私聊
4. 服务器启动时使用的webhook URL称为主URL，私聊用的URL称为个人URL

术语定义

• 主URL: 服务器启动时使用的webhook URL，用于群聊消息发送
• 个人URL: 用户注册的专属webhook URL，用于私聊消息发送

功能要求

1. 注册功能: 支持 .register url <url> 指令注册/更新个人URL
2. 私聊接口: 封装私聊消息发送功能为API接口（暂不对用户开放命令）
3. 检测接口: 提供单个和批量检测用户是否有个人URL的接口
4. 数据库支持: 在users表中添加webhook_url字段

项目概览

项目结构

WPSBotGame/
├── app.py                    # FastAPI主应用
├── config.py                 # 配置管理
├── core/
│   ├── database.py          # SQLite数据库操作
│   ├── middleware.py        # 中间件
│   └── models.py            # 数据模型
├── routers/
│   ├── callback.py          # Callback路由处理
│   ├── health.py            # 健康检查
│   └── private.py           # 私聊相关API（新增）
├── games/                   # 游戏模块
│   └── ...                 # 各种游戏
└── utils/
    ├── parser.py           # 指令解析
    └── message.py          # 消息发送

分析

当前状态

1. users 表已有基础字段：user_id, username, created_at, last_active
2. routers/callback.py 中已有 .register 命令处理名称注册
3. utils/message.py 中的 MessageSender 类使用全局webhook URL发送消息
4. 数据库已支持动态添加列（_add_column_if_not_exists方法）
5. init_tables() 方法在表创建后会进行兼容性检查，使用 _add_column_if_not_exists 安全添加新列

关键技术点

1. 数据库层:

   • 在init_tables()中使用_add_column_if_not_exists添加webhook_url字段（TEXT类型，可为NULL）
   • 确保兼容性：如果表已存在且没有该列，会自动添加
   • 添加set_user_webhook_url(user_id, webhook_url)方法
   • 添加get_user_webhook_url(user_id)方法
   • 添加has_webhook_url(user_id)方法
   • 添加check_users_webhook_urls(user_ids)批量检查方法

2. 注册命令扩展:

   • 修改handle_register_command支持.register url <url>子命令
   • 保留原有的.register <name>功能
   • URL验证（基本格式检查）

3. 私聊消息发送:

   • 封装私聊消息发送功能到utils/message.py
   • 创建send_private_message(user_id, content, msg_type='text')函数
   • 如果用户有个人URL则使用个人URL，否则返回错误

4. API接口:

   • 创建routers/private.py路由文件
   • POST /api/private/send - 发送私聊消息
   • GET /api/private/check/{user_id} - 检查单个用户是否有个人URL
   • POST /api/private/check-batch - 批量检查多个用户

提议的解决方案

方案概述

1. 数据库扩展: 在users表添加webhook_url字段，并实现相关CRUD方法
2. 注册命令扩展: 扩展.register命令支持url子命令
3. 私聊功能封装: 创建私聊消息发送工具函数
4. API接口: 创建私聊相关的RESTful API接口

设计决策

• 个人URL存储在users表中，与用户信息关联
• 私聊功能暂不提供用户命令，仅作为API接口供系统调用
• URL验证采用基本格式检查（http/https开头）
• 批量检查接口支持传入用户ID列表，返回每个用户的URL状态

当前执行步骤："3. 执行阶段完成"

实施清单：

1. 在core/database.py的init_tables()方法末尾添加webhook_url字段兼容性检查
2. 在core/database.py中添加set_user_webhook_url方法
3. 在core/database.py中添加get_user_webhook_url方法
4. 在core/database.py中添加has_webhook_url方法
5. 在core/database.py中添加check_users_webhook_urls方法
6. 在core/models.py文件末尾添加PrivateMessageRequest模型
7. 在core/models.py中添加CheckBatchRequest模型
8. 在core/models.py中添加CheckBatchResponse模型
9. 在core/models.py的导入中添加List类型
10. 修改routers/callback.py的handle_register_command函数支持url子命令
11. 在utils/message.py文件末尾添加send_private_message函数
12. 创建新文件routers/private.py，包含所有私聊相关API接口
13. 在app.py中导入private路由模块
14. 在app.py中注册private路由

详细实施计划

文件1: core/database.py

修改点1: 在init_tables()方法中添加webhook_url字段兼容性检查

位置: 在init_tables()方法的末尾，第324行logger.info("数据库表初始化完成")之前

修改内容:

# 兼容性检查：为users表添加webhook_url字段
self._add_column_if_not_exists('users', 'webhook_url', 'TEXT')

修改点2: 添加set_user_webhook_url方法

位置: 在# ===== 用户相关操作 =====部分，update_user_name方法之后（约第414行之后）

方法签名:

def set_user_webhook_url(self, user_id: int, webhook_url: str) -> bool:
    """设置用户webhook URL
    
    Args:
        user_id: 用户ID
        webhook_url: Webhook URL
        
    Returns:
        是否成功
    """

实现逻辑:

• 使用try-except包装
• 确保用户存在（调用get_or_create_user）
• UPDATE users SET webhook_url = ? WHERE user_id = ?
• 记录成功/失败日志
• 返回True/False，异常时返回False

修改点3: 添加get_user_webhook_url方法

位置: 紧接set_user_webhook_url方法之后

方法签名:

def get_user_webhook_url(self, user_id: int) -> Optional[str]:
    """获取用户webhook URL
    
    Args:
        user_id: 用户ID
        
    Returns:
        Webhook URL，如果不存在返回None
    """

实现逻辑:

• SELECT webhook_url FROM users WHERE user_id = ?
• 如果查询结果为None，返回None
• 如果webhook_url为None或空字符串，返回None
• 否则返回URL字符串

修改点4: 添加has_webhook_url方法

位置: 紧接get_user_webhook_url方法之后

方法签名:

def has_webhook_url(self, user_id: int) -> bool:
    """检查用户是否有个人webhook URL
    
    Args:
        user_id: 用户ID
        
    Returns:
        是否有个人URL
    """

实现逻辑:

• 调用get_user_webhook_url
• 检查返回值是否不为None且不为空字符串

修改点5: 添加check_users_webhook_urls方法（批量检查）

位置: 紧接has_webhook_url方法之后

方法签名:

def check_users_webhook_urls(self, user_ids: List[int]) -> Dict[int, bool]:
    """批量检查用户是否有个人webhook URL
    
    Args:
        user_ids: 用户ID列表
        
    Returns:
        字典 {user_id: has_url}
    """

实现逻辑:

• 如果user_ids为空，返回空字典
• 使用IN子句查询：SELECT user_id, webhook_url FROM users WHERE user_id IN (?)
• 构建结果字典：初始化为所有user_id为False
• 遍历查询结果，如果webhook_url不为None且不为空字符串，则设为True
• 返回结果字典

文件2: routers/callback.py

修改点1: 修改handle_register_command函数支持url子命令

位置: 第226-260行的handle_register_command函数

修改内容:

• 提取命令和参数后，检查第一个参数是否为"url"
• 如果是"url"，提取URL参数，验证URL格式（http/https开头），调用db.set_user_webhook_url
• 如果不是"url"，保持原有逻辑（注册名称）
• 更新帮助信息，包含两种用法

新的函数逻辑:

# 提取参数
_, args = CommandParser.extract_command_args(command)
args = args.strip()

# 检查是否为url子命令
parts = args.split(maxsplit=1)
if parts and parts[0].lower() == 'url':
    # 处理URL注册
    if len(parts) < 2:
        return "❌ 请提供webhook URL！\n\n正确格式：`.register url <URL>`\n\n示例：\n`.register url https://example.com/webhook?key=xxx`"
    webhook_url = parts[1].strip()
    # URL验证
    if not webhook_url.startswith(('http://', 'https://')):
        return "❌ URL格式无效！必须以 http:// 或 https:// 开头。"
    # 设置URL
    db = get_db()
    success = db.set_user_webhook_url(user_id, webhook_url)
    if success:
        return f"✅ Webhook URL注册成功！\n\n**您的个人URL**：{webhook_url}\n\n私聊消息将发送到此URL。"
    else:
        return "❌ 注册失败！请稍后重试。"
else:
    # 原有的名称注册逻辑
    ...

文件3: utils/message.py

修改点1: 添加send_private_message函数

位置: 在文件末尾，get_message_sender函数之后

函数签名:

async def send_private_message(user_id: int, content: str, msg_type: str = 'text') -> bool:
    """发送私聊消息到用户个人webhook URL
    
    Args:
        user_id: 目标用户ID
        content: 消息内容
        msg_type: 消息类型 ('text' 或 'markdown')
        
    Returns:
        是否发送成功，如果用户没有个人URL则返回False
    """

实现逻辑:

• 从数据库获取用户webhook URL
• 如果URL不存在，记录日志并返回False
• 创建MessageSender实例（使用用户的个人URL）
• 根据msg_type调用send_text或send_markdown
• 返回发送结果

文件4: core/models.py (新增数据模型)

修改点1: 添加PrivateMessageRequest模型

位置: 文件末尾

模型定义:

class PrivateMessageRequest(BaseModel):
    """私聊消息请求模型"""
    user_id: int = Field(..., description="目标用户ID")
    content: str = Field(..., description="消息内容")
    msg_type: str = Field(default="text", description="消息类型: text 或 markdown")

修改点2: 添加CheckBatchRequest模型

位置: 紧接PrivateMessageRequest之后

模型定义:

class CheckBatchRequest(BaseModel):
    """批量检查请求模型"""
    user_ids: List[int] = Field(..., description="用户ID列表")

修改点3: 添加CheckBatchResponse模型

位置: 紧接CheckBatchRequest之后

模型定义:

class CheckBatchResponse(BaseModel):
    """批量检查响应模型"""
    results: Dict[int, bool] = Field(..., description="用户ID到是否有URL的映射")

注意: core/models.py需要添加from typing import List导入（如果尚未导入）

文件5: routers/private.py (新建文件)

文件结构:

"""私聊相关API路由"""
import logging
from typing import List, Dict
from fastapi import APIRouter, HTTPException
from fastapi.responses import JSONResponse

from core.database import get_db
from core.models import PrivateMessageRequest, CheckBatchRequest, CheckBatchResponse
from utils.message import send_private_message

logger = logging.getLogger(__name__)

router = APIRouter()

接口1: POST /api/private/send

位置: router定义之后

函数签名:

@router.post("/private/send")
async def send_private(request: PrivateMessageRequest):
    """发送私聊消息
    
    请求体:
    {
        "user_id": 123456,
        "content": "消息内容",
        "msg_type": "text"  // 可选，默认为"text"
    }
    """

实现逻辑:

• 验证msg_type（必须是"text"或"markdown"），否则返回400错误
• 调用send_private_message
• 如果返回False（用户没有个人URL或发送失败），返回400错误和相应消息
• 成功则返回JSONResponse({"success": True, "message": "消息发送成功"})

接口2: GET /api/private/check/{user_id}

位置: send_private之后

函数签名:

@router.get("/private/check/{user_id}")
async def check_user_webhook(user_id: int):
    """检查用户是否有个人webhook URL"""

实现逻辑:

• 调用db.has_webhook_url(user_id)
• 返回JSONResponse({"user_id": user_id, "has_webhook_url": bool})

接口3: POST /api/private/check-batch

位置: check_user_webhook之后

函数签名:

@router.post("/private/check-batch")
async def check_users_webhook_batch(request: CheckBatchRequest):
    """批量检查用户是否有个人webhook URL
    
    请求体:
    {
        "user_ids": [123456, 789012, ...]
    }
    """

实现逻辑:

• 调用db.check_users_webhook_urls(request.user_ids)
• 返回CheckBatchResponse(results=...)

文件6: app.py

修改点1: 导入private路由

位置: 第13行，导入语句中

修改内容:

from routers import callback, health, private

修改点2: 注册private路由

位置: 第75-76行，路由注册部分

修改内容:

app.include_router(callback.router, prefix="/api", tags=["callback"])
app.include_router(health.router, tags=["health"])
app.include_router(private.router, prefix="/api", tags=["private"])

任务进度

[2025-11-03_09:45:56]

• 已修改：

  1. core/database.py - 添加webhook_url字段兼容性检查和4个数据库方法
  2. core/models.py - 添加3个API数据模型和List类型导入
  3. routers/callback.py - 扩展handle_register_command支持url子命令
  4. utils/message.py - 添加send_private_message函数
  5. routers/private.py - 新建文件，包含3个私聊相关API接口
  6. app.py - 导入并注册private路由

• 更改：

  1. 在users表中添加webhook_url字段支持（兼容性检查）
  2. 实现用户webhook URL的CRUD操作（设置、获取、检查、批量检查）
  3. 扩展.register命令支持.register url <url>子命令
  4. 封装私聊消息发送功能为独立函数
  5. 创建私聊相关的RESTful API接口（发送、单个检查、批量检查）
  6. 注册新的API路由到FastAPI应用

• 原因： 实现用户专属webhook URL注册和私聊消息发送功能，为其他系统提供API接口调用

• 阻碍因素： 无

• 状态：未确认

最终审查

待审查阶段完成...