ewoooc/services/ai_history_service.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI 生成歷史記錄服務
提供 AI 生成結果的 CRUD 操作
"""

import json
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
from sqlalchemy import desc, func, or_
from database.manager import get_session
from database.ai_models import AIGenerationHistory, AIPromptTemplate

logger = logging.getLogger(__name__)


class AIHistoryService:
    """AI 生成歷史記錄服務"""

    @staticmethod
    def save_generation(
        generation_type: str,
        output_content: str,
        product_name: str = None,
        input_keywords: List[str] = None,
        input_style: str = None,
        input_trend_topic: str = None,
        model_name: str = None,
        generation_duration: float = None,
        created_by: int = None,
        ai_provider: str = 'ollama',
        input_tokens: int = 0,
        output_tokens: int = 0
    ) -> Optional[int]:
        """
        儲存 AI 生成結果

        Args:
            generation_type: 生成類型 (copy/recommend/weather_analysis)
            output_content: 生成的內容
            product_name: 商品名稱
            input_keywords: 輸入的關鍵字列表
            input_style: 文案風格
            input_trend_topic: 趨勢話題
            model_name: 使用的模型名稱
            generation_duration: 生成耗時（秒）
            created_by: 創建者用戶 ID
            ai_provider: AI 提供者 ('ollama' 或 'gemini')
            input_tokens: 輸入 token 數量
            output_tokens: 輸出 token 數量

        Returns:
            新建記錄的 ID，失敗則返回 None
        """
        session = get_session()
        try:
            history = AIGenerationHistory(
                generation_type=generation_type,
                product_name=product_name,
                input_keywords=AIHistoryService._safe_json_dumps(input_keywords),
                input_style=input_style,
                input_trend_topic=input_trend_topic,
                output_content=output_content,
                model_name=model_name,
                generation_duration=generation_duration,
                created_by=created_by,
                ai_provider=ai_provider,
                input_tokens=input_tokens,
                output_tokens=output_tokens
            )
            session.add(history)
            session.commit()
            logger.info(f"AI 生成記錄已儲存 | ID: {history.id} | Type: {generation_type} | Provider: {ai_provider}")
            return history.id
        except Exception as e:
            session.rollback()
            logger.error(f"儲存 AI 生成記錄失敗: {e}")
            return None
        finally:
            session.close()

    @staticmethod
    def _safe_json_dumps(data) -> Optional[str]:
        """安全的 JSON 序列化"""
        if data is None:
            return None
        try:
            return json.dumps(data, ensure_ascii=False)
        except (TypeError, ValueError) as e:
            logger.warning(f"JSON 序列化失敗，使用字串轉換: {e}")
            return json.dumps(str(data), ensure_ascii=False)

    @staticmethod
    def get_history_list(
        page: int = 1,
        per_page: int = 20,
        generation_type: str = None,
        search: str = None,
        is_favorite: bool = None,
        start_date: datetime = None,
        end_date: datetime = None,
        created_by: int = None
    ) -> Dict[str, Any]:
        """
        獲取生成歷史列表（分頁）

        Args:
            page: 頁碼（從 1 開始）
            per_page: 每頁數量
            generation_type: 篩選生成類型
            search: 搜尋關鍵字（商品名稱或內容）
            is_favorite: 篩選收藏
            start_date: 開始日期
            end_date: 結束日期
            created_by: 創建者用戶 ID

        Returns:
            包含 items, total, page, per_page, total_pages 的字典
        """
        # 輸入驗證
        page = max(1, page)
        per_page = max(1, min(100, per_page))  # 限制每頁最多 100 筆

        session = get_session()
        try:
            query = session.query(AIGenerationHistory)

            # 應用篩選條件
            if generation_type:
                query = query.filter(AIGenerationHistory.generation_type == generation_type)

            if search:
                search_pattern = f"%{search}%"
                query = query.filter(
                    or_(
                        AIGenerationHistory.product_name.like(search_pattern),
                        AIGenerationHistory.output_content.like(search_pattern),
                        AIGenerationHistory.input_trend_topic.like(search_pattern)
                    )
                )

            if is_favorite is not None:
                query = query.filter(AIGenerationHistory.is_favorite.is_(is_favorite))

            if start_date:
                query = query.filter(AIGenerationHistory.created_at >= start_date)

            if end_date:
                query = query.filter(AIGenerationHistory.created_at <= end_date)

            if created_by:
                query = query.filter(AIGenerationHistory.created_by == created_by)

            # 計算總數
            total = query.count()

            # 排序和分頁
            query = query.order_by(desc(AIGenerationHistory.created_at))
            offset = (page - 1) * per_page
            items = query.offset(offset).limit(per_page).all()

            return {
                'items': [item.to_dict() for item in items],
                'total': total,
                'page': page,
                'per_page': per_page,
                'total_pages': (total + per_page - 1) // per_page
            }

        except Exception as e:
            logger.error(f"獲取 AI 生成歷史列表失敗: {e}")
            return {
                'items': [],
                'total': 0,
                'page': page,
                'per_page': per_page,
                'total_pages': 0
            }
        finally:
            session.close()

    @staticmethod
    def get_by_id(history_id: int) -> Optional[Dict[str, Any]]:
        """
        根據 ID 獲取單筆記錄

        Args:
            history_id: 記錄 ID

        Returns:
            記錄字典，不存在則返回 None
        """
        session = get_session()
        try:
            history = session.query(AIGenerationHistory).filter_by(id=history_id).first()
            return history.to_dict() if history else None
        except Exception as e:
            logger.error(f"獲取 AI 生成記錄失敗: {e}")
            return None
        finally:
            session.close()

    @staticmethod
    def update(
        history_id: int,
        output_content: str = None,
        rating: int = None,
        is_favorite: bool = None,
        is_used: bool = None,
        notes: str = None
    ) -> bool:
        """
        更新生成記錄

        Args:
            history_id: 記錄 ID
            output_content: 更新後的內容
            rating: 評分 (1-5)
            is_favorite: 是否收藏
            is_used: 是否已使用
            notes: 備註

        Returns:
            是否更新成功
        """
        session = get_session()
        try:
            history = session.query(AIGenerationHistory).filter_by(id=history_id).first()
            if not history:
                return False

            if output_content is not None:
                history.output_content = output_content
            if rating is not None:
                history.rating = max(1, min(5, rating))  # 限制 1-5
            if is_favorite is not None:
                history.is_favorite = is_favorite
            if is_used is not None:
                history.is_used = is_used
            if notes is not None:
                history.notes = notes

            history.updated_at = datetime.now()
            session.commit()
            logger.info(f"AI 生成記錄已更新 | ID: {history_id}")
            return True

        except Exception as e:
            session.rollback()
            logger.error(f"更新 AI 生成記錄失敗: {e}")
            return False
        finally:
            session.close()

    @staticmethod
    def delete(history_id: int) -> bool:
        """
        刪除生成記錄

        Args:
            history_id: 記錄 ID

        Returns:
            是否刪除成功
        """
        session = get_session()
        try:
            history = session.query(AIGenerationHistory).filter_by(id=history_id).first()
            if not history:
                return False

            session.delete(history)
            session.commit()
            logger.info(f"AI 生成記錄已刪除 | ID: {history_id}")
            return True

        except Exception as e:
            session.rollback()
            logger.error(f"刪除 AI 生成記錄失敗: {e}")
            return False
        finally:
            session.close()

    @staticmethod
    def batch_delete(history_ids: List[int]) -> Dict[str, Any]:
        """
        批次刪除生成記錄（優化版本：使用批次刪除）

        Args:
            history_ids: 記錄 ID 列表

        Returns:
            包含 success_count 和 failed_ids 的字典
        """
        if not history_ids:
            return {'success_count': 0, 'failed_ids': []}

        session = get_session()
        try:
            # 先查詢存在的 ID
            existing_ids = session.query(AIGenerationHistory.id).filter(
                AIGenerationHistory.id.in_(history_ids)
            ).all()
            existing_id_set = {row[0] for row in existing_ids}

            # 找出不存在的 ID
            failed_ids = [hid for hid in history_ids if hid not in existing_id_set]

            # 批次刪除存在的記錄
            if existing_id_set:
                deleted_count = session.query(AIGenerationHistory).filter(
                    AIGenerationHistory.id.in_(existing_id_set)
                ).delete(synchronize_session=False)
                session.commit()
            else:
                deleted_count = 0

            logger.info(f"批次刪除 AI 生成記錄 | 成功: {deleted_count} | 失敗: {len(failed_ids)}")
            return {
                'success_count': deleted_count,
                'failed_ids': failed_ids
            }

        except Exception as e:
            session.rollback()
            logger.error(f"批次刪除 AI 生成記錄失敗: {e}")
            return {
                'success_count': 0,
                'failed_ids': list(history_ids)
            }
        finally:
            session.close()

    @staticmethod
    def toggle_favorite(history_id: int) -> Optional[bool]:
        """
        切換收藏狀態

        Args:
            history_id: 記錄 ID

        Returns:
            新的收藏狀態，失敗則返回 None
        """
        session = get_session()
        try:
            history = session.query(AIGenerationHistory).filter_by(id=history_id).first()
            if not history:
                return None

            history.is_favorite = not history.is_favorite
            history.updated_at = datetime.now()
            session.commit()
            return history.is_favorite

        except Exception as e:
            session.rollback()
            logger.error(f"切換收藏狀態失敗: {e}")
            return None
        finally:
            session.close()

    @staticmethod
    def get_statistics(days: int = 30, created_by: int = None) -> Dict[str, Any]:
        """
        獲取生成統計資料

        Args:
            days: 統計天數
            created_by: 用戶 ID（可選）

        Returns:
            統計資料字典
        """
        # 輸入驗證
        days = max(1, min(365, days))  # 限制 1-365 天

        session = get_session()
        try:
            start_date = datetime.now() - timedelta(days=days)

            # 建立基礎篩選條件（避免重複使用 query 物件）
            def build_base_filters():
                filters = [AIGenerationHistory.created_at >= start_date]
                if created_by:
                    filters.append(AIGenerationHistory.created_by == created_by)
                return filters

            base_filters = build_base_filters()

            # 總數
            total_count = session.query(AIGenerationHistory).filter(*base_filters).count()

            # 按類型統計
            type_stats = session.query(
                AIGenerationHistory.generation_type,
                func.count(AIGenerationHistory.id)
            ).filter(*base_filters).group_by(AIGenerationHistory.generation_type).all()

            # 收藏數（獨立查詢，避免污染）
            favorite_count = session.query(AIGenerationHistory).filter(
                *base_filters,
                AIGenerationHistory.is_favorite.is_(True)
            ).count()

            # 已使用數（獨立查詢，避免污染）
            used_count = session.query(AIGenerationHistory).filter(
                *base_filters,
                AIGenerationHistory.is_used.is_(True)
            ).count()

            # 平均生成時間
            avg_duration = session.query(
                func.avg(AIGenerationHistory.generation_duration)
            ).filter(
                *base_filters,
                AIGenerationHistory.generation_duration.isnot(None)
            ).scalar() or 0

            # 熱門商品
            popular_products = session.query(
                AIGenerationHistory.product_name,
                func.count(AIGenerationHistory.id).label('count')
            ).filter(
                *base_filters,
                AIGenerationHistory.product_name.isnot(None)
            ).group_by(
                AIGenerationHistory.product_name
            ).order_by(desc('count')).limit(10).all()

            return {
                'total_count': total_count,
                'type_stats': {t: c for t, c in type_stats},
                'favorite_count': favorite_count,
                'used_count': used_count,
                'avg_duration': round(avg_duration, 2),
                'popular_products': [{'name': p, 'count': c} for p, c in popular_products],
                'period_days': days
            }

        except Exception as e:
            logger.error(f"獲取 AI 統計資料失敗: {e}")
            return {
                'total_count': 0,
                'type_stats': {},
                'favorite_count': 0,
                'used_count': 0,
                'avg_duration': 0,
                'popular_products': [],
                'period_days': days
            }
        finally:
            session.close()


class AITemplateService:
    """AI 提示模板服務"""

    @staticmethod
    def get_all_templates(template_type: str = None, active_only: bool = True) -> List[Dict[str, Any]]:
        """
        獲取所有模板

        Args:
            template_type: 篩選類型
            active_only: 是否只返回啟用的模板

        Returns:
            模板列表
        """
        session = get_session()
        try:
            query = session.query(AIPromptTemplate)

            if template_type:
                query = query.filter(AIPromptTemplate.template_type == template_type)

            if active_only:
                query = query.filter(AIPromptTemplate.is_active.is_(True))

            templates = query.order_by(AIPromptTemplate.is_system.desc(), AIPromptTemplate.name).all()
            return [t.to_dict() for t in templates]

        except Exception as e:
            logger.error(f"獲取 AI 模板列表失敗: {e}")
            return []
        finally:
            session.close()

    @staticmethod
    def get_by_id(template_id: int) -> Optional[Dict[str, Any]]:
        """根據 ID 獲取模板"""
        session = get_session()
        try:
            template = session.query(AIPromptTemplate).filter_by(id=template_id).first()
            return template.to_dict() if template else None
        except Exception as e:
            logger.error(f"獲取 AI 模板失敗: {e}")
            return None
        finally:
            session.close()

    @staticmethod
    def create_template(
        name: str,
        template_type: str,
        user_prompt_template: str,
        description: str = None,
        system_prompt: str = None,
        default_temperature: float = 0.7,
        default_style: str = None,
        created_by: int = None
    ) -> Optional[int]:
        """創建新模板"""
        session = get_session()
        try:
            template = AIPromptTemplate(
                name=name,
                description=description,
                template_type=template_type,
                system_prompt=system_prompt,
                user_prompt_template=user_prompt_template,
                default_temperature=default_temperature,
                default_style=default_style,
                is_system=False,
                created_by=created_by
            )
            session.add(template)
            session.commit()
            return template.id
        except Exception as e:
            session.rollback()
            logger.error(f"創建 AI 模板失敗: {e}")
            return None
        finally:
            session.close()

    @staticmethod
    def update_template(template_id: int, **kwargs) -> bool:
        """更新模板"""
        session = get_session()
        try:
            template = session.query(AIPromptTemplate).filter_by(id=template_id).first()
            if not template:
                return False

            # 不允許修改系統模板的核心內容
            if template.is_system:
                allowed_fields = ['is_active']
            else:
                allowed_fields = ['name', 'description', 'system_prompt', 'user_prompt_template',
                                 'default_temperature', 'default_style', 'is_active']

            for field in allowed_fields:
                if field in kwargs and kwargs[field] is not None:
                    setattr(template, field, kwargs[field])

            template.updated_at = datetime.now()
            session.commit()
            return True

        except Exception as e:
            session.rollback()
            logger.error(f"更新 AI 模板失敗: {e}")
            return False
        finally:
            session.close()

    @staticmethod
    def delete_template(template_id: int) -> bool:
        """刪除模板（不能刪除系統模板）"""
        session = get_session()
        try:
            template = session.query(AIPromptTemplate).filter_by(id=template_id).first()
            if not template or template.is_system:
                return False

            session.delete(template)
            session.commit()
            return True

        except Exception as e:
            session.rollback()
            logger.error(f"刪除 AI 模板失敗: {e}")
            return False
        finally:
            session.close()


# 建立全域服務實例
ai_history_service = AIHistoryService()
ai_template_service = AITemplateService()