141 lines
4.4 KiB
Python
141 lines
4.4 KiB
Python
import mysql.connector
|
|
from mysql.connector import Error
|
|
import os
|
|
from dotenv import load_dotenv
|
|
from typing import List, Dict, Any, Optional
|
|
from ...core.config import AWS_REGION_NAMES_EN, AWS_PRICING_EBS
|
|
|
|
# 加载环境变量
|
|
load_dotenv()
|
|
|
|
# 数据库连接配置
|
|
DB_CONFIG = {
|
|
"host": os.getenv("MYSQL_HOST", "localhost"),
|
|
"user": os.getenv("MYSQL_USER", "root"),
|
|
"password": os.getenv("MYSQL_PASSWORD", ""),
|
|
"database": os.getenv("MYSQL_DATABASE", "aws_pricing")
|
|
}
|
|
|
|
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 search_instances_v2(
|
|
region: Optional[str] = None,
|
|
cpu_cores: Optional[int] = None,
|
|
memory_gb: Optional[float] = None,
|
|
disk_gb: Optional[int] = None,
|
|
operating_system: str = "Linux"
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
从MySQL数据库搜索符合条件的AWS实例
|
|
"""
|
|
try:
|
|
# 获取区域的英文名称
|
|
region_name = None
|
|
if region:
|
|
region_name = AWS_REGION_NAMES_EN.get(region)
|
|
|
|
# 连接到MySQL数据库
|
|
conn = mysql.connector.connect(**DB_CONFIG)
|
|
cursor = conn.cursor(dictionary=True)
|
|
|
|
# 构建SQL查询
|
|
query = """
|
|
SELECT
|
|
id,
|
|
locations,
|
|
area_en,
|
|
area_cn,
|
|
instance_type,
|
|
price,
|
|
operating_system,
|
|
vcpu,
|
|
memory,
|
|
gp3,
|
|
gp2,
|
|
updatetime
|
|
FROM aws_price
|
|
WHERE 1=1
|
|
"""
|
|
params = []
|
|
|
|
# 添加过滤条件
|
|
if region_name:
|
|
query += " AND area_en = %s"
|
|
params.append(region_name)
|
|
|
|
if cpu_cores:
|
|
query += " AND vcpu = %s"
|
|
params.append(cpu_cores)
|
|
|
|
if memory_gb:
|
|
query += " AND memory = %s"
|
|
params.append(memory_gb)
|
|
|
|
if operating_system:
|
|
query += " AND operating_system = %s"
|
|
params.append(operating_system)
|
|
|
|
query += " order by price"
|
|
# 执行查询
|
|
cursor.execute(query, params)
|
|
results = cursor.fetchall()
|
|
|
|
# 处理结果
|
|
instances = []
|
|
for row in results:
|
|
hourly_price = float(row['price'])
|
|
gp3_price = float(row['gp3'])
|
|
monthly_price = hourly_price * 730 # 730小时/月
|
|
|
|
# 计算存储价格
|
|
disk_monthly_price = 0
|
|
if disk_gb and disk_gb > 0:
|
|
# 从区域代码获取区域英文名称，反向查找
|
|
region_code = None
|
|
for code, name in AWS_REGION_NAMES_EN.items():
|
|
if name == row['area_en']:
|
|
region_code = code
|
|
break
|
|
|
|
# disk_monthly_price = await calculate_ebs_price(region_code, disk_gb) if region_code else 0
|
|
disk_monthly_price = gp3_price * disk_gb
|
|
|
|
# 计算总价格
|
|
total_monthly_price = monthly_price + disk_monthly_price
|
|
|
|
instances.append({
|
|
"instance_type": row['instance_type'],
|
|
"description": f"{row['vcpu']}核 {row['memory']}GB {row['instance_type']}",
|
|
"cpu": row['vcpu'],
|
|
"memory": row['memory'],
|
|
"disk_gb": disk_gb if disk_gb else 0,
|
|
"hourly_price": hourly_price,
|
|
"monthly_price": monthly_price,
|
|
"disk_monthly_price": disk_monthly_price,
|
|
"total_monthly_price": total_monthly_price,
|
|
"region": row['area_en'],
|
|
"operating_system": row['operating_system']
|
|
})
|
|
|
|
# 按总价格排序
|
|
instances.sort(key=lambda x: x['total_monthly_price'])
|
|
|
|
# 关闭连接
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
return instances
|
|
except Error as e:
|
|
print(f"MySQL数据库错误: {e}")
|
|
raise Exception(f"数据库查询错误: {e}")
|
|
except Exception as e:
|
|
print(f"搜索实例时出错: {e}")
|
|
raise Exception(f"搜索实例时出错: {e}") |