从零开始搭建微信推送服务

一、准备阶段

1. 注册微信公众号

   • 登录微信公众平台
   • 选择「服务号」类型（需企业资质）或「订阅号」（个人可用但功能受限）
   • 完成资质认证（服务号需300元认证费）

2. 服务器准备

   # 推荐配置
   CPU: 2核+ ｜ 内存: 2GB+ ｜ 带宽: 3Mbps+
   系统: Ubuntu 22.04 LTS
   安装组件: Nginx + MySQL + Redis + Python3.10+

   • 购买云服务器（阿里云/腾讯云）
   • 配置域名并申请SSL证书（Let's Encrypt免费证书）
   • 备案域名（国内服务器必需）

3. 微信后台配置

   • 开发 → 基本配置 → 启用服务器配置
   • 生成AppID和AppSecret
   • 设置IP白名单（服务器公网IP）
   • 配置消息加密密钥（EncodingAESKey）

二、核心开发阶段

1. Token生成与绑定系统

   # Token生成逻辑（示例）
   import uuid
   from datetime import datetime, timedelta
   
   def generate_token(openid):
       token = str(uuid.uuid4()).replace('-','')[:16]
       expire_time = datetime.now() + timedelta(days=365)
       # 存储到数据库
       sql = '''
       INSERT INTO user_tokens (openid, token, expire_time) 
       VALUES (%s, %s, %s)
       '''
       cursor.execute(sql, (openid, token, expire_time))
       return token

2. 用户关注处理

   # 微信消息处理框架（使用werobot库）
   import werobot
   
   robot = werobot.WeRoBot(token='YOUR_WECHAT_TOKEN')
   robot.config["APP_ID"] = "YOUR_APPID"
   robot.config["APP_SECRET"] = "YOUR_APPSECRET"
   
   @robot.subscribe
   def subscribe_handler(message):
       openid = message.source
       token = generate_token(openid)
       return f"Token已生成：{token}\n请妥善保存"

3. 消息推送接口开发

   # Flask消息处理接口
   from flask import Flask, request
   import requests
   
   app = Flask(__name__)
   
   @app.route('/<token>.send', methods=['GET', 'POST'])
   def send_message(token):
       # 验证Token
       openid = redis.get(f"token:{token}")
       if not openid:
           return "Invalid Token", 403
       
       # 获取消息内容
       text = request.args.get('text') or request.form.get('text')
       desp = request.args.get('desp') or request.form.get('desp', '')
       
       # 调用微信API
       access_token = get_access_token()  # 需实现AccessToken获取
       url = f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={access_token}"
       payload = {
           "touser": openid,
           "msgtype": "text",
           "text": {"content": f"{text}\n\n{desp}"}
       }
       response = requests.post(url, json=payload)
       return "Message Sent" if response.json().get('errcode') == 0 else "Error"

三、服务部署

1. Nginx配置

   server {
       listen 443 ssl;
       server_name wx.yourdomain.com;
       
       ssl_certificate /path/to/fullchain.pem;
       ssl_certificate_key /path/to/privkey.pem;
       
       location / {
           proxy_pass http://127.0.0.1:5000;
           proxy_set_header Host $host;
           proxy_set_header X-Real-IP $remote_addr;
       }
   }

2. 数据库设计

   CREATE TABLE user_tokens (
       id INT AUTO_INCREMENT PRIMARY KEY,
       openid VARCHAR(32) UNIQUE,
       token VARCHAR(16) UNIQUE,
       create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
       expire_time DATETIME,
       last_used DATETIME
   );
   
   CREATE INDEX idx_token ON user_tokens(token);

3. 定时任务配置

   # 每小时刷新AccessToken
   */60 * * * * /usr/bin/python3 /path/to/refresh_token.py

四、安全增强措施

1. 请求签名验证（可选）

   def verify_signature(token, timestamp, nonce, signature):
       tmp_list = sorted([token, timestamp, nonce])
       tmp_str = ''.join(tmp_list)
       import hashlib
       sha1 = hashlib.sha1(tmp_str.encode()).hexdigest()
       return sha1 == signature

2. 频率限制中间件

   from flask_limiter import Limiter
   from flask_limiter.util import get_remote_address
   
   limiter = Limiter(
       app=app,
       key_func=get_remote_address,
       default_limits=["200 per day", "50 per hour"]
   )

五、测试验证

1. 单元测试用例

   def test_token_generation():
       token = generate_token('test_openid')
       assert len(token) == 16
       assert redis.exists(f'token:{token}')
   
   def test_message_send():
       with app.test_client() as c:
           resp = c.get('/VALID_TOKEN.send?text=Test')
           assert b'Message Sent' in resp.data

2. 全流程测试

   • 使用微信开发者工具模拟用户关注
   • 通过Postman发送测试请求

   • 使用ab进行压力测试：

     ab -n 1000 -c 50 https://wx.yourdomain.com/TOKEN.send?text=test

六、维护与优化

1. 监控体系搭建

   • 使用Prometheus + Grafana监控：

     • 请求QPS
     • 微信API调用成功率
     • Token使用分布

2. 消息队列优化

   # 使用Celery处理异步任务
   from celery import Celery
   
   celery = Celery('tasks', broker='redis://localhost:6379/0')
   
   @celery.task
   def async_send_message(openid, content):
       # 实际发送逻辑
       pass

3. 灾备方案

   • 部署双活服务器（不同可用区）
   • 定期备份数据库（每日全量+每小时增量）
   • 配置微信API故障自动切换（备用公众号）

注意事项

1. 微信接口限制

   • 客服消息：用户48小时内需有互动
   • 频率限制：单个用户30条/分钟，全体5000条/天
   • 内容安全：需自行实现敏感词过滤

2. 法律合规

   • 用户协议中明确消息推送用途
   • 提供退订功能（发送TD退订）
   • 遵守《网络安全法》和《个人信息保护法》

如需扩展Bark推送功能，需额外实现：

1. APNs证书申请（Apple开发者账号）
2. 设备注册管理系统
3. HTTP/2长连接维护