表单获客分层实战：四步构建精准用户画像与自动化触达系统

一、核心架构设计

表单获客分层系统的核心是通过用户行为数据自动划分价值等级，实现差异化营销。系统由数据采集层、规则引擎层、用户画像层、触达执行层构成。

1.1 数据采集层配置

在网站全局head标签内插入以下JavaScript代码：

``` ```

1.2 数据库表结构

创建MySQL用户分层表：

``` CREATE TABLE user_tiers ( id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(64) NOT NULL, email VARCHAR(255) NOT NULL, tier_level ENUM('hot', 'warm', 'cold') DEFAULT 'cold', score INT DEFAULT 0, last_submit_time TIMESTAMP, submit_count INT DEFAULT 0, form_data JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_tier (tier_level), INDEX idx_score (score), UNIQUE KEY uk_email (email) ); ```

二、分层规则配置

2.1 评分规则定义

在config/rules.yaml中配置评分规则：

``` scoring_rules: - field: company_size values: "1000+": 30 "500-999": 20 "100-499": 15 "50-99": 10 "1-49": 5 - field: budget values: "100万以上": 40 "50-100万": 30 "10-50万": 20 "10万以下": 5 - field: job_title values: "总监及以上": 25 "经理": 15 "主管": 10 "员工": 5 - field: submit_time condition: "within_24h" score: 20 tier_thresholds: hot: 70 warm: 40 cold: 0 ```

2.2 实时计算脚本

创建Python计算脚本calculate_tier.py：

``` import yaml from datetime import datetime, timedelta class TierCalculator: def __init__(self, config_path='config/rules.yaml'): with open(config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) def calculate_score(self, form_data): total_score = 0 字段匹配评分 for rule in self.config['scoring_rules']: field = rule['field'] if field in form_data: value = form_data[field] if 'values' in rule and value in rule['values']: total_score += rule['values'][value] elif 'condition' in rule: if rule['condition'] == 'within_24h': submit_time = datetime.fromisoformat(form_data['submit_time']) if datetime.now() - submit_time < timedelta(hours=24): total_score += rule['score'] 确定分层 thresholds = self.config['tier_thresholds'] if total_score >= thresholds['hot']: tier = 'hot' elif total_score >= thresholds['warm']: tier = 'warm' else: tier = 'cold' return { 'score': total_score, 'tier': tier, 'calculated_at': datetime.now().isoformat() } ```

三、数据采集与处理

3.1 前端数据收集

创建form-tracker.js：

``` class FormTracker { constructor(config) { this.config = config; this.forms = document.querySelectorAll('form[data-track]'); this.init(); } init() { this.forms.forEach(form => { form.addEventListener('submit', (e) => { this.handleSubmit(e, form); }); }); } async handleSubmit(event, form) { event.preventDefault(); const formData = new FormData(form); const data = {}; this.config.fields.forEach(field => { if (formData.has(field)) { data[field] = formData.get(field); } }); data['form_id'] = form.id || form.getAttribute('id'); data['page_url'] = window.location.href; data['submit_time'] = new Date().toISOString(); data['user_agent'] = navigator.userAgent; try { const response = await fetch(this.config.endpoint, { method: 'POST', headers: { 'Content-Type': 'application/json', 'X-Tracking-ID': this.config.trackingId }, body: JSON.stringify(data) }); if (response.ok) { form.submit(); // 提交原始表单 } } catch (error) { console.error('Tracking failed:', error); form.submit(); // 降级处理，仍然提交表单 } } } if (window.formTierConfig) { new FormTracker(window.formTierConfig); } ```

3.2 后端API接口

创建Flask处理接口：

