侧边栏壁纸
博主头像
张种恩的技术小栈博主等级

行动起来,活在当下

  • 累计撰写 748 篇文章
  • 累计创建 65 个标签
  • 累计收到 39 条评论

目 录CONTENT

文章目录

使用 Python 来统计阿里云费用账单信息

zze
zze
2021-12-01 / 0 评论 / 1 点赞 / 883 阅读 / 14288 字

方案一:手动导出 CSV 再利用脚本本地统计

首先在阿里云用户中心按实例、按天导出账单 CSV:
image.png

修改以下脚本变量,然后运行:

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)
1

评论区