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

### 一、准备阶段
1. **注册微信公众号**
   - 登录[微信公众平台](https://mp.weixin.qq.com/)
   - 选择「服务号」类型（需企业资质）或「订阅号」（个人可用但功能受限）
   - 完成资质认证（服务号需300元认证费）

2. **服务器准备**
   ```bash
   # 推荐配置
   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生成与绑定系统**
   ```python
   # 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. **用户关注处理**
   ```python
   # 微信消息处理框架（使用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. 消息推送接口开发
   ```python
   # 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配置**
   ```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. **数据库设计**
   ```sql
   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. **定时任务配置**
   ```bash
   # 每小时刷新AccessToken
   */60 * * * * /usr/bin/python3 /path/to/refresh_token.py
   ```

---

### 四、安全增强措施
1. **请求签名验证（可选）**
   ```python
   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. **频率限制中间件**
   ```python
   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. **单元测试用例**
   ```python
   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进行压力测试：
     ```bash
     ab -n 1000 -c 50 https://wx.yourdomain.com/TOKEN.send?text=test
     ```

---

### 六、维护与优化
1. **监控体系搭建**
   - 使用Prometheus + Grafana监控：
     - 请求QPS
     - 微信API调用成功率
     - Token使用分布

2. **消息队列优化**
   ```python
   # 使用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长连接维护

AI摘要

本文详细介绍了如何从零开始搭建微信推送服务，包括准备阶段、核心开发阶段、服务部署、安全增强措施、测试验证以及维护与优化。涉及到注册微信公众号、服务器准备、微信后台配置、Token生成与绑定系统、用户关注处理、消息推送接口开发、Nginx配置、数据库设计、定时任务配置、请求签名验证、频率限制中间件、单元测试用例、全流程测试、监控体系搭建、消息队列优化、灾备方案以及注意事项等多个方面。

一、准备阶段

注册微信公众号
- 登录微信公众平台
- 选择「服务号」类型（需企业资质）或「订阅号」（个人可用但功能受限）
- 完成资质认证（服务号需300元认证费）
服务器准备
```
# 推荐配置
CPU: 2核+ ｜ 内存: 2GB+ ｜ 带宽: 3Mbps+
系统: Ubuntu 22.04 LTS
安装组件: Nginx + MySQL + Redis + Python3.10+
```
- 购买云服务器（阿里云/腾讯云）
- 配置域名并申请SSL证书（Let's Encrypt免费证书）
- 备案域名（国内服务器必需）
微信后台配置
- 开发 → 基本配置 → 启用服务器配置
- 生成AppID和AppSecret
- 设置IP白名单（服务器公网IP）
- 配置消息加密密钥（EncodingAESKey）

二、核心开发阶段

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

用户关注处理

# 微信消息处理框架（使用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请妥善保存"

消息推送接口开发

# 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"

三、服务部署

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;
    }
}

数据库设计

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);

定时任务配置

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

四、安全增强措施

请求签名验证（可选）

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

频率限制中间件

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"]
)

五、测试验证

单元测试用例

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

全流程测试
- 使用微信开发者工具模拟用户关注
- 通过Postman发送测试请求
- 使用ab进行压力测试：
```
ab -n 1000 -c 50 https://wx.yourdomain.com/TOKEN.send?text=test
```

六、维护与优化

监控体系搭建
- 使用Prometheus + Grafana监控：
  - 请求QPS
  - 微信API调用成功率
  - Token使用分布

消息队列优化

# 使用Celery处理异步任务
from celery import Celery

celery = Celery('tasks', broker='redis://localhost:6379/0')

@celery.task
def async_send_message(openid, content):
    # 实际发送逻辑
    pass

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

注意事项

微信接口限制
- 客服消息：用户48小时内需有互动
- 频率限制：单个用户30条/分钟，全体5000条/天
- 内容安全：需自行实现敏感词过滤
法律合规
- 用户协议中明确消息推送用途
- 提供退订功能（发送TD退订）
- 遵守《网络安全法》和《个人信息保护法》

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

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

如果觉得我的文章对你有用，请随意赞赏

END

本文作者： xuan
文章标题：从零开始搭建微信推送服务
本文地址：https://blog.ybyq.wang/archives/184.html
版权说明：若无注明，本文皆Xuan's blog原创，转载请保留文章出处。

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

一、准备阶段

二、核心开发阶段

三、服务部署

四、安全增强措施

五、测试验证

六、维护与优化

注意事项

※最新文章推荐※

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款

基于uniapp+vant的健身会员管理系统h5前端

handsome主题添加服务器状态栏教程最新版，附源代码

解决宝塔Linux面板打不开 - 404 Not Found(Nginx)

handsome主题美化及优化：10.1.0最新版 - 1

handsome主题美化及优化：10.1.0最新版 - 2

晚霞与夕阳

Git报错：Another git process seems to be running in this repository

发现我的两篇csdn教程文章被转载

区块链是什么？小白必看

handsome主题添加服务器状态栏教程最新版，附源代码

一、准备阶段

二、核心开发阶段

三、服务部署

四、安全增强措施

五、测试验证

六、维护与优化

注意事项

※最新文章推荐※

发表评论 取消回复 使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款