``` from flask import Flask, request, jsonify from calculate_tier import TierCalculator import mysql.connector import json app = Flask(__name__) calculator = TierCalculator() 数据库配置 db_config = { 'host': 'localhost', 'user': 'form_user', 'password': 'your_password', 'database': 'form_tracking' } @app.route('/track', methods=['POST']) def track_form(): try: data = request.json 计算分层 result = calculator.calculate_score(data) 存储到数据库 conn = mysql.connector.connect(db_config) cursor = conn.cursor() insert_query = """ INSERT INTO user_tiers (user_id, email, tier_level, score, last_submit_time, submit_count, form_data) VALUES (%s, %s, %s, %s, %s, 1, %s) ON DUPLICATE KEY UPDATE tier_level = VALUES(tier_level), score = VALUES(score), last_submit_time = VALUES(last_submit_time), submit_count = submit_count + 1, form_data = VALUES(form_data) """ cursor.execute(insert_query, ( data.get('email', ''), data.get('email', ''), result['tier'], result['score'], data['submit_time'], json.dumps(data, ensure_ascii=False) )) conn.commit() cursor.close() conn.close() return jsonify({ 'success': True, 'tier': result['tier'], 'score': result['score'] }) except Exception as e: return jsonify({ 'success': False, 'error': str(e) }), 500 if __name__ == '__main__': app.run(port=5000, debug=True) ```

四、自动化触达系统

4.1 邮件营销配置

创建邮件模板目录结构：

``` templates/emails/ ├── hot-lead.html 高意向客户 ├── warm-lead.html 中等意向客户 ├── cold-lead.html 低意向客户 └── config.yaml 邮件发送配置 ```

config.yaml配置：

``` smtp: host: smtp.yourdomain.com port: 587 username: noreply@yourdomain.com password: your_smtp_password use_tls: true templates: hot: subject: "【优先处理】专属客户经理已为您就位" delay_minutes: 5 cc: ["sales@yourdomain.com"] warm: subject: "您需要的产品资料已发送" delay_minutes: 30 cold: subject: "感谢您关注我们" delay_minutes: 60 include_newsletter: true ```

4.2 自动化工作流

创建Celery定时任务worker.py：

``` from celery import Celery import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import yaml import mysql.connector from datetime import datetime, timedelta import jinja2 app = Celery('form_tasks', broker='redis://localhost:6379/0') 加载配置 with open('templates/emails/config.yaml', 'r', encoding='utf-8') as f: email_config = yaml.safe_load(f) 数据库配置 db_config = { 'host': 'localhost', 'user': 'form_user', 'password': 'your_password', 'database': 'form_tracking' } @app.task def send_tiered_emails(): """发送分层邮件""" conn = mysql.connector.connect(db_config) cursor = conn.cursor(dictionary=True) 查询需要发送邮件的用户 query = """ SELECT FROM user_tiers WHERE tier_level IN ('hot', 'warm', 'cold') AND last_submit_time >= %s AND email_sent = 0 """ one_hour_ago = datetime.now() - timedelta(hours=1) cursor.execute(query, (one_hour_ago,)) users = cursor.fetchall() for user in users: send_email.delay(user) 标记已发送 update_query = "UPDATE user_tiers SET email_sent = 1 WHERE id = %s" cursor.execute(update_query, (user['id'],)) conn.commit() cursor.close() conn.close() @app.task def send_email(user_data): """发送单封邮件""" tier = user_data['tier_level'] config = email_config['templates'][tier] 加载模板 template_loader = jinja2.FileSystemLoader('templates/emails/') template_env = jinja2.Environment(loader=template_loader) template = template_env.get_template(f'{tier}-lead.html') 渲染内容 html_content = template.render( name=user_data['form_data'].get('name', ''), company=user_data['form_data'].get('company', ''), tier=tier ) 创建邮件 msg = MIMEMultipart('alternative') msg['Subject'] = config['subject'] msg['From'] = email_config['smtp']['username'] msg['To'] = user_data['email'] if 'cc' in config: msg['Cc'] = ', '.join(config['cc']) msg.attach(MIMEText(html_content, 'html')) 发送邮件 with smtplib.SMTP(email_config['smtp']['host'], email_config['smtp']['port']) as server: if email_config['smtp']['use_tls']: server.starttls() server.login( email_config['smtp']['username'], email_config['smtp']['password'] ) server.send_message(msg) @app.task def sync_to_crm(): """同步高意向客户到CRM系统""" conn = mysql.connector.connect(db_config) cursor = conn.cursor(dictionary=True) query = """ SELECT FROM user_tiers WHERE tier_level = 'hot' AND crm_synced = 0 AND last_submit_time >= %s """ twentyfour_hours_ago = datetime.now() - timedelta(hours=24) cursor.execute(query, (twentyfour_hours_ago,)) hot_leads = cursor.fetchall() for lead in hot_leads: 调用CRM API crm_data = { 'email': lead['email'], 'name': lead['form_data'].get('name', ''), 'company': lead['form_data'].get('company', ''), 'score': lead['score'], 'source': 'form_tiering_system' } 这里添加实际的CRM API调用代码 response = requests.post(CRM_API_URL, json=crm_data) 标记已同步 update_query = "UPDATE user_tiers SET crm_synced = 1 WHERE id = %s" cursor.execute(update_query, (lead['id'],)) conn.commit() cursor.close() conn.close() 设置定时任务 app.conf.beat_schedule = { 'send-emails-every-5-minutes': { 'task': 'worker.send_tiered_emails', 'schedule': 300.0, 每5分钟 }, 'sync-to-crm-every-hour': { 'task': 'worker.sync_to_crm', 'schedule': 3600.0, 每小时 }, } ```

