无需复杂开发环境,全是免费轻量工具:
pip install notion-client注册Notion后,点击「New Page」,选择「Database」-「Table」,重命名页面为「日常分层运营用户库」
默认的「Name」字段直接改为「用户ID/手机号」(唯一标识,必填),点击表头右侧的「+」,依次添加以下字段并严格配置类型:
这是连接Python的核心,步骤不能错:

在电脑桌面新建一个txt文件,重命名为「user_segmentation.py」,务必确保后缀名是.py(Windows需先开启显示文件扩展名)
将以下代码完整复制到「user_segmentation.py」中,只修改最上方的3个配置项:
from notion_client import Client
from datetime import datetime, timedelta
-- 仅需修改此处3个配置项 --
NOTION_TOKEN = "secret_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" 替换为你的Internal Integration Token
DATABASE_ID = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" 替换为你的32位数据库ID
分层规则(可根据业务调整天数和金额阈值)
RULES = {
"high_value_threshold": 1000, 高价值最低累计消费
"active_days": 7, 活跃判定:最近7天内有活跃
"sleep_days": 30, 沉睡判定:最近30天内无活跃但超7天
"churn_days": 90 流失判定:最近90天内无活跃
}
初始化Notion客户端
notion = Client(auth=NOTION_TOKEN)
获取当前时间(UTC+8,适配国内业务)
now = datetime.utcnow() + timedelta(hours=8)
def get_all_users():
"""获取数据库所有用户"""
users = []
has_more = True
start_cursor = None
while has_more:
response = notion.databases.query(
database_id=DATABASE_ID,
start_cursor=start_cursor
)
users.extend(response["results"])
has_more = response["has_more"]
start_cursor = response["next_cursor"]
return users
def calculate_segment(user):
"""根据字段计算分层标签"""
提取必要字段,处理空值
try:
total_spent = float(user["properties"]["累计消费金额"]["number"] or 0)
except (KeyError, TypeError):
total_spent = 0
try:
last_active_str = user["properties"]["最近活跃日期"]["date"]["start"]
last_active = datetime.fromisoformat(last_active_str.replace("Z", "+00:00")) + timedelta(hours=8)
except (KeyError, TypeError, ValueError):
return "低价值流失" 无活跃时间直接归为流失
计算活跃天数差
days_since_active = (now - last_active).days
按规则判断
if total_spent >= RULES["high_value_threshold"]:
if days_since_active <= RULES["active_days"]:
return "高价值活跃"
elif days_since_active <= RULES["sleep_days"]:
return "高价值沉睡"
else:
return "低价值流失"
else:
if days_since_active <= RULES["active_days"]:
return "低价值活跃"
else:
return "低价值流失"
def update_user_segment(page_id, segment):
"""更新用户的自动分层标签"""
notion.pages.update(
page_id=page_id,
properties={
"自动分层标签": {
"select": {
"name": segment
}
}
}
)
if __name__ == "__main__":
print(f"开始执行分层任务,当前时间:{now.strftime('%Y-%m-%d %H:%M:%S')}")
all_users = get_all_users()
print(f"共获取到 {len(all_users)} 条用户数据")
updated_count = 0
for user in all_users:
current_segment = None
获取当前标签(如果有)
try:
current_segment = user["properties"]["自动分层标签"]["select"]["name"]
except (KeyError, TypeError):
pass
计算新标签
new_segment = calculate_segment(user)
只有标签变化时才更新,节省API调用次数
if new_segment != current_segment:
update_user_segment(user["id"], new_segment)
updated_count += 1
user_id = user["properties"]["用户ID/手机号"]["title"][0]["text"]["content"]
print(f"已更新用户 {user_id} 的标签:{current_segment or '无'} → {new_segment}")
print(f"分层任务完成,共更新 {updated_count} 条用户数据")
无需云服务器,本地Windows即可免费定时:
用系统自带的crontab即可:
0 1 /usr/local/bin/python3 /Users/你的用户名/Desktop/user_segmentation.py >> /Users/你的用户名/Desktop/segmentation_log.txt 2>&1
日志会自动保存到桌面的「segmentation_log.txt」中,方便查看运行情况
上一篇: 数字化运营工具体系构建与效能提升实战
下一篇: 构建高效的用户运营激活体系与实战策略
易频IT社区是综合性互联网IT技术门户网站,专注分享网络技术、服务器运维、网络安全、编程开发、系统架构、云计算、大数据等行业干货,实时更新IT行业资讯、零基础教程、实战案例,为IT从业者、技术爱好者提供专业的学习交流平台。
Copyright © 2021-2026 易频IT社区. All Rights Reserved. 备案号:闽ICP备2023013482号 网站地图