90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
import boto3
|
|
import json
|
|
from ...core.config import AWS_PRICING_EBS
|
|
from typing import Dict, Any, Optional
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
# 加载环境变量
|
|
load_dotenv()
|
|
|
|
|
|
# 配置boto3使用环境变量
|
|
boto3.setup_default_session(
|
|
aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
|
|
aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
|
|
region_name=os.getenv('AWS_DEFAULT_REGION')
|
|
)
|
|
|
|
async def calculate_ebs_price(region: str, disk_gb: int) -> float:
|
|
"""
|
|
计算EBS存储价格
|
|
"""
|
|
if region in AWS_PRICING_EBS:
|
|
price_dimensions = AWS_PRICING_EBS[region]
|
|
else:
|
|
price_dimensions = 0.1
|
|
|
|
return price_dimensions * disk_gb
|
|
|
|
async def calculate_ec2_price(instance_type: str, region: str, operating_system: str = "Linux",
|
|
disk_gb: int = 0) -> Dict[str, float]:
|
|
"""
|
|
计算EC2实例价格
|
|
"""
|
|
try:
|
|
# 创建Pricing API客户端
|
|
pricing_client = boto3.client('pricing', region_name='us-east-1')
|
|
|
|
# 构建基础过滤器
|
|
filters = [
|
|
{'Type': 'TERM_MATCH', 'Field': 'instanceType', 'Value': instance_type},
|
|
{'Type': 'TERM_MATCH', 'Field': 'regionCode', 'Value': region},
|
|
{'Type': 'TERM_MATCH', 'Field': 'productFamily', 'Value': 'Compute Instance'},
|
|
{'Type': 'TERM_MATCH', 'Field': 'serviceCode', 'Value': 'AmazonEC2'},
|
|
{'Type': 'TERM_MATCH', 'Field': 'tenancy', 'Value': 'Shared'},
|
|
{'Type': 'TERM_MATCH', 'Field': 'operatingSystem', 'Value': operating_system},
|
|
{'Type': 'TERM_MATCH', 'Field': 'preInstalledSw', 'Value': 'NA'},
|
|
{'Type': 'TERM_MATCH', 'Field': 'termType', 'Value': 'OnDemand'},
|
|
{'Type': 'TERM_MATCH', 'Field': 'capacitystatus', 'Value': 'Used'},
|
|
{'Type': 'TERM_MATCH', 'Field': 'currentGeneration', 'Value': 'Yes'}
|
|
]
|
|
|
|
# 获取实例价格
|
|
response = pricing_client.get_products(
|
|
ServiceCode='AmazonEC2',
|
|
Filters=filters,
|
|
MaxResults=1
|
|
)
|
|
|
|
if not response['PriceList']:
|
|
raise Exception(f"未找到实例 {instance_type} 的价格信息")
|
|
|
|
price_list = json.loads(response['PriceList'][0])
|
|
terms = price_list['terms']['OnDemand']
|
|
price_dimensions = list(terms.values())[0]['priceDimensions']
|
|
price_per_hour = float(list(price_dimensions.values())[0]['pricePerUnit']['USD'])
|
|
|
|
# 计算存储价格
|
|
disk_monthly_price = await calculate_ebs_price(region, disk_gb) if disk_gb > 0 else 0
|
|
|
|
# 计算每月价格
|
|
monthly_price = price_per_hour * 730 # 730小时/月
|
|
total_monthly_price = monthly_price + disk_monthly_price
|
|
|
|
return {
|
|
"hourly_price": price_per_hour,
|
|
"monthly_price": monthly_price,
|
|
"disk_monthly_price": disk_monthly_price,
|
|
"total_monthly_price": total_monthly_price
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"计算AWS价格时出错: {str(e)}")
|
|
# 返回默认值或者可以抛出异常
|
|
# return {
|
|
# "hourly_price": 0,
|
|
# "monthly_price": 0,
|
|
# "disk_monthly_price": 0,
|
|
# "total_monthly_price": 0
|
|
# } |