### 使用 Python 开发满意度外呼和调研信息记录系统的最佳实践和实现方法 #### 1. 系统架构概述 为了高效实现所需功能，推荐采用三层架构设计：前端负责用户交互，后端处理核心业务逻辑，数据库存储数据。以下是具体的技术栈建议： - **前端**: Flask 或 FastAPI 结合 Jinja2 模板渲染页面。 - **后端**: 利用 SQLAlchemy 或 Peewee ORM 进行数据库操作。 - **数据库**: SQLite（小型项目）、PostgreSQL 或 MySQL。 --- #### 2. 功能模块划分与实现细节 ##### (1) 数据录入模块 支持手动录入和批量导入两种方式。 ###### 手动录入 创建 HTML 表单供用户填写必要字段，如客户 ID、呼叫结果、满意度评分等。提交后由后端保存至数据库。 ```html <form action="/submit_call_record" method="post"> <label for="customer_id">Customer ID:</label> <input type="text" id="customer_id" name="customer_id"><br><br> <label for="call_result">Call Result:</label> <select id="call_result" name="call_result"> <option value="success">Success</option> <option value="failure">Failure</option> </select><br><br> <label for="satisfaction_score">Satisfaction Score (0-10):</label> <input type="number" id="satisfaction_score" name="satisfaction_score" min="0" max="10"><br><br> <button type="submit">Submit</button> </form> ``` 后端接收请求并存入数据库： ```python from flask import request, jsonify from models import CallRecord, db @app.route('/submit_call_record', methods=['POST']) def submit_call_record(): data = request.form record = CallRecord( customer_id=data['customer_id'], call_result=data['call_result'], satisfaction_score=int(data['satisfaction_score']), ) db.session.add(record) db.session.commit() return jsonify({"status": "success", "message": "Record added."}) ``` ###### 批量导入 允许上传 CSV 文件，解析其内容并逐条写入数据库。 ```python import csv from io import StringIO @app.route('/upload_records', methods=['POST']) def upload_records(): file = request.files.get('file') if not file or not file.filename.endswith('.csv'): return jsonify({"error": "Invalid file format"}), 400 stream = StringIO(file.stream.read().decode("UTF8"), newline=None) reader = csv.DictReader(stream) records_to_add = [] errors = [] for row in reader: try: record = CallRecord( customer_id=row['customer_id'], call_result=row['call_result'], satisfaction_score=int(row['satisfaction_score']), ) records_to_add.append(record) except Exception as e: errors.append(f"Row Error: {row}, Reason: {str(e)}") if records_to_add: db.session.bulk_save_objects(records_to_add) db.session.commit() return jsonify({ "status": "partial success" if errors else "success", "errors": errors, "count_added": len(records_to_add), }) ``` --- ##### (2) 查询功能模块 提供灵活的查询接口，支持按多种条件过滤数据。 ```python @app.route('/query_records', methods=['GET']) def query_records(): filters = {} if 'start_date' in request.args and 'end_date' in request.args: start_date = request.args['start_date'] end_date = request.args['end_date'] filters.update({'created_at': {'$gte': start_date, '$lte': end_date}}) results = CallRecord.query.filter_by(**filters).all() serialized_results = [ { "id": r.id, "customer_id": r.customer_id, "call_result": r.call_result, "satisfaction_score": r.satisfaction_score, "created_at": r.created_at.isoformat(), } for r in results ] return jsonify(serialized_results) ``` --- ##### (3) 高频高风险客户判定规则 每月统计符合条件的客户列表，并将其导出为报告文件。 ```sql -- SQL 示例：找出过去一个月内频繁失败或低分的客户 WITH monthly_stats AS ( SELECT customer_id, COUNT(*) AS total_calls, SUM(CASE WHEN call_result = 'failure' THEN 1 ELSE 0 END) AS failure_count, AVG(satisfaction_score) AS avg_satisfaction FROM call_records WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 1 MONTH) GROUP BY customer_id ) SELECT * FROM monthly_stats WHERE failure_count > 3 OR avg_satisfaction < 3; ``` Python 实现： ```python from sqlalchemy.sql import text def get_high_risk_customers(monthly=True): sql_query = """ WITH monthly_stats AS ( SELECT customer_id, COUNT(*) AS total_calls, SUM(CASE WHEN call_result = :fail_status THEN 1 ELSE 0 END) AS failure_count, AVG(satisfaction_score) AS avg_satisfaction FROM call_records WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL :interval MONTH) GROUP BY customer_id ) SELECT * FROM monthly_stats WHERE failure_count > :threshold_failures OR avg_satisfaction < :threshold_satisfaction; """ params = { "fail_status": "failure", "interval": 1 if monthly else 3, "threshold_failures": 3, "threshold_satisfaction": 3, } with db.engine.connect() as conn: result = conn.execute(text(sql_query), params) return [{"customer_id": r["customer_id"], "stats": dict(r.items())} for r in result] ``` 调用示例： ```python high_risk_clients = get_high_risk_customers() print(high_risk_clients[:5]) # 输出前五名高风险客户 ``` --- #### 3. 技术选型说明 - **ORM 工具**: SQLAlchemy 支持复杂的查询语句构建，适合本项目的多表关联需求[^7]。 - **Web 框架**: Flask/FastAPI 易于上手且社区活跃，能满足中小型项目的开发需求[^8]。 - **数据库**: PostgreSQL 提供强大的 JSONB 类型支持，便于存储半结构化数据[^9]。 --- #### 4. 安全性与扩展性考虑 - 对敏感数据加密传输，防止泄露。 - 添加缓存机制减少重复查询压力。 - 将高频访问的数据迁移到 Redis 中提高响应速度。 --- ###