如何每隔几分钟获取Twilio账户及子账户的使用数据?
解决方案:Twilio账户近实时使用监控(分钟级粒度)
针对Twilio Usage Records API仅支持日期参数的限制,以下是几种实现近实时(1/10分钟级)监控主账户及子账户使用情况的可行方案:
方案1:轮询Today API + 本地增量计算
这是最直接的落地方案,利用Today资源实时更新的特性,定期调用API并计算两次调用间的差值,得到时间段内的实际使用量。
核心逻辑
- 每隔N分钟(比如1分钟)调用一次主账户或子账户的
/v2/Usage/Records/TodayAPI - 本地存储每次调用返回的各指标总量(如通话时长、短信发送条数)
- 用当前总量减去上一次记录的总量,得到该时间段内的使用增量
代码示例(Python)
import requests import time from datetime import datetime # 基础配置 TWILIO_MAIN_SID = "你的主账户SID" TWILIO_MAIN_TOKEN = "你的主账户Auth Token" SUB_ACCOUNT_LIST = ["子账户SID1", "子账户SID2"] # 需要监控的子账户列表 POLL_INTERVAL = 60 # 轮询间隔(秒) # 存储历史使用数据,避免程序重启丢失可写入文件/数据库 last_usage_cache = {} def get_today_usage(account_sid, auth_token): api_url = f"https://api.twilio.com/2010-04-01/Accounts/{account_sid}/Usage/Records/Today.json" resp = requests.get(api_url, auth=(account_sid, auth_token)) resp.raise_for_status() return resp.json() def calc_time_range_usage(current_data, account_id): increment_result = {} current_records = {item["category"]: float(item["usage"]) for item in current_data["usage_records"]} if account_id not in last_usage_cache: last_usage_cache[account_id] = current_records return increment_result for category, current_val in current_records.items(): last_val = last_usage_cache[account_id].get(category, 0.0) increment_result[category] = round(current_val - last_val, 2) last_usage_cache[account_id] = current_records return increment_result if __name__ == "__main__": # 初始化首次数据 main_init_data = get_today_usage(TWILIO_MAIN_SID, TWILIO_MAIN_TOKEN) last_usage_cache[TWILIO_MAIN_SID] = {r["category"]: float(r["usage"]) for r in main_init_data["usage_records"]} for sub_sid in SUB_ACCOUNT_LIST: sub_init_data = get_today_usage(sub_sid, TWILIO_MAIN_TOKEN) last_usage_cache[sub_sid] = {r["category"]: float(r["usage"]) for r in sub_init_data["usage_records"]} # 持续轮询监控 while True: time.sleep(POLL_INTERVAL) current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # 监控主账户 main_current_data = get_today_usage(TWILIO_MAIN_SID, TWILIO_MAIN_TOKEN) main_increment = calc_time_range_usage(main_current_data, TWILIO_MAIN_SID) print(f"[{current_time}] 主账户({TWILIO_MAIN_SID})近{POLL_INTERVAL}秒使用增量:") for cat, val in main_increment.items(): if val > 0: print(f" {cat}: {val} {main_current_data['usage_records'][0]['unit']}") # 监控子账户 for sub_sid in SUB_ACCOUNT_LIST: sub_current_data = get_today_usage(sub_sid, TWILIO_MAIN_TOKEN) sub_increment = calc_time_range_usage(sub_current_data, sub_sid) print(f"[{current_time}] 子账户({sub_sid})近{POLL_INTERVAL}秒使用增量:") for cat, val in sub_increment.items(): if val > 0: print(f" {cat}: {val} {sub_current_data['usage_records'][0]['unit']}")
Curl示例(单次调用主账户Today资源)
curl -X GET "https://api.twilio.com/2010-04-01/Accounts/你的主账户SID/Usage/Records/Today.json" \ -u "你的主账户SID:你的主账户Auth Token"
方案2:使用Usage Triggers实现阈值告警
如果核心需求是异常使用告警而非持续监控,可配置Twilio Usage Triggers,当特定指标达到阈值时触发Webhook通知:
- 通过API或Twilio控制台创建Trigger,指定监控类别(如
sms-outbound)、阈值(如每分钟10条)和通知URL - 当使用量触发阈值时,Twilio会向你的Webhook发送请求,实现实时告警
代码示例(Python创建Trigger)
from twilio.rest import Client client = Client(TWILIO_MAIN_SID, TWILIO_MAIN_TOKEN) trigger = client.usage.triggers.create( trigger_value=10, usage_category="sms-outbound", callback_url="你的告警Webhook地址", friendly_name="每分钟短信发送量告警" ) print(f"创建的Trigger SID: {trigger.sid}")
方案3:导出详细日志到本地存储(高精度历史查询)
若需要长期保留分钟级使用数据并支持任意时间段查询,可定期导出Twilio的详细使用日志:
- 使用Twilio Export API定期导出小时粒度的详细使用记录,本地过滤后可实现分钟级查询
- 将导出数据存储到自有数据库(如MySQL、PostgreSQL)
- 直接在本地数据库中查询任意时间范围的使用数据
注意事项
- 轮询间隔建议不小于30秒,避免触发Twilio API速率限制
- 子账户监控需确保主账户拥有对应权限,或使用子账户独立的Auth Token
- 本地存储历史数据时,建议写入文件或数据库,避免程序重启后丢失数据
内容的提问来源于stack exchange,提问作者chilli10
相关产品推荐
相关产品推荐

