方案概述
- 使用Twilio的API进行电话呼叫
- 实现基本的呼叫逻辑
- 添加简单的用户界面
实现代码
1. 安装必要的库
pip install twilio flask
2. 主程序代码
from twilio.rest import Client
from flask import Flask, request, jsonify, render_template
import time
import threading
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Twilio账户信息 - 请替换为你的实际账户信息
ACCOUNT_SID = 'your_account_sid'
AUTH_TOKEN = 'your_auth_token'
TWILIO_PHONE_NUMBER = '+1234567890' # 你的Twilio电话号码
# 初始化Twilio客户端
client = Client(ACCOUNT_SID, AUTH_TOKEN)
app = Flask(__name__)
# 存储呼叫任务
call_tasks = {}
class CallTask:
def __init__(self, phone_number, message, call_time=None):
self.phone_number = phone_number
self.message = message
self.status = "pending"
self.call_time = call_time or datetime.now()
self.call_sid = None
self.start_time = None
self.end_time = None
def start_call(self):
try:
self.status = "calling"
self.start_time = datetime.now()
# 使用Twilio发起呼叫
call = client.calls.create(
url='http://demo.twilio.com/docs/voice.xml', # 这里使用Twilio的示例，实际应替换为你的TwiML
to=self.phone_number,
from_=TWILIO_PHONE_NUMBER,
record=True
)
self.call_sid = call.sid
logger.info(f"Call started to {self.phone_number}, SID: {self.call_sid}")
# 检查呼叫状态
self.monitor_call()
except Exception as e:
self.status = "failed"
logger.error(f"Failed to start call: {str(e)}")
def monitor_call):
"""监控呼叫状态"""
while True:
call = client.calls(self.call_sid).fetch()
if call.status in ['completed', 'failed', 'busy', 'no-answer']:
self.status = call.status
self.end_time = datetime.now()
logger.info(f"Call ended with status: {call.status}")
break
time.sleep(5)
@app.route('/')
def index():
"""显示主界面"""
return render_template('index.html')
@app.route('/make_call', methods=['POST'])
def make_call():
"""发起呼叫的API接口"""
data = request.json
phone_number = data.get('phone_number')
message = data.get('message', '')
if not phone_number:
return jsonify({'error': 'Phone number is required'}), 400
# 创建呼叫任务
task_id = str(int(time.time()))
call_task = CallTask(phone_number, message)
call_tasks[task_id] = call_task
# 在新线程中启动呼叫
threading.Thread(target=call_task.start_call).start()
return jsonify({
'task_id': task_id,
'status': 'queued',
'phone_number': phone_number
})
@app.route('/call_status/<task_id>')
def call_status(task_id):
"""获取呼叫状态"""
call_task = call_tasks.get(task_id)
if not call_task:
return jsonify({'error': 'Task not found'}), 404
return jsonify({
'task_id': task_id,
'status': call_task.status,
'phone_number': call_task.phone_number,
'start_time': str(call_task.start_time) if call_task.start_time else None,
'end_time': str(call_task.end_time) if call_task.end_time else None,
'call_sid': call_task.call_sid
})
@app.route('/call_history')
def call_history():
"""获取呼叫历史"""
calls = client.calls.list(limit=20)
call_history = []
for call in calls:
call_history.append({
'sid': call.sid,
'from': call.from_formatted,
'to': call.to_formatted,
'status': call.status,
'start_time': str(call.start_time),
'duration': call.duration
})
return jsonify(call_history)
if __name__ == '__main__':
app.run(debug=True)
3. HTML模板 (templates/index.html)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>自动电话呼叫系统</title>
<style>
body {
font-family: Arial, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
}
.form-group {
margin-bottom: 15px;
}
label {
display: block;
margin-bottom: 5px;
font-weight: bold;
}
input, textarea {
width: 100%;
padding: 8px;
box-sizing: border-box;
}
button {
background-color: #4CAF50;
color: white;
padding: 10px 15px;
border: none;
cursor: pointer;
}
button:hover {
background-color: #45a049;
}
#status {
margin-top: 20px;
padding: 10px;
border: 1px solid #ddd;
}
.call-item {
border-bottom: 1px solid #eee;
padding: 10px 0;
}
</style>
</head>
<body>
<h1>自动电话呼叫系统</h1>
<div class="form-group">
<label for="phone_number">电话号码:</label>
<input type="text" id="phone_number" placeholder="输入电话号码，包括国家代码，例如 +8613800138000">
</div>
<div class="form-group">
<label for="message">消息内容 (可选):</label>
<textarea id="message" rows="4" placeholder="输入要播放的消息内容"></textarea>
</div>
<button id="call_button">发起呼叫</button>
<div id="status"></div>
<h2>呼叫历史</h2>
<div id="call_history"></div>
<script>
document.getElementById('call_button').addEventListener('click', async () => {
const phoneNumber = document.getElementById('phone_number').value;
const message = document.getElementById('message').value;
if (!phoneNumber) {
alert('请输入电话号码');
return;
}
const statusDiv = document.getElementById('status');
statusDiv.innerHTML = '正在发起呼叫...';
try {
const response = await fetch('/make_call', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
phone_number: phoneNumber,
message: message
})
});
const data = await response.json();
if (response.ok) {
statusDiv.innerHTML = `呼叫已排队，任务ID: ${data.task_id}`;
// 轮询检查呼叫状态
checkCallStatus(data.task_id);
} else {
statusDiv.innerHTML = `错误: ${data.error}`;
}
} catch (error) {
statusDiv.innerHTML = `请求失败: ${error.message}`;
}
});
async function checkCallStatus(taskId) {
const statusDiv = document.getElementById('status');
try {
const response = await fetch(`/call_status/${taskId}`);
const data = await response.json();
if (response.ok) {
statusDiv.innerHTML = `呼叫状态: ${data.status}<br>
电话号码: ${data.phone_number}<br>
开始时间: ${data.start_time || '未开始'}<br>
结束时间: ${data.end_time || '未结束'}`;
// 如果呼叫未完成，继续轮询
if (data.status === 'pending' || data.status === 'calling') {
setTimeout(() => checkCallStatus(taskId), 2000);
}
} else {
statusDiv.innerHTML += `<br>获取状态失败: ${data.error}`;
}
} catch (error) {
statusDiv.innerHTML += `<br>获取状态请求失败: ${error.message}`;
}
}
// 加载呼叫历史
async function loadCallHistory() {
const historyDiv = document.getElementById('call_history');
try {
const response = await fetch('/call_history');
const data = await response.json();
if (response.ok) {
if (data.length === 0) {
historyDiv.innerHTML = '<p>没有呼叫记录</p>';
return;
}
let html = '';
data.forEach(call => {
html += `
<div class="call-item">
<strong>${call.from} → ${call.to}</strong><br>
状态: ${call.status}<br>
时间: ${call.start_time}<br>
时长: ${call.duration}秒
</div>
`;
});
historyDiv.innerHTML = html;
} else {
historyDiv.innerHTML = '<p>加载历史记录失败</p>';
}
} catch (error) {
historyDiv.innerHTML = '<p>加载历史记录请求失败</p>';
}
}
// 页面加载时获取呼叫历史
window.addEventListener('load', loadCallHistory);
</script>
</body>
</html>
功能说明
发起呼叫：
- 输入电话号码和可选消息内容
- 点击按钮发起呼叫
- 系统会返回任务ID并显示呼叫状态
状态监控：
- 实时显示呼叫状态（pending/calling/completed/failed等）
- 显示呼叫开始和结束时间
呼叫历史：
- 显示最近的20条呼叫记录
- 包括呼叫状态、时长等信息
部署说明
- 注册Twilio账号并获取ACCOUNT_SID和AUTH_TOKEN
- 购买Twilio电话号码并替换代码中的TWILIO_PHONE_NUMBER
- 创建TwiML应用或使用Twilio Studio定义呼叫流程
- 运行Flask应用：
python app.py
扩展功能建议
- 批量呼叫：添加CSV导入功能，支持批量呼叫
- 语音合成：集成TTS服务，动态生成语音内容
- 呼叫转移：实现IVR菜单和呼叫转移功能
- 数据库集成：使用数据库存储呼叫记录
- 认证系统：添加用户认证和权限管理
注意事项
- 使用Twilio等服务需要遵守相关法律法规
- 自动呼叫系统可能受到不同国家/地区的法律限制
- 商业使用需要考虑服务费用和通话质量
- 需要处理各种异常情况（网络问题、账户余额不足等）
