当前位置:网站首页 >  教程

从零搭建轻量级精准智能推送系统 三步实现标签化精准用户触达

时间: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),方便多设备操作

相关推荐

最新

热门

推荐

精选

标签

易频IT社区是综合性互联网IT技术门户网站,专注分享网络技术、服务器运维、网络安全、编程开发、系统架构、云计算、大数据等行业干货,实时更新IT行业资讯、零基础教程、实战案例,为IT从业者、技术爱好者提供专业的学习交流平台。

Copyright © 2021-2026 易频IT社区. All Rights Reserved. 备案号:闽ICP备2023013482号 网站地图