excel_exporter.py 6.1 KB
"""
Excel导出模块
将发送记录导出为Excel文件
"""

import pandas as pd
from datetime import datetime
from typing import List, Dict, Optional
from .database import Database


class ExcelExporter:
    """Excel导出器"""

    def __init__(self, database: Database):
        self.database = database

    def export_to_excel(self, file_path: str, start_date: str = None, end_date: str = None) -> bool:
        """
        导出发送记录到Excel文件

        Args:
            file_path: 导出文件路径
            start_date: 开始日期 (YYYY-MM-DD)
            end_date: 结束日期 (YYYY-MM-DD)

        Returns:
            bool: 导出是否成功
        """
        try:
            # 获取数据
            records = self.database.get_sent_messages_for_export(start_date, end_date)

            if not records:
                print("没有找到符合条件的记录")
                return False

            # 转换为DataFrame
            df = pd.DataFrame(records)

            # 数据处理
            if not df.empty:
                # 处理时间格式
                if '发送时间' in df.columns:
                    df['发送时间'] = pd.to_datetime(df['发送时间']).dt.strftime('%Y-%m-%d %H:%M:%S')

                # 处理标签列(如果是JSON格式)
                if '标签' in df.columns:
                    df['标签'] = df['标签'].apply(self._format_tags)

                # 处理触发方式
                if '触发方式' in df.columns:
                    df['触发方式'] = df['触发方式'].apply(
                        lambda x: '对话触发' if x == 'conversation' else '订单触发' if x == 'order' else x
                    )

            # 导出到Excel
            with pd.ExcelWriter(file_path, engine='openpyxl') as writer:
                # 主数据表
                df.to_excel(writer, sheet_name='发送记录', index=False)

                # 统计信息表
                stats_df = self._generate_statistics(records)
                stats_df.to_excel(writer, sheet_name='统计信息', index=False)

                # 格式化Excel
                self._format_excel(writer, df)

            return True

        except Exception as e:
            print(f"Excel导出失败: {e}")
            return False

    def _format_tags(self, tags_json: str) -> str:
        """格式化标签字段"""
        if not tags_json:
            return ""

        try:
            import json
            tags = json.loads(tags_json)
            return ", ".join(tags) if isinstance(tags, list) else str(tags)
        except:
            return str(tags_json)

    def _generate_statistics(self, records: List[Dict]) -> pd.DataFrame:
        """生成统计信息"""
        if not records:
            return pd.DataFrame()

        stats = []

        # 总体统计
        total_count = len(records)
        stats.append({"统计项目": "总发送数量", "数值": total_count})

        # 按触发方式统计
        conversation_count = sum(1 for r in records if r.get('触发方式') == 'conversation')
        order_count = sum(1 for r in records if r.get('触发方式') == 'order')

        stats.append({"统计项目": "对话触发数量", "数值": conversation_count})
        stats.append({"统计项目": "订单触发数量", "数值": order_count})

        # 按日期统计
        date_counts = {}
        for record in records:
            sent_time = record.get('发送时间', '')
            if sent_time:
                try:
                    date = sent_time.split()[0]  # 提取日期部分
                    date_counts[date] = date_counts.get(date, 0) + 1
                except:
                    pass

        # 添加日期统计
        for date, count in sorted(date_counts.items()):
            stats.append({"统计项目": f"{date} 发送数量", "数值": count})

        return pd.DataFrame(stats)

    def _format_excel(self, writer, df: pd.DataFrame):
        """格式化Excel工作表"""
        try:
            from openpyxl.styles import Font, PatternFill, Alignment
            from openpyxl.utils.dataframe import dataframe_to_rows

            # 获取工作表
            worksheet = writer.sheets['发送记录']

            # 设置标题行格式
            header_font = Font(bold=True, color="FFFFFF")
            header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")

            for cell in worksheet[1]:
                cell.font = header_font
                cell.fill = header_fill
                cell.alignment = Alignment(horizontal="center")

            # 调整列宽
            column_widths = {
                'A': 20,  # 对话ID
                'B': 15,  # 客户姓名
                'C': 15,  # 订单ID
                'D': 40,  # 发送内容
                'E': 20,  # 发送时间
                'F': 20,  # 对话创建时间
                'G': 20,  # 最后对话时间
                'H': 20,  # 标签
                'I': 12,  # 触发方式
            }

            for col, width in column_widths.items():
                worksheet.column_dimensions[col].width = width

            # 设置文本换行
            for row in worksheet.iter_rows(min_row=2):
                for cell in row:
                    cell.alignment = Alignment(wrap_text=True, vertical="top")

        except Exception as e:
            print(f"Excel格式化失败: {e}")

    def export_simple_csv(self, file_path: str, start_date: str = None, end_date: str = None) -> bool:
        """
        导出简单的CSV文件(备用方案)

        Args:
            file_path: 导出文件路径
            start_date: 开始日期
            end_date: 结束日期

        Returns:
            bool: 导出是否成功
        """
        try:
            records = self.database.get_sent_messages_for_export(start_date, end_date)

            if not records:
                return False

            df = pd.DataFrame(records)
            df.to_csv(file_path, index=False, encoding='utf-8-sig')
            return True

        except Exception as e:
            print(f"CSV导出失败: {e}")
            return False