import os import json import random import pandas as pd from flask import Flask, render_template_string, request, jsonify from faker import Faker import logging # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) fake = Faker(['zh_CN']) # Default Scoring Model DEFAULT_MODEL = { "demographic": [ {"field": "role", "operator": "contains", "value": "CEO", "score": 20, "desc": "职位包含 CEO"}, {"field": "role", "operator": "contains", "value": "总监", "score": 15, "desc": "职位包含 总监"}, {"field": "role", "operator": "contains", "value": "经理", "score": 10, "desc": "职位包含 经理"}, {"field": "industry", "operator": "equals", "value": "互联网", "score": 10, "desc": "行业为 互联网"}, {"field": "company_size", "operator": "gt", "value": 100, "score": 15, "desc": "公司规模 > 100人"} ], "behavioral": [ {"field": "website_visits", "operator": "gt", "value": 5, "score": 10, "desc": "访问官网 > 5次"}, {"field": "email_opens", "operator": "gt", "value": 0, "score": 5, "desc": "打开过邮件"}, {"field": "downloaded_whitepaper", "operator": "equals", "value": True, "score": 20, "desc": "下载过白皮书"}, {"field": "webinar_attended", "operator": "equals", "value": True, "score": 15, "desc": "参加过研讨会"} ] } def evaluate_rule(lead, rule): field = rule.get('field') operator = rule.get('operator') target = rule.get('value') score = rule.get('score', 0) val = lead.get(field) if val is None: return 0 matched = False try: if operator == 'equals': # Handle boolean/string comparison carefully if isinstance(target, bool): matched = bool(val) == target elif isinstance(val, str) and isinstance(target, str): matched = val.lower() == target.lower() else: matched = val == target elif operator == 'contains': matched = str(target).lower() in str(val).lower() elif operator == 'gt': matched = float(val) > float(target) elif operator == 'lt': matched = float(val) < float(target) elif operator == 'gte': matched = float(val) >= float(target) elif operator == 'lte': matched = float(val) <= float(target) except Exception as e: logger.warning(f"Error evaluating rule {rule} for value {val}: {e}") matched = False return score if matched else 0 @app.route('/') def index(): return render_template_string(HTML_TEMPLATE) @app.route('/api/generate-leads', methods=['POST']) def generate_leads(): try: count = request.json.get('count', 10) leads = [] industries = ['互联网', '金融', '制造业', '教育', '医疗', '零售'] roles = ['CEO', 'CTO', '市场总监', '销售经理', '研发工程师', '运营专员', '采购经理'] for _ in range(count): leads.append({ "id": fake.uuid4(), "name": fake.name(), "company": fake.company(), "role": random.choice(roles), "industry": random.choice(industries), "company_size": random.randint(10, 5000), "email": fake.email(), "website_visits": random.randint(0, 50), "email_opens": random.randint(0, 20), "downloaded_whitepaper": random.choice([True, False]), "webinar_attended": random.choice([True, False]), "last_contact_days": random.randint(1, 100) }) return jsonify(leads) except Exception as e: logger.error(f"Error generating leads: {e}") return jsonify({"error": str(e)}), 500 @app.route('/api/score', methods=['POST']) def score_leads(): try: data = request.json leads = data.get('leads', []) model = data.get('model', DEFAULT_MODEL) results = [] for lead in leads: total_score = 0 breakdown = [] # Demographic for rule in model.get('demographic', []): points = evaluate_rule(lead, rule) if points > 0: total_score += points breakdown.append({"desc": rule['desc'], "score": points, "type": "基本属性"}) # Behavioral for rule in model.get('behavioral', []): points = evaluate_rule(lead, rule) if points > 0: total_score += points breakdown.append({"desc": rule['desc'], "score": points, "type": "行为数据"}) # Determine Grade grade = 'D' if total_score >= 80: grade = 'A' elif total_score >= 60: grade = 'B' elif total_score >= 40: grade = 'C' results.append({ **lead, "score": total_score, "grade": grade, "breakdown": breakdown }) # Sort by score desc results.sort(key=lambda x: x['score'], reverse=True) return jsonify(results) except Exception as e: logger.error(f"Error scoring leads: {e}") return jsonify({"error": str(e)}), 500 @app.route('/api/upload', methods=['POST']) def upload_file(): if 'file' not in request.files: return jsonify({"error": "No file part"}), 400 file = request.files['file'] if file.filename == '': return jsonify({"error": "No selected file"}), 400 try: if file.filename.endswith('.csv'): df = pd.read_csv(file) elif file.filename.endswith(('.xls', '.xlsx')): df = pd.read_excel(file) else: return jsonify({"error": "Unsupported file format. Please use CSV or Excel."}), 400 # Ensure required columns exist or fill with defaults required_fields = ['name', 'company', 'role', 'industry', 'company_size', 'website_visits', 'email_opens'] for field in required_fields: if field not in df.columns: if field == 'name': df['name'] = 'Unknown' elif field == 'company': df['company'] = 'Unknown' else: df[field] = 0 # Default numeric leads = df.to_dict('records') # Add ID if missing for lead in leads: if 'id' not in lead: lead['id'] = fake.uuid4() # Normalize boolean fields if 'downloaded_whitepaper' in lead: lead['downloaded_whitepaper'] = bool(lead['downloaded_whitepaper']) else: lead['downloaded_whitepaper'] = False if 'webinar_attended' in lead: lead['webinar_attended'] = bool(lead['webinar_attended']) else: lead['webinar_attended'] = False return jsonify(leads) except Exception as e: logger.error(f"Error processing file: {e}") return jsonify({"error": f"File processing failed: {str(e)}"}), 500 HTML_TEMPLATE = """ 销售线索智能评分引擎 | Lead Scoring Engine
${ toast.message }

销售线索智能评分引擎

基于多维数据的智能化潜客分级系统

评分模型配置

当前版本: v1.0

基本属性规则 (Demographic)

${ rule.desc }
${ rule.field } ${ rule.operator } ${ rule.value }
+${ rule.score }

行为数据规则 (Behavioral)

${ rule.desc }
${ rule.field } ${ rule.operator } ${ rule.value }
+${ rule.score }

线索质量分布

线索列表 (${ leads.length })

A级(High) B级(Med) C级(Low)
姓名/公司 职位 行为指标 总分 等级 得分详情
${ lead.name }
${ lead.company } (${ lead.industry })
${ lead.role }
${ lead.company_size }人
访客:${lead.website_visits} 白皮书
${ lead.score || '-' } ${ lead.grade } -
${ item.desc } +${ item.score }
+${ lead.breakdown.length - 2 } 更多...

暂无数据,请点击右上角生成数据或导入文件

""" if __name__ == '__main__': app.run(host='0.0.0.0', port=7860, debug=True)