在这里插入图片描述

文章目录

- 方案概述
- 实现代码
- - 1. 安装必要的库
  - 2. 主程序代码
  - 3. HTML模板 (templates/index.html)
- 功能说明
- 部署说明
- 扩展功能建议
- 注意事项

方案概述

使用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_numberself.message = messageself.status = "pending"self.call_time = call_time or datetime.now()self.call_sid = Noneself.start_time = Noneself.end_time = Nonedef 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的示例，实际应替换为你的TwiMLto=self.phone_number,from_=TWILIO_PHONE_NUMBER,record=True)self.call_sid = call.sidlogger.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.statusself.end_time = datetime.now()logger.info(f"Call ended with status: {call.status}")breaktime.sleep(5)@app.route('/')
def index():"""显示主界面"""return render_template('index.html')@app.route('/make_call', methods=['POST'])
def make_call():"""发起呼叫的API接口"""data = request.jsonphone_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'}), 404return 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>