"""Data models for billing records and account information""" from dataclasses import dataclass, field from datetime import datetime from typing import Optional, Dict, Any, List, Tuple @dataclass class BillingRecord: """账单记录数据模型""" date: datetime account_id: str environment: str product: str cost: float region: Optional[str] = None usage_type: Optional[str] = None tag_component: Optional[str] = field(default=None) tag_servicetype: Optional[str] = field(default=None) tag_tenant: Optional[str] = field(default=None) tag_usedby: Optional[str] = field(default=None) tags: Dict[str, str] = field(default_factory=dict) raw_data: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """转换为字典""" return { 'date': self.date.strftime('%Y-%m-%d'), 'account_id': self.account_id, 'environment': self.environment, 'product': self.product, 'cost': self.cost, 'region': self.region, 'usage_type': self.usage_type, 'tag_component': self.tag_component, 'tag_servicetype': self.tag_servicetype, 'tag_tenant': self.tag_tenant, 'tag_usedby': self.tag_usedby, **self.tags } def to_csv_row(self) -> Dict[str, Any]: """转换为CSV行""" row = self.to_dict() # 移除raw_data，CSV不需要 row.pop('raw_data', None) return row @classmethod def from_bigquery_row(cls, row: Dict[str, Any], environment: str, cost_field: str = 'line_item_unblended_cost', product_field: str = 'line_item_product_code', date_field: str = 'line_item_usage_start_date', supports_tags: bool = False) -> 'BillingRecord': """从BigQuery行创建记录 Args: row: BigQuery查询结果行 environment: 环境名称 cost_field: 成本字段名 product_field: 产品字段名 date_field: 日期字段名 supports_tags: 是否支持标签 Returns: BillingRecord实例 """ # 解析日期 date_value = row.get(date_field) if isinstance(date_value, str): record_date = datetime.strptime(date_value, '%Y-%m-%d') else: record_date = date_value # 解析标签 tags = {} tag_component = None tag_servicetype = None tag_tenant = None tag_usedby = None if supports_tags: # 提取特定标签 tag_component = row.get('resource_tags_user_component') tag_servicetype = row.get('resource_tags_user_servicetype') tag_tenant = row.get('resource_tags_user_tenant') tag_usedby = row.get('resource_tags_user_usedby') # 提取所有resource_tags_user_开头的字段 for key, value in row.items(): if key.startswith('resource_tags_user_') and value: tag_name = key.replace('resource_tags_user_', '') tags[f'tag_{tag_name}'] = value return cls( date=record_date, account_id=str(row.get('bill_payer_account_id', '')), environment=environment, product=row.get(product_field, 'Unknown'), cost=float(row.get(cost_field, 0)), region=row.get('product_region'), usage_type=row.get('line_item_usage_type'), tag_component=tag_component, tag_servicetype=tag_servicetype, tag_tenant=tag_tenant, tag_usedby=tag_usedby, tags=tags, raw_data=row ) @dataclass class AccountInfo: """AWS账户信息数据模型""" account_id: str account_name: str environment: str dataset_name: Optional[str] = None table_name: Optional[str] = None is_shared: bool = False payer_account_id: Optional[str] = None status: str = 'active' tags: Dict[str, str] = field(default_factory=dict) metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """转换为字典""" return { 'account_id': self.account_id, 'account_name': self.account_name, 'environment': self.environment, 'dataset_name': self.dataset_name, 'table_name': self.table_name, 'is_shared': self.is_shared, 'payer_account_id': self.payer_account_id, 'status': self.status, 'tags': self.tags, **self.metadata } @classmethod def from_csv_row(cls, row: Dict[str, Any]) -> 'AccountInfo': """从CSV行创建账户信息 Args: row: CSV行数据 Returns: AccountInfo实例 """ # 处理布尔值 shared_value = str(row.get('is_shared_dev_account', '')).strip().lower() is_shared = shared_value in {'yes', 'true', '1', 'y'} return cls( account_id=str(row.get('account_id', '')), account_name=row.get('account_name', ''), environment=row.get('environment', 'unknown'), dataset_name=row.get('dataset_name') or None, table_name=row.get('table_name') or None, is_shared=is_shared, payer_account_id=row.get('payer_account_id'), status=row.get('status', 'active'), metadata={k: v for k, v in row.items() if k not in ['account_id', 'account_name', 'environment', 'dataset_name', 'table_name', 'is_shared_dev_account', 'payer_account_id', 'status']} ) @dataclass class CostSummary: """成本汇总数据模型""" total_cost: float account_costs: Dict[str, float] = field(default_factory=dict) service_costs: Dict[str, float] = field(default_factory=dict) tag_costs: Dict[str, float] = field(default_factory=dict) daily_costs: Dict[str, float] = field(default_factory=dict) def add_record(self, record: BillingRecord): """添加一条记录到汇总 Args: record: 账单记录 """ self.total_cost += record.cost # 按账户汇总 if record.account_id: self.account_costs[record.account_id] = \ self.account_costs.get(record.account_id, 0) + record.cost # 按服务汇总 if record.product: self.service_costs[record.product] = \ self.service_costs.get(record.product, 0) + record.cost # 按标签汇总 if record.tag_usedby: self.tag_costs[record.tag_usedby] = \ self.tag_costs.get(record.tag_usedby, 0) + record.cost # 按日期汇总 date_str = record.date.strftime('%Y-%m-%d') self.daily_costs[date_str] = \ self.daily_costs.get(date_str, 0) + record.cost def get_top_accounts(self, limit: int = 10) -> List[Tuple[str, float]]: """获取成本最高的账户 Args: limit: 返回数量限制 Returns: [(账户ID, 成本)]列表 """ sorted_accounts = sorted( self.account_costs.items(), key=lambda x: x[1], reverse=True ) return sorted_accounts[:limit] def get_top_services(self, limit: int = 10) -> List[Tuple[str, float]]: """获取成本最高的服务 Args: limit: 返回数量限制 Returns: [(服务名, 成本)]列表 """ sorted_services = sorted( self.service_costs.items(), key=lambda x: x[1], reverse=True ) return sorted_services[:limit]