"""Billing data export module""" import concurrent.futures from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional, Tuple, Iterator from ..common.config import ConfigManager from ..common.utils import parse_date_range, format_date_for_display, chunk_list from ..common.cache import CacheManager from ..data.bigquery import BigQueryClient from ..data.csv_handler import CSVHandler from ..data.models import BillingRecord from .mysql_writer import AWSMySQLWriter from .environments import EnvironmentManager from .account_manager import AccountManager class BillingExporter: """账单数据导出器""" def __init__(self, config_path: Optional[str] = None, max_workers: int = 5): """初始化导出器 Args: config_path: 配置文件路径 max_workers: 最大并发工作线程数 """ self.config_manager = ConfigManager(config_path) self.env_manager = EnvironmentManager(self.config_manager) self.account_manager = AccountManager(config_manager=self.config_manager) self.bq_client = BigQueryClient() self.csv_handler = CSVHandler() self.cache_manager = CacheManager() self.max_workers = max_workers self.mysql_writer = AWSMySQLWriter(self.config_manager) # 统计信息 self.stats = { 'total_files': 0, 'skipped_files': 0, 'exported_files': 0, 'total_rows': 0, 'total_cost': 0.0, 'errors': [] } def export(self, environments: List[str] = None, start_date: datetime = None, end_date: datetime = None, days: int = None, last_week: bool = False, account_ids: List[str] = None, overwrite: bool = False, refresh_cache: bool = False) -> Dict: """执行导出任务 Args: environments: 要导出的环境列表 start_date: 开始日期 end_date: 结束日期 days: 最近天数 last_week: 是否导出上周数据 account_ids: 指定账户ID列表 overwrite: 是否覆盖已存在文件 refresh_cache: 是否刷新缓存 Returns: 导出统计信息 """ # 重置统计 self.stats = { 'total_files': 0, 'skipped_files': 0, 'exported_files': 0, 'total_rows': 0, 'total_cost': 0.0, 'errors': [] } # 确定要处理的环境 if not environments: environments = self.env_manager.list_environments() print(f"\n{'='*60}") print(f"AWS 账单数据导出") print(f"{'='*60}") print(f"环境: {', '.join(environments)}") # 如果用户指定了账户ID，按环境归组，避免跨环境重复导出 accounts_by_env = {} if account_ids: accounts_by_env = self._group_requested_accounts(account_ids) # 处理每个环境 for env_name in environments: if not self.env_manager.validate_environment(env_name): continue print(f"\n处理环境: {env_name}") print("-" * 40) # 获取环境配置 env_config = self.env_manager.get_environment(env_name) exclude_days = env_config.get('exclude_recent_days', 2) # 解析日期范围 date_start, date_end = parse_date_range( start_date.strftime('%Y-%m-%d') if start_date else None, end_date.strftime('%Y-%m-%d') if end_date else None, days, last_week, exclude_days ) # 计算实际的安全结束日期 safe_end_date = self._get_safe_end_date(date_end, exclude_days) safe_end_date_obj = datetime.strptime(safe_end_date, '%Y-%m-%d').date() # 如果用户请求的结束日期被调整了，显示提示 if safe_end_date_obj < date_end.date(): print(f"⚠️ 注意: AWS 账单数据有 {exclude_days} 天延迟") print(f" 请求日期范围: {date_start.date()} 至 {date_end.date()}") print(f" 实际查询范围: {date_start.date()} 至 {safe_end_date_obj}") print(f" (只有截至 {safe_end_date_obj} 的数据是准确的)") else: print(f"日期范围: {date_start.date()} 至 {date_end.date()}") # 获取账户列表 if account_ids: if env_name == 'prod': accounts = accounts_by_env.get('prod', []) elif env_name == 'shared': accounts = accounts_by_env.get('shared', []) else: accounts = accounts_by_env.get('nonprod', []) if not accounts: print(f"⚠️ 指定账户未归类到 {env_name} 环境，跳过") continue else: cutover_date = self.config_manager.get( 'settings.awsBillingCutoverDate', '2025-10-01' ) cutover = datetime.strptime(cutover_date, '%Y-%m-%d').date() use_cur2 = date_start.date() >= cutover if use_cur2: cur2_dataset = self.config_manager.get( 'settings.awsBillingNoPayerDataset', 'gcp-digital-bi.aws_billing_no_payer' ) print(f"使用 CUR2 数据集: {cur2_dataset} (按月表 aws_billing_YYYYMM01)") accounts = self.account_manager.get_accounts_by_environment( env_name, refresh_cache, validate_tables=not use_cur2 ) if not accounts: print(f"⚠️ 环境 {env_name} 没有找到账户") continue print(f"账户数量: {len(accounts)}") # 并发处理账户 self._process_accounts_parallel( env_name, accounts, date_start, date_end, overwrite ) # 打印统计信息 self._print_statistics() return self.stats def _group_requested_accounts(self, account_ids: List[str]) -> Dict[str, List[Dict]]: """将指定账户按环境归组，避免跨环境重复导出 Args: account_ids: 用户指定的账户ID列表 Returns: 以环境名称为键的账户列表 """ grouped: Dict[str, List[Dict]] = {} seen = set() for account_id in account_ids or []: if account_id in seen: continue seen.add(account_id) account_info = self.account_manager.get_account_info(account_id) if not account_info: print(f"⚠️ 未在账户详情中找到 {account_id}，将按 nonprod 处理") account_name = account_info.account_name if account_info else account_id classified_env = self.account_manager.classify_account(account_id) if classified_env == 'unknown': print(f"⚠️ 无法确定账户 {account_id} 所属环境，按 nonprod 处理") classified_env = 'nonprod' grouped.setdefault(classified_env, []).append({ 'account_id': account_id, 'account_name': account_name, 'environment': classified_env, 'is_shared': account_info.is_shared if account_info else False, 'dataset_name': account_info.dataset_name if account_info else None, 'table_name': account_info.table_name if account_info else None }) return grouped def _process_accounts_parallel(self, env_name: str, accounts: List[Dict], start_date: datetime, end_date: datetime, overwrite: bool): """并发处理多个账户 Args: env_name: 环境名称 accounts: 账户列表 start_date: 开始日期 end_date: 结束日期 overwrite: 是否覆盖已存在文件 """ with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = [] for account in accounts: future = executor.submit( self._export_account_data, env_name, account, start_date, end_date, overwrite ) futures.append((account['account_id'], future)) # 等待所有任务完成 for account_id, future in futures: try: future.result(timeout=300) except Exception as e: error_msg = f"账户 {account_id} 导出失败: {e}" print(f"✗ {error_msg}") self.stats['errors'].append(error_msg) def _export_account_data(self, env_name: str, account: Dict, start_date: datetime, end_date: datetime, overwrite: bool): """导出单个账户的数据 Args: env_name: 环境名称 account: 账户信息 start_date: 开始日期 end_date: 结束日期 overwrite: 是否覆盖已存在文件 """ account_id = account['account_id'] # 确定账户的实际环境类型 # 如果账户信息中有environment字段，使用它；否则使用传入的env_name actual_env = account.get('environment', env_name) # 按月份分组导出 current_date = start_date while current_date <= end_date: # 计算月份的开始和结束日期 month_start = current_date.replace(day=1) # 计算月份最后一天 if month_start.month == 12: month_end = month_start.replace(year=month_start.year + 1, month=1, day=1) - timedelta(days=1) else: month_end = month_start.replace(month=month_start.month + 1, day=1) - timedelta(days=1) # 调整到实际的日期范围内 export_start = max(month_start, start_date) export_end = min(month_end, end_date) # 导出该月份的数据 account_name = account.get('account_name', account_id) self._export_month_data( actual_env, account_id, account_name, export_start, export_end, overwrite, account, ) # 移动到下个月 current_date = month_end + timedelta(days=1) def _export_month_data(self, env_name: str, account_id: str, account_name: str, start_date: datetime, end_date: datetime, overwrite: bool, account: Optional[Dict] = None): """导出一个月的数据 Args: env_name: 环境名称 account_id: 账户ID start_date: 开始日期 end_date: 结束日期 overwrite: 是否覆盖文件 """ year_month = start_date.strftime('%Y-%m') output_path = self.env_manager.build_output_path(env_name, account_id, year_month) self.stats['total_files'] += 1 dataset_name = account.get('dataset_name') if account else None table_name = account.get('table_name') if account else None env_config = self.env_manager.get_environment(env_name) or {} default_dataset = env_config.get('dataset_name') # 获取新数据 query = self._build_query( env_name, account_id, start_date, end_date, dataset_name=dataset_name, table_name=table_name, ) if not query: return try: # 获取新数据列表 resolved_dataset = self._resolve_dataset_for_date( dataset_name, start_date, default_dataset ) resolved_table = self._resolve_table_for_date( table_name, start_date, account_id ) new_data = list( self._fetch_data_iterator( env_name, account_id, query, resolved_dataset, resolved_table ) ) if not new_data: print(f" 跳过: {output_path.name} (无数据)") self.stats['skipped_files'] += 1 return # 如果文件已存在且不是覆盖模式，则合并数据 final_data = new_data if output_path.exists() and not overwrite: existing_data = self.csv_handler.read_file(str(output_path)) if existing_data: final_data = self._merge_and_deduplicate_data( existing_data, new_data, env_name ) overlap_count = len(existing_data) + len(new_data) - len(final_data) if overlap_count > 0: print(f" 合并: {output_path.name} (覆盖了 {overlap_count} 条重复记录)") else: print(f" 更新: {output_path.name} (新增 {len(new_data)} 条记录)") else: print(f" 创建: {output_path.name}") else: action = "覆盖" if output_path.exists() else "创建" print(f" {action}: {output_path.name}") # 写入合并后的数据 self.csv_handler.write_file(str(output_path), final_data) # 更新统计 self.stats['exported_files'] += 1 self.stats['total_rows'] += len(final_data) print(f" ✓ {output_path.name}: {len(final_data):,} 行") if self.mysql_writer.enabled: self.mysql_writer.sync( account_identifier=account_id, account_name=account_name, environment=env_name, records=final_data, start_date=start_date, end_date=end_date, source_file=output_path ) except Exception as e: error_msg = f"导出失败 {account_id}/{year_month}: {e}" print(f" ✗ {error_msg}") self.stats['errors'].append(error_msg) def _build_query(self, env_name: str, account_id: str, start_date: datetime, end_date: datetime, dataset_name: Optional[str] = None, table_name: Optional[str] = None) -> Optional[str]: """构建查询语句 Args: env_name: 环境名称 account_id: 账户ID start_date: 开始日期 end_date: 结束日期 Returns: SQL查询语句 """ env_config = self.env_manager.get_environment(env_name) if not env_config: return None dataset = self._resolve_dataset_for_date( dataset_name, start_date, env_config.get('dataset_name') ) date_field = env_config['date_field'] cost_field = env_config['cost_field'] product_field = env_config['product_field'] supports_tags = env_config.get('supports_tags', False) is_cur2_dataset = self._is_cur2_dataset(dataset) region_field = 'product_region' if is_cur2_dataset: product_field = self.config_manager.get( 'settings.awsCur2ProductField', 'product_product_family' ) region_field = self.config_manager.get( 'settings.awsCur2RegionField', 'product_region' ) if region_field == 'product_region': region_field = 'product_region_code' partition_field = 'line_item_usage_end_date' if is_cur2_dataset else date_field # 对于共享环境，动态检查表是否真的有标签字段 if env_name == 'shared' and supports_tags and not is_cur2_dataset: table_name_for_check = table_name or f"{account_id}_aws_billing_data" supports_tags = self._verify_table_has_tags(dataset, table_name_for_check) if not supports_tags: print(f"⚠️ 警告：共享账户 {account_id} 的表中未找到tag字段，使用简化查询") else: print(f"✅ 共享账户 {account_id} 支持tag数据，使用完整查询") # 确定表名 if not table_name: table_pattern = env_config.get('table_pattern', '{account_id}_aws_billing_data') table_name = table_pattern.format(account_id=account_id) table_name = self._resolve_table_for_date( table_name, start_date, account_id ) # 计算安全结束日期（考虑数据延迟） exclude_days = env_config.get('exclude_recent_days', 2) safe_end_date = self._get_safe_end_date(end_date, exclude_days) # 如果安全结束日期早于开始日期，返回空查询 if safe_end_date < start_date.strftime('%Y-%m-%d'): return f""" SELECT DATE('1900-01-01') as date, 0.0 as cost, 'NoData' as product, '{account_id}' as account_id WHERE FALSE """ account_filter = "" if is_cur2_dataset: account_filter = f"AND line_item_usage_account_id = '{account_id}'" line_item_filter = self._build_line_item_type_filter() if is_cur2_dataset: if supports_tags: tag_component = ("COALESCE((SELECT kv.value FROM UNNEST(resource_tags.key_value) AS kv " "WHERE kv.key = 'user_component' LIMIT 1), '')") tag_servicetype = ("COALESCE((SELECT kv.value FROM UNNEST(resource_tags.key_value) AS kv " "WHERE kv.key = 'user_servicetype' LIMIT 1), '')") tag_tenant = ("COALESCE((SELECT kv.value FROM UNNEST(resource_tags.key_value) AS kv " "WHERE kv.key = 'user_tenant' LIMIT 1), '')") tag_usedby = ("COALESCE((SELECT kv.value FROM UNNEST(resource_tags.key_value) AS kv " "WHERE kv.key = 'user_usedby' LIMIT 1), '')") base_fields = [ f"DATE({date_field}) as date", f"'{account_id}' as account_id", f"'{env_name}' as environment", f"{product_field} as product", "line_item_line_item_type as line_item_type", f"{tag_component} as tag_component", f"{tag_servicetype} as tag_servicetype", f"COALESCE({region_field}, '') as region", f"{tag_tenant} as tag_tenant", f"{tag_usedby} as tag_usedby", f"{cost_field} as cost" ] group_fields = [ "date", "account_id", "environment", "product", "line_item_type", "tag_component", "tag_servicetype", "region", "tag_tenant", "tag_usedby" ] select_fields = [ "date", "account_id", "environment", "product", "line_item_type", "tag_component", "tag_servicetype", "region", "tag_tenant", "tag_usedby", "ROUND(SUM(cost), 4) as cost" ] else: base_fields = [ f"DATE({date_field}) as date", f"'{account_id}' as account_id", f"'{env_name}' as environment", f"{product_field} as product", "line_item_line_item_type as line_item_type", f"COALESCE({region_field}, '') as region", f"{cost_field} as cost" ] group_fields = ["date", "account_id", "environment", "product", "line_item_type", "region"] select_fields = [ "date", "account_id", "environment", "product", "line_item_type", "region", "ROUND(SUM(cost), 4) as cost" ] query = f""" WITH base AS ( SELECT {','.join(base_fields)} FROM `{dataset}.{table_name}` WHERE DATE({partition_field}) >= '{start_date.strftime('%Y-%m-%d')}' AND DATE({partition_field}) <= '{safe_end_date}' {line_item_filter} AND {cost_field} IS NOT NULL AND {cost_field} != 0 {account_filter} ) SELECT {','.join(select_fields)} FROM base GROUP BY {','.join(group_fields)} HAVING ABS(cost) > 0.001 ORDER BY date DESC, cost DESC """ return query # 构建查询字段和分组字段（CUR 1.0） if supports_tags: # 支持标签的环境：10个字段输出（date, account_id, environment, product, tag_component, tag_servicetype, region, tag_tenant, tag_usedby, cost） tag_component = "COALESCE(resource_tags_user_component, '')" tag_servicetype = "COALESCE(resource_tags_user_servicetype, '')" tag_tenant = "COALESCE(resource_tags_user_tenant, '')" tag_usedby = "COALESCE(resource_tags_user_usedby, '')" select_fields = [ f"DATE({date_field}) as date", f"'{account_id}' as account_id", f"'{env_name}' as environment", f"{product_field} as product", "line_item_line_item_type as line_item_type", f"{tag_component} as tag_component", f"{tag_servicetype} as tag_servicetype", f"COALESCE({region_field}, '') as region", f"{tag_tenant} as tag_tenant", f"{tag_usedby} as tag_usedby", f"ROUND(SUM({cost_field}), 4) as cost" ] group_fields = [ f"DATE({date_field})", f"{product_field}", "line_item_line_item_type", "resource_tags_user_component", "resource_tags_user_servicetype", "product_region", "resource_tags_user_tenant", "resource_tags_user_usedby" ] else: # 非标签环境：增量输出region列（date, account_id, environment, product, region, cost） select_fields = [ f"DATE({date_field}) as date", f"'{account_id}' as account_id", f"'{env_name}' as environment", f"{product_field} as product", "line_item_line_item_type as line_item_type", f"COALESCE({region_field}, '') as region", f"ROUND(SUM({cost_field}), 4) as cost" ] group_fields = [f"DATE({date_field})", f"{product_field}", "line_item_line_item_type", region_field] # 构建查询 query = f""" SELECT {','.join(select_fields)} FROM `{dataset}.{table_name}` WHERE DATE({partition_field}) >= '{start_date.strftime('%Y-%m-%d')}' AND DATE({partition_field}) <= '{safe_end_date}' {line_item_filter} AND {cost_field} IS NOT NULL AND {cost_field} != 0 {account_filter} GROUP BY {','.join(group_fields)} HAVING ABS(cost) > 0.001 ORDER BY DATE({date_field}) DESC, cost DESC """ return query def _fetch_data_iterator(self, env_name: str, account_id: str, query: str, dataset_name: Optional[str] = None, table_name: Optional[str] = None) -> Iterator[Dict]: """获取数据迭代器 Args: env_name: 环境名称 account_id: 账户ID query: SQL查询 Yields: 数据行字典 """ env_config = self.env_manager.get_environment(env_name) supports_tags = env_config.get('supports_tags', False) is_cur2_dataset = self._is_cur2_dataset(dataset_name) # 对于共享环境，需要重新动态检查 if env_name == 'shared' and supports_tags and not is_cur2_dataset: table_name_for_check = table_name or f"{account_id}_aws_billing_data" dataset = dataset_name or env_config['dataset_name'] supports_tags = self._verify_table_has_tags(dataset, table_name_for_check) # 执行查询 for row in self.bq_client.execute_query(query): # 基础输出格式 output_row = { 'date': str(row.get('date', '')), 'account_id': str(row.get('account_id', account_id)), 'environment': env_name, # 使用传入的env_name保证一致性 'product': row.get('product', ''), 'line_item_type': row.get('line_item_type', ''), 'cost': float(row.get('cost', 0)) } # 添加标签字段和扩展字段 if supports_tags: output_row.update({ 'tag_component': row.get('tag_component', ''), 'tag_servicetype': row.get('tag_servicetype', ''), 'region': row.get('region', ''), 'tag_tenant': row.get('tag_tenant', ''), 'tag_usedby': row.get('tag_usedby', '') }) else: output_row['region'] = row.get('region', '') # 更新总成本 self.stats['total_cost'] += output_row['cost'] yield output_row def _resolve_dataset_for_date(self, dataset_name: Optional[str], start_date: datetime, default_dataset: Optional[str]) -> str: cutover_date = self.config_manager.get( 'settings.awsBillingCutoverDate', '2025-10-01' ) cutover = datetime.strptime(cutover_date, '%Y-%m-%d').date() if start_date.date() >= cutover: return self.config_manager.get( 'settings.awsBillingNoPayerDataset', 'gcp-digital-bi.aws_billing_no_payer' ) return dataset_name or default_dataset or self.config_manager.get( 'settings.awsBillingNoPayerDataset', 'gcp-digital-bi.aws_billing_no_payer' ) def _resolve_table_for_date(self, table_name: Optional[str], start_date: datetime, account_id: Optional[str] = None) -> str: cutover_date = self.config_manager.get( 'settings.awsBillingCutoverDate', '2025-10-01' ) cutover = datetime.strptime(cutover_date, '%Y-%m-%d').date() if start_date.date() >= cutover: return f"aws_billing_{start_date.strftime('%Y%m01')}" if table_name: return table_name if account_id: return f"{account_id}_aws_billing_data" return "" def _is_cur2_dataset(self, dataset_name: Optional[str]) -> bool: cur2_dataset = self.config_manager.get( 'settings.awsBillingNoPayerDataset', 'gcp-digital-bi.aws_billing_no_payer' ) return bool(dataset_name) and dataset_name == cur2_dataset def _build_line_item_type_filter(self) -> str: include_types = self.config_manager.get('settings.awsLineItemTypeInclude') or [] exclude_types = self.config_manager.get('settings.awsLineItemTypeExclude') or [] include_types = [str(item).strip() for item in include_types if str(item).strip()] exclude_types = [str(item).strip() for item in exclude_types if str(item).strip()] if include_types: quoted = ", ".join(f"'{item}'" for item in include_types) return f"AND line_item_line_item_type IN ({quoted})" if exclude_types: quoted = ", ".join(f"'{item}'" for item in exclude_types) return f"AND line_item_line_item_type NOT IN ({quoted})" return "AND line_item_line_item_type IS NOT NULL" def _merge_and_deduplicate_data(self, existing_data: List[Dict], new_data: List[Dict], env_name: str) -> List[Dict]: """合并并去重数据，新数据优先 Args: existing_data: 现有数据 new_data: 新数据 env_name: 环境名称 Returns: 合并去重后的数据 """ data_dict = {} # 确定去重键（基于环境确定需要的字段） env_config = self.env_manager.get_environment(env_name) supports_tags = env_config.get('supports_tags', False) if supports_tags: # 支持标签的环境使用更复杂的键 def get_key(x): return ( x.get('date', ''), x.get('account_id', ''), x.get('product', ''), x.get('line_item_type', ''), x.get('tag_component', ''), x.get('tag_servicetype', ''), x.get('region', ''), x.get('tag_tenant', ''), x.get('tag_usedby', '') ) else: # 不支持标签的环境使用简化键 def get_key(x): return ( x.get('date', ''), x.get('account_id', ''), x.get('product', ''), x.get('line_item_type', ''), x.get('region', '') ) # 先处理现有数据 for item in existing_data: # 确保cost字段是数字 if isinstance(item.get('cost'), str): try: item['cost'] = float(item['cost']) except (ValueError, TypeError): item['cost'] = 0.0 key = get_key(item) data_dict[key] = item # 再处理新数据，会覆盖重复的key for item in new_data: key = get_key(item) data_dict[key] = item # 转换回列表并排序 merged_data = list(data_dict.values()) merged_data.sort( key=lambda x: ( x.get('date') or '', x.get('product') or '', x.get('region') or '' ) ) return merged_data def _get_safe_end_date(self, end_date: datetime, exclude_recent_days: int = 1) -> str: """获取考虑延迟的安全结束日期 由于 AWS 账单数据有延迟，需要排除最近几天的数据以确保准确性。 例如，如果今天是 2025-09-01，exclude_recent_days=2， 则只有 2025-08-30 及之前的数据是准确的。 Args: end_date: 原始结束日期 exclude_recent_days: 排除最近天数（账单数据延迟天数） Returns: 安全的结束日期字符串 """ from datetime import date, timedelta if isinstance(end_date, str): end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() elif isinstance(end_date, datetime): end_date_obj = end_date.date() else: end_date_obj = end_date today = date.today() # 计算安全日期：今天减去延迟天数 # 例如：今天是 2025-09-01，延迟2天，安全日期是 2025-08-30 safe_date = today - timedelta(days=exclude_recent_days) # 返回用户请求的结束日期和安全日期中较早的一个 actual_end_date = min(end_date_obj, safe_date) return actual_end_date.strftime('%Y-%m-%d') def _verify_table_has_tags(self, dataset_name: str, table_name: str) -> bool: """验证表是否包含resource_tags字段 Args: dataset_name: 数据集名称 table_name: 表名 Returns: 是否包含标签字段 """ try: check_query = f""" SELECT column_name FROM `{dataset_name}.INFORMATION_SCHEMA.COLUMNS` WHERE table_name = '{table_name}' AND column_name LIKE 'resource_tags_%' LIMIT 1 """ results = list(self.bq_client.execute_query(check_query)) return len(results) > 0 except Exception as e: print(f"检查tag字段时出错: {e}") return False def _print_statistics(self): """打印导出统计信息""" print(f"\n{'='*60}") print("导出统计") print(f"{'='*60}") print(f"总文件数: {self.stats['total_files']}") print(f"已导出: {self.stats['exported_files']}") print(f"已跳过: {self.stats['skipped_files']}") print(f"总行数: {self.stats['total_rows']:,}") print(f"总成本: ${self.stats['total_cost']:,.2f}") if self.stats['errors']: print(f"\n错误 ({len(self.stats['errors'])}):") for error in self.stats['errors'][:5]: # 只显示前5个错误 print(f" - {error}") if len(self.stats['errors']) > 5: print(f" ... 还有 {len(self.stats['errors']) - 5} 个错误")