WebDemo/utils/report/report_generator.py

import json
import jinja2
import pandas as pd
from datetime import datetime
from pathlib import Path
from core.config.config_manager import ConfigManager
from core.utils.logger import Logger
from core.singleton import Singleton


class ReportGenerator(metaclass=Singleton):
    def __init__(self):
        self.config = ConfigManager()
        self.logger = Logger.get_logger()
        self.template_loader = jinja2.FileSystemLoader(
            searchpath=Path(__file__).parent.parent.parent / 'resources' / 'templates'
        )
        self.template_env = jinja2.Environment(loader=self.template_loader)
        self.report_dir = Path(__file__).parent.parent.parent / 'resources' / 'reports'
        self.report_dir.mkdir(exist_ok=True)

    def generate_test_report(self, test_results, report_type='html', report_name=None):
        """生成测试报告"""
        if report_type == 'html':
            return self.generate_html_report(test_results, report_name)
        elif report_type == 'json':
            return self.generate_json_report(test_results, report_name)
        elif report_type == 'excel':
            return self.generate_excel_report(test_results, report_name)
        else:
            raise ValueError(f"不支持的报告类型: {report_type}")

    def generate_html_report(self, test_results, report_name=None):
        """生成HTML报告"""
        try:
            template = self.template_env.get_template("report_template.html")

            # 准备报告数据
            report_data = self._prepare_report_data(test_results)

            # 渲染模板
            html_content = template.render(report_data)

            # 生成报告文件名
            if not report_name:
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                report_name = f"test_report_{timestamp}.html"

            # 保存报告
            report_path = self.report_dir / report_name
            with open(report_path, "w", encoding="utf-8") as f:
                f.write(html_content)

            self.logger.info(f"HTML报告已生成: {report_path}")
            return report_path
        except Exception as e:
            self.logger.error(f"生成HTML报告失败: {str(e)}")
            raise

    def generate_json_report(self, test_results, report_name=None):
        """生成JSON报告"""
        try:
            # 准备报告数据
            report_data = self._prepare_report_data(test_results)

            # 生成报告文件名
            if not report_name:
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                report_name = f"test_report_{timestamp}.json"

            # 保存报告
            report_path = self.report_dir / report_name
            with open(report_path, "w", encoding="utf-8") as f:
                json.dump(report_data, f, indent=4, ensure_ascii=False)

            self.logger.info(f"JSON报告已生成: {report_path}")
            return report_path
        except Exception as e:
            self.logger.error(f"生成JSON报告失败: {str(e)}")
            raise

    def generate_excel_report(self, test_results, report_name=None):
        """生成Excel报告"""
        try:
            # 准备数据
            df_data = []
            for result in test_results:
                df_data.append({
                    '测试用例': result['name'],
                    '状态': result['status'],
                    '执行时间(秒)': result['duration'],
                    '开始时间': result['start_time'],
                    '结束时间': result['end_time'],
                    '错误信息': result.get('error', ''),
                    '截图': result.get('screenshot', '')
                })

            df = pd.DataFrame(df_data)

            # 生成报告文件名
            if not report_name:
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                report_name = f"test_report_{timestamp}.xlsx"

            # 保存报告
            report_path = self.report_dir / report_name
            with pd.ExcelWriter(report_path, engine='openpyxl') as writer:
                df.to_excel(writer, sheet_name='测试结果', index=False)

                # 添加摘要信息
                summary_data = {
                    '指标': ['总测试数', '通过数', '失败数', '通过率', '总执行时间'],
                    '值': [
                        len(test_results),
                        sum(1 for r in test_results if r['status'] == 'PASS'),
                        sum(1 for r in test_results if r['status'] == 'FAIL'),
                        f"{(sum(1 for r in test_results if r['status'] == 'PASS') / len(test_results)) * 100 if test_results else 0:.2f}%",
                        f"{sum(r['duration'] for r in test_results):.2f}秒"
                    ]
                }
                summary_df = pd.DataFrame(summary_data)
                summary_df.to_excel(writer, sheet_name='测试摘要', index=False)

            self.logger.info(f"Excel报告已生成: {report_path}")
            return report_path
        except Exception as e:
            self.logger.error(f"生成Excel报告失败: {str(e)}")
            raise

    def _prepare_report_data(self, test_results):
        """准备报告数据"""
        total_duration = sum(r['duration'] for r in test_results)
        passed_count = sum(1 for r in test_results if r['status'] == 'PASS')
        failed_count = sum(1 for r in test_results if r['status'] == 'FAIL')
        success_rate = (passed_count / len(test_results)) * 100 if test_results else 0

        # 按模块分组
        modules = {}
        for result in test_results:
            module_name = result.get('module', '其他')
            if module_name not in modules:
                modules[module_name] = {
                    'total': 0,
                    'passed': 0,
                    'failed': 0,
                    'tests': []
                }

            modules[module_name]['total'] += 1
            if result['status'] == 'PASS':
                modules[module_name]['passed'] += 1
            else:
                modules[module_name]['failed'] += 1
            modules[module_name]['tests'].append(result)

        return {
            "title": "自动化测试报告",
            "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "results": test_results,
            "modules": modules,
            "total_tests": len(test_results),
            "passed": passed_count,
            "failed": failed_count,
            "success_rate": f"{success_rate:.2f}",
            "total_duration": f"{total_duration:.2f}",
            "environment": self.config.get('environment', '未知环境')
        }

    def send_report_by_email(self, report_paths, recipients=None):
        """通过邮件发送报告"""
        from utils.report.email_notifier import EmailNotifier
        email_notifier = EmailNotifier()

        if not recipients:
            recipients = self.config.get('report.email_recipients', [])

        if not recipients:
            self.logger.warning("没有配置邮件接收人，跳过发送邮件")
            return

        subject = f"自动化测试报告 - {datetime.now().strftime('%Y-%m-%d %H:%M')}"
        body = f"自动化测试执行完成，请查看附件中的测试报告。\n\n执行环境: {self.config.get('environment', '未知环境')}"

        try:
            email_notifier.send_email(
                recipients=recipients,
                subject=subject,
                body=body,
                attachments=report_paths
            )
            self.logger.info("测试报告已通过邮件发送")
        except Exception as e:
            self.logger.error(f"发送邮件失败: {str(e)}")