4.3 监控与报表

创建监控SQL查询：

``` -- 实时分层统计 SELECT tier_level, COUNT() as count, AVG(score) as avg_score, MAX(last_submit_time) as latest_submit FROM user_tiers WHERE last_submit_time >= DATE_SUB(NOW(), INTERVAL 7 DAY) GROUP BY tier_level ORDER BY CASE tier_level WHEN 'hot' THEN 1 WHEN 'warm' THEN 2 WHEN 'cold' THEN 3 END; -- 转化率分析 SELECT DATE(last_submit_time) as submit_date, tier_level, COUNT() as submissions, SUM(CASE WHEN crm_synced = 1 THEN 1 ELSE 0 END) as crm_conversions, ROUND(SUM(CASE WHEN crm_synced = 1 THEN 1 ELSE 0 END) 100.0 / COUNT(), 2) as conversion_rate FROM user_tiers WHERE last_submit_time >= DATE_SUB(NOW(), INTERVAL 30 DAY) GROUP BY DATE(last_submit_time), tier_level ORDER BY submit_date DESC; ```

五、部署与维护

5.1 Docker部署配置

创建docker-compose.yml：

``` version: '3.8' services: mysql: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: root_password MYSQL_DATABASE: form_tracking MYSQL_USER: form_user MYSQL_PASSWORD: user_password volumes: - mysql_data:/var/lib/mysql ports: - "3306:3306" redis: image: redis:7-alpine ports: - "6379:6379" api: build: . ports: - "5000:5000" environment: DB_HOST: mysql DB_PASSWORD: user_password depends_on: - mysql - redis celery-worker: build: . command: celery -A worker.app worker --loglevel=info depends_on: - redis - mysql celery-beat: build: . command: celery -A worker.app beat --loglevel=info depends_on: - redis - mysql volumes: mysql_data: ```

5.2 Nginx反向代理配置

/etc/nginx/sites-available/form-tiering.conf：

``` server { listen 80; server_name api.yourdomain.com; location / { proxy_pass http://localhost:5000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } 静态文件服务 location /static/ { alias /var/www/form-tiering/static/; expires 30d; } } ```

执行以下命令启用配置：

``` sudo ln -s /etc/nginx/sites-available/form-tiering.conf /etc/nginx/sites-enabled/ sudo nginx -t sudo systemctl reload nginx ```

5.3 系统启动命令

创建启动脚本start.sh：

``` !/bin/bash 启动所有服务 docker-compose up -d 等待MySQL就绪 sleep 30 初始化数据库 mysql -h localhost -u form_user -puser_password form_tracking < init.sql 检查服务状态 echo "API服务状态：" curl -f http://localhost:5000/health || echo "API服务异常" echo "Celery Worker状态：" docker-compose exec celery-worker celery -A worker.app inspect ping echo "系统启动完成，访问地址：https://api.yourdomain.com" ```

给脚本添加执行权限：

``` chmod +x start.sh ./start.sh ```

系统启动后，将前端JavaScript代码中的endpoint地址修改为你的域名，表单提交数据即可自动进入分层系统。所有配置文件和脚本都已提供完整代码，可直接复制使用。