从零搭建轻量级精准智能推送系统 三步实现标签化精准用户触达
时间:2026年06月14日 10:42:41
来源:易频IT社区
一、准备工具:确定3个核心组件,无需复杂服务器
本次用本地轻量级工具链,不用云服务器、不用AI算法框架,适合新手直接上手:
1. 数据存储工具:SQLite(Python内置,无需安装)
2. 推送交互工具:微信公众平台测试号(无需认证,零门槛,测试号地址:https://mp.weixin.qq.com/debug/cgi-bin/sandbox?t=sandbox/login)
3. 开发执行工具:Python3.8+(下载地址:https://www.python.org/downloads/,安装时务必勾选Add Python to PATH)
二、第一步:构建标签化用户数据库,完成精准画像基础
1. 创建Python文件,初始化SQLite数据库
新建名为`push_system.py`的文件,复制以下完整代码并运行:
```python
import sqlite3
初始化数据库连接,若不存在则自动创建push_data.db文件
conn = sqlite3.connect('push_data.db')
cursor = conn.cursor()
创建用户标签表,包含ID、微信openid、兴趣标签、上次推送时间4个必填字段
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_tags (
id INTEGER PRIMARY KEY AUTOINCREMENT,
openid TEXT NOT NULL UNIQUE,
interests TEXT NOT NULL,
last_push_time TEXT DEFAULT NULL
)
''')
提交并关闭连接
conn.commit()
conn.close()
print("✅ 标签化用户数据库创建成功!")
```
运行方式:打开电脑命令提示符/终端,进入`push_system.py`所在文件夹,输入`python push_system.py`回车即可。
2. 添加3个不同标签的测试用户
继续在`push_system.py`末尾追加以下代码(注意覆盖原有末尾的`conn.commit()`和`conn.close()`之后的部分,直接加在原初始化代码块下方),然后再次运行文件:
```python
追加测试用户代码
conn = sqlite3.connect('push_data.db')
cursor = conn.cursor()
测试用户openid获取:登录微信公众平台测试号,扫码关注后在「用户列表」复制
test_users = [
格式:(openid, 兴趣标签用逗号分隔, 上次推送时间留空)
("你的测试号用户1的openid", "科技,Python教程,职场", ""),
("你的测试号用户2的openid", "美食,烘焙,家常菜", ""),
("你的测试号用户3的openid", "健身,减脂,塑形", "")
]
try:
cursor.executemany('INSERT OR IGNORE INTO user_tags (openid, interests) VALUES (?, ?)', test_users)
conn.commit()
print("✅ 测试用户添加成功!")
except Exception as e:
print(f"❌ 添加失败:{e}")
finally:
conn.close()
```
注意:必须替换代码里的`你的测试号用户X的openid`为实际获取的内容!
三、第二步:对接微信公众平台测试号,实现标签匹配推送
1. 获取测试号的3个关键接口参数
登录微信公众平台测试号后,复制以下3个参数备用:
- AppID(测试号信息里直接显示)
- AppSecret(测试号信息里点击「重置」获取,仅显示1次,请务必保存)
- 模板消息ID(点击「新增测试模板」,填写标题「精准推送测试」,内容「今天给您推荐{{first.DATA}}!{{keyword1.DATA}}」,提交后自动生成ID,复制备用)
2. 编写标签匹配+推送的核心代码
再次打开`push_system.py`,删除所有测试用户添加的代码(保留第一步的初始化数据库代码块),然后追加以下完整代码并保存:
```python
import requests
import json
import time
from datetime import datetime
微信测试号的3个关键参数,替换为自己的!
APP_ID = "你的测试号AppID"
APP_SECRET = "你的测试号AppSecret"
TEMPLATE_ID = "你的测试号模板消息ID"
每个标签对应的推荐内容,可自由修改
TAG_CONTENTS = {
"科技": ("热门科技动态", "ChatGPT4o mini发布、苹果Vision Pro2谍照曝光"),
"Python教程": ("Python入门干货", "一行代码实现Excel批量转PDF"),
"职场": ("职场效率工具", "Notion AI 自动生成周报模板"),
"美食": ("今日快手菜", "15分钟搞定番茄炒蛋盖浇饭"),
"烘焙": ("新手烘焙入门", "零失败曲奇饼干教程"),
"家常菜": ("周末硬菜推荐", "可乐鸡翅超详细步骤"),
"健身": ("今日居家训练", "10分钟平板支撑进阶训练"),
"减脂": ("低卡早餐推荐", "燕麦牛奶搭配水煮蛋,饱腹感强"),
"塑形": ("瑜伽入门体式", "猫牛式缓解腰背酸痛")
}
获取微信access_token的函数(有效期2小时)
def get_access_token():
url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={APP_ID}&secret={APP_SECRET}"
response = requests.get(url).json()
if "access_token" in response:
return response["access_token"]
else:
print(f"❌ 获取access_token失败:{response}")
return None
标签匹配+推送的主函数
def push_by_tags():
access_token = get_access_token()
if not access_token:
return
conn = sqlite3.connect('push_data.db')
cursor = conn.cursor()
获取所有用户信息
cursor.execute('SELECT openid, interests FROM user_tags')
users = cursor.fetchall()
push_count = 0
for openid, interests_str in users:
分割用户标签,取第一个标签做匹配(新手版简单逻辑,可后续优化)
first_tag = interests_str.split(",")[0].strip()
if first_tag not in TAG_CONTENTS:
print(f"⚠️ 用户{openid}的标签{first_tag}无对应内容,跳过")
continue
准备模板消息数据
first, keyword1 = TAG_CONTENTS[first_tag]
template_data = {
"touser": openid,
"template_id": TEMPLATE_ID,
"data": {
"first": {"value": first, "color": "173177"},
"keyword1": {"value": keyword1, "color": "173177"}
}
}
发送模板消息
push_url = f"https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={access_token}"
response = requests.post(push_url, data=json.dumps(template_data)).json()
if response.get("errcode") == 0:
更新用户上次推送时间
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cursor.execute('UPDATE user_tags SET last_push_time = ? WHERE openid = ?', (current_time, openid))
push_count += 1
print(f"✅ 用户{openid}(标签:{first_tag})推送成功!")
else:
print(f"❌ 用户{openid}推送失败:{response}")
conn.commit()
conn.close()
print(f"\n🎉 本次共推送成功{push_count}位用户!")
if __name__ == "__main__":
push_by_tags()
```
3. 运行推送代码
再次打开命令提示符/终端,进入文件夹,输入`python push_system.py`回车,关注测试号的微信用户即可收到对应的标签化精准消息!
四、第三步:优化落地细节,避免推送干扰
1. 限制每日推送次数
在`push_by_tags()`函数的获取所有用户信息代码块后,添加过滤逻辑:
```python
过滤代码(加在cursor.execute('SELECT openid, interests FROM user_tags')之后)
today_date = datetime.now().strftime("%Y-%m-%d")
cursor.execute('SELECT openid, interests, last_push_time FROM user_tags')
users = []
for row in cursor.fetchall():
openid, interests_str, last_push = row
若今日已推送过,则跳过
if last_push and last_push.startswith(today_date):
print(f"⚠️ 用户{openid}今日已推送,跳过")
continue
users.append((openid, interests_str))
```
2. 自动定时推送(Windows系统)
不需要下载第三方定时工具,用Windows自带任务计划程序:
1. 搜索「任务计划程序」并打开
2. 点击右侧「创建基本任务」
3. 名称填写「每日精准推送」,下一步
4. 选择「每天」,设置具体推送时间(比如18:00),下一步
5. 选择「启动程序」,下一步
6. 程序或脚本栏:填写Python的完整路径(比如`C:\Users\你的用户名\AppData\Local\Programs\Python\Python310\python.exe`)
7. 添加参数栏:填写`push_system.py`的完整路径(比如`"D:\Work\push_system\push_system.py"`,注意加双引号,避免路径含空格报错)
8. 起始于栏:填写`push_system.py`所在文件夹的完整路径(比如`D:\Work\push_system`,无需引号)
9. 下一步,完成
3. 自动定时推送(Mac/Linux系统)
用系统自带的crontab:
1. 打开终端,输入`crontab -e`回车
2. 按`i`键进入编辑模式,添加一行(比如每天18:00推送):
```
0 18 /usr/bin/python3 /Users/你的用户名/Desktop/push_system.py
```
注意:替换Python3路径(用`which python3`获取)和文件完整路径
3. 按`Esc`键退出编辑,输入`:wq`回车保存
五、简单优化方向(可选,有基础后可尝试)
1. 将标签匹配逻辑从「取第一个」改为「按用户点击历史选最热」
2. 接入百度AI的免费文本分类API(地址:https://cloud.baidu.com/product/nlp/text_classify),自动给新关注用户打标签
3. 将SQLite替换为免费的云数据库MySQL(比如腾讯云免费版,地址:https://cloud.tencent.com/product/cdb/free),方便多设备操作