-de8bd8f33c3e44a59907dafe1884f228.png)
方案一：手动导出 CSV 再利用脚本本地统计
首先在阿里云用户中心按实例、按天导出账单 CSV：
修改以下脚本变量，然后运行：
import decimal
import datetime
def getEveryDay(begin_date, end_date):
date_list = []
begin_date = datetime.datetime.strptime(begin_date, "%Y%m%d")
end_date = datetime.datetime.strptime(end_date, "%Y%m%d")
while begin_date <= end_date:
date_str = begin_date.strftime("%Y%m%d")
date_list.append(date_str)
begin_date += datetime.timedelta(days=1)
return date_list
class Bill:
date = ''
product = ''
consume_type = ''
amount = 0
def __init__(self, date, product, consume_type, amount):
self.date = date
self.product = product
self.consume_type = consume_type
self.amount = amount
def analysis_gen_doc(bill_filepath, start_date, end_date):
bill_lst = []
bill_dict = {}
range_bill_dict = {}
all_day_amount_total = 0
range_day_amount_total = 0
with open(bill_filepath, 'r') as f:
loop_count = 0
for line in f.readlines():
loop_count += 1
if loop_count > 1:
bill_arr = line.split(',')
bill_lst.append(Bill(bill_arr[1], bill_arr[8], bill_arr[10], decimal.Decimal(bill_arr[15])))
for bill in bill_lst:
all_day_amount_total += bill.amount
if bill.date not in bill_dict:
bill_dict[bill.date] = {bill.consume_type: {bill.product: bill.amount}}
else:
if bill.consume_type not in bill_dict[bill.date]:
bill_dict[bill.date][bill.consume_type] = {bill.product: bill.amount}
else:
if bill.product not in bill_dict[bill.date][bill.consume_type]:
bill_dict[bill.date][bill.consume_type][bill.product] = bill.amount
else:
bill_dict[bill.date][bill.consume_type][bill.product] += bill.amount
range_day_lst = getEveryDay(start_date, end_date)
all_day_consume_doc_lst = []
range_day_consume_doc_lst = []
range_group_consume_doc_lst = []
for date, bill_info in bill_dict.items():
all_day_consume_doc_lst.append(date + ': ')
for consume_type, product_info in bill_info.items():
all_day_consume_doc_lst.append('\t{}: '.format(consume_type))
for product_name, amount in product_info.items():
all_day_consume_doc_lst.append('\t\t{}: {}'.format(product_name, round(amount, 2)))
if date in range_day_lst:
range_day_consume_doc_lst.append(date + ': ')
for consume_type, product_info in bill_info.items():
range_day_consume_doc_lst.append('\t{}: '.format(consume_type))
for product_name, amount in product_info.items():
range_day_consume_doc_lst.append('\t\t{}: {}'.format(product_name, round(amount, 2)))
if consume_type not in range_bill_dict:
range_bill_dict[consume_type] = {product_name: amount}
else:
if product_name not in range_bill_dict[consume_type]:
range_bill_dict[consume_type][product_name] = amount
else:
range_bill_dict[consume_type][product_name] += amount
all_day_consume_doc_lst.append('总费用：{}'.format(round(all_day_amount_total, 2)))
for consume_type, product_info in range_bill_dict.items():
range_group_consume_doc_lst.append(consume_type + ': ')
consume_type_amount_total = 0
for product_name, amount in product_info.items():
range_group_consume_doc_lst.append('\t{}: {}'.format(product_name, round(amount, 2)))
range_day_amount_total += amount
consume_type_amount_total += amount
range_group_consume_doc_lst.append('总费用：{}'.format(round(consume_type_amount_total, 2)))
range_day_consume_doc_lst.append('总费用：{}'.format(round(range_day_amount_total, 2)))
# 一个月每天的消费情况
with open('all_day_consume_doc.txt', 'w') as f:
f.write('\n'.join(all_day_consume_doc_lst))
# 时间区间内每天的消费情况
with open('range_day_consume_doc.txt', 'w') as f:
f.write('\n'.join(range_day_consume_doc_lst))
# 时间区间内按消费类型统计的消费情况
with open('range_group_consume_doc.txt', 'w') as f:
f.write('\n'.join(range_group_consume_doc_lst))
if __name__ == '__main__':
# 导出的阿里云账单 csv 文件
bill_filepath = './202111_txy.csv'
# 开始时间
start_date = '20211121'
# 结束时间
end_date = '20211126'
analysis_gen_doc(bill_filepath, start_date, end_date)
会在脚本当前目录生成如下三个文件：
all_day_consume_doc.txt：CSV 包含的所有时间每天的产品消费明细；range_day_consume_doc.txt：选定时间区间的每天的产品消费明细；range_group_consume_doc.txt：选定时间区间的按消费类型聚合的消费统计信息；
比如 all_day_consume_doc.txt 内容格式如下：
...
20211128:
后付费:
CDN: 44.79
大数据开发治理平台 DataWorks: 15.28
块存储: 256.33
云服务器 ECS: 6.47
弹性公网IP: 28.95
数据库自治服务: 26.20
NAT网关: 12.96
云原生大数据计算服务 MaxCompute: 56.69
消息队列 MQ: 569.44
对象存储: 66.46
云原生关系型数据库 PolarDB: 796.06
Prometheus监控服务: 151.67
云解析 PrivateZone: 0.11
负载均衡: 45.60
日志服务: 189.43
应用实时监控服务: 27.20
全站加速: 0.02
数据管理: 6.40
20211129:
预付费:
云原生数据仓库AnalyticDB MySQL版: 3695.80
后付费:
CDN: 57.65
大数据开发治理平台 DataWorks: 137.68
块存储: 256.53
云服务器 ECS: 18.17
弹性公网IP: 54.91
数据库自治服务: 27.30
NAT网关: 13.15
云原生大数据计算服务 MaxCompute: 103.68
消息队列 MQ: 570.20
对象存储: 80.37
云原生关系型数据库 PolarDB: 844.83
Prometheus监控服务: 152.28
云解析 PrivateZone: 0.11
负载均衡: 45.60
日志服务: 201.04
应用实时监控服务: 27.20
全站加速: 0.91
数据管理: 6.40
20211130:
预付费:
数字证书管理服务（原SSL证书）: 1256.30
云服务器 ECS: 59241.60
检索分析服务 Elasticsearch版: 23235.19
云数据库 Redis 版: 4406.40
微服务引擎: 17453.70
关系型数据库: 43574.40
负载均衡: 6201.60
后付费:
CDN: 52.03
大数据开发治理平台 DataWorks: 15.28
块存储: 256.56
云服务器 ECS: 17.96
弹性公网IP: 54.39
数据库自治服务: 28.81
NAT网关: 13.02
云原生大数据计算服务 MaxCompute: 121.97
消息队列 MQ: 586.12
对象存储: 80.58
云原生关系型数据库 PolarDB: 898.60
Prometheus监控服务: 155.04
云解析 PrivateZone: 0.12
负载均衡: 45.60
日志服务: 208.11
应用实时监控服务: 28.00
全站加速: 0.85
数据管理: 6.40
总费用：666659.97
方案二：通过调用阿里云 API 进行统计
阿里云 API 官方文档地址：
这里我比较懒，懒得看其它接口了，所有功能都是基于 DescribeInstanceBill 实现的。
#!/usr/bin/env python
# coding=utf-8
# dependency: pip install aliyun-python-sdk-bssopenapi==2.0.3
import json
import datetime
import calendar
import decimal
from aliyunsdkcore.client import AcsClient
from flask import request
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcore.auth.credentials import AccessKeyCredential
from aliyunsdkcore.auth.credentials import StsTokenCredential
from aliyunsdkbssopenapi.request.v20171214.DescribeInstanceBillRequest import DescribeInstanceBillRequest
def get_billing_cycle_by_billing_date(billing_date):
begin_date = datetime.datetime.strptime(billing_date, "%Y-%m-%d")
date_str = begin_date.strftime("%Y-%m")
return date_str
def get_every_day_by_range(begin_date, end_date):
date_list = []
begin_date = datetime.datetime.strptime(begin_date, "%Y-%m-%d")
end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d")
while begin_date <= end_date:
date_str = begin_date.strftime("%Y-%m-%d")
date_list.append(date_str)
begin_date += datetime.timedelta(days=1)
return date_list
def get_every_day_by_month(month):
year, month = month.split('-')
_, end_day = calendar.monthrange(int(year), int(month))
begin_date = '{}-{}-1'.format(year, month)
end_date = '{}-{}-{}'.format(year, month, end_day)
return get_every_day_by_range(begin_date, end_date)
class Bill:
billing_date = ''
product_name = ''
subscription_type = ''
payment_amount = 0
def __init__(self, billing_date, product_name, subscription_type, payment_amount):
self.billing_date = billing_date
self.product_name = product_name
self.subscription_type = subscription_type
self.payment_amount = payment_amount
def get_bill_page(billing_date, max_results=100, next_token=None):
credentials = AccessKeyCredential(access_key_id, access_key_secret)
# use STS Token
# credentials = StsTokenCredential('<your-access-key-id>', '<your-access-key-secret>', '<your-sts-token>')
client = AcsClient(region_id='cn-shenzhen', credential=credentials)
request = DescribeInstanceBillRequest()
request.set_accept_format('json')
if next_token != None:
request.set_NextToken(next_token)
request.set_MaxResults(max_results)
billing_cycle = get_billing_cycle_by_billing_date(billing_date)
request.set_BillingCycle(billing_cycle)
request.set_BillingDate(billing_date)
request.set_Granularity("DAILY")
response = client.do_action_with_exception(request)
bill_dict = json.loads(str(response, encoding='utf-8'))
return bill_dict
def get_bill_lst_by_day(billing_date):
print('---> 获取 {} 的费用明细 <---'.format(billing_date))
first_loop = True
bill_lst = []
bill_dict = None
while first_loop or (bill_dict.get('Code') == 'Success' and 'NextToken' in bill_dict.get('Data')):
if first_loop:
bill_dict = get_bill_page(billing_date)
first_loop = False
else:
next_token = bill_dict['Data']['NextToken']
bill_dict = get_bill_page(billing_date, next_token=next_token)
if bill_dict.get('Code') != 'Success':
raise Exception("请求失败，错误信息：%s" % bill_dict.get('Message'))
bill_items = bill_dict['Data']['Items']
for bill_item in bill_items:
product_name = bill_item['ProductName']
subscription_type = '预付费' if bill_item['SubscriptionType'] == 'Subscription' else '后付费'
payment_amount = decimal.Decimal(str(bill_item['PaymentAmount']))
bill_lst.append(
Bill(billing_date=billing_date, product_name=product_name, subscription_type=subscription_type,
payment_amount=payment_amount))
return bill_lst
def get_bill_lst_by_day_range(billing_date_begin, billing_date_end):
bill_lst = []
billing_date_lst = get_every_day_by_range(billing_date_begin, billing_date_end)
for billing_date in billing_date_lst:
bill_lst.extend(get_bill_lst_by_day(billing_date))
return bill_lst
def get_bill_lst_by_month(billing_cycle):
bill_lst = []
billing_date_lst = get_every_day_by_month(billing_cycle)
for billing_date in billing_date_lst:
bill_lst.extend(get_bill_lst_by_day(billing_date))
return bill_lst
def gen_doc(billing_date_begin, billing_date_end):
bill_dict = {}
bill_group_dict = {}
amount_total = 0
bill_lst = get_bill_lst_by_day_range(billing_date_begin, billing_date_end)
for bill in bill_lst:
amount_total += bill.payment_amount
if bill.billing_date not in bill_dict:
bill_dict[bill.billing_date] = {bill.subscription_type: {bill.product_name: bill.payment_amount}}
else:
if bill.subscription_type not in bill_dict[bill.billing_date]:
bill_dict[bill.billing_date][bill.subscription_type] = {bill.product_name: bill.payment_amount}
else:
if bill.product_name not in bill_dict[bill.billing_date][bill.subscription_type]:
bill_dict[bill.billing_date][bill.subscription_type][bill.product_name] = bill.payment_amount
else:
bill_dict[bill.billing_date][bill.subscription_type][bill.product_name] += bill.payment_amount
if bill.subscription_type not in bill_group_dict:
bill_group_dict[bill.subscription_type] = {bill.product_name: bill.payment_amount}
else:
if bill.product_name not in bill_group_dict[bill.subscription_type]:
bill_group_dict[bill.subscription_type][bill.product_name] = bill.payment_amount
else:
bill_group_dict[bill.subscription_type][bill.product_name] += bill.payment_amount
consume_doc_lst = []
consume_group_doc_lst = []
for date, bill_info in bill_dict.items():
consume_doc_lst.append(date + ': ')
for consume_type, product_info in bill_info.items():
consume_doc_lst.append('\t{}: '.format(consume_type))
for product_name, amount in product_info.items():
consume_doc_lst.append('\t\t{}: {}'.format(product_name, round(amount, 2)))
consume_doc_lst.append('总费用：{}'.format(round(amount_total, 2)))
amount_total = 0
for subscription_type, product_info in bill_group_dict.items():
consume_group_doc_lst.append('{}({}~{}): '.format(subscription_type, billing_date_begin, billing_date_end))
for product_name, amount in product_info.items():
amount_total += amount
consume_group_doc_lst.append('\t{}: {}'.format(product_name, amount))
consume_group_doc_lst.append('总费用：{}'.format(round(amount_total, 2)))
# 时间区间内每天的消费情况
with open('[{}]-[{}].txt'.format(billing_date_begin, billing_date_end), 'w') as f:
f.write('\n'.join(consume_doc_lst))
# 时间区间内每天的消费情况
with open('[{}]-[{}]-group.txt'.format(billing_date_begin, billing_date_end), 'w') as f:
f.write('\n'.join(consume_group_doc_lst))
if __name__ == '__main__':
# 日期
billing_date_begin = '2021-11-21'
billing_date_end = '2021-11-26'
# 阿里云 AccessKey
access_key_id = 'xxx'
# 阿里云 AccessKeySecret
access_key_secret = 'xxx'
gen_doc(billing_date_begin, billing_date_end)
评论区