#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Google Gemini AI 服務模組 負責與 Google Gemini API 互動，提供文案生成、關鍵字提取等功能 支援 gemini-1.5-flash, gemini-2.5-flash, gemini-2.5-pro 模型 """ import os import time import logging from typing import Optional, Dict, Any, List from dataclasses import dataclass, field from datetime import date from services.gemini_guard import ( gemini_disabled_message, get_gemini_api_key, is_gemini_fallback_enabled, ) logger = logging.getLogger(__name__) # Gemini 設定 - 支援環境變數覆蓋；API key 一律透過 gemini_guard 取得 DEFAULT_GEMINI_MODEL = os.getenv('GEMINI_MODEL', 'gemini-1.5-flash') GEMINI_TIMEOUT = int(os.getenv('GEMINI_TIMEOUT', '60')) # 秒 # Gemini 定價 (USD per 1M tokens) - 2024/2025 定價 GEMINI_PRICING = { 'gemini-1.5-flash': {'input': 0.075, 'output': 0.30}, 'gemini-1.5-flash-latest': {'input': 0.075, 'output': 0.30}, 'gemini-1.5-pro': {'input': 1.25, 'output': 5.00}, 'gemini-1.5-pro-latest': {'input': 1.25, 'output': 5.00}, 'gemini-2.5-flash': {'input': 0.10, 'output': 0.40}, 'gemini-2.5-flash-preview-05-20': {'input': 0.10, 'output': 0.40}, 'gemini-2.5-pro': {'input': 1.25, 'output': 10.00}, 'gemini-2.5-pro-preview-05-06': {'input': 1.25, 'output': 10.00}, } # 可選模型清單（供 UI 選擇） AVAILABLE_GEMINI_MODELS = [ {'id': 'gemini-1.5-flash', 'name': 'Gemini 1.5 Flash', 'description': '最便宜、速度快'}, {'id': 'gemini-2.5-flash', 'name': 'Gemini 2.5 Flash', 'description': '性價比最佳'}, {'id': 'gemini-2.5-pro', 'name': 'Gemini 2.5 Pro', 'description': '最強大、高品質'}, ] @dataclass class GeminiResponse: """Gemini 回應結構""" success: bool content: str model: str error: Optional[str] = None total_duration: Optional[float] = None input_tokens: int = 0 output_tokens: int = 0 total_tokens: int = 0 input_cost: float = 0.0 output_cost: float = 0.0 total_cost: float = 0.0 class GeminiService: """Google Gemini AI 服務""" # 連線狀態快取 _connection_cache = {'status': None, 'timestamp': 0, 'models': []} _CACHE_TTL = 300 # 快取 5 分鐘 def __init__(self, api_key: str = None, model: str = None): """ 初始化 Gemini 服務 Args: api_key: Google API Key（預設從環境變數取得） model: 預設模型（預設 gemini-1.5-flash） """ self.api_key = api_key or get_gemini_api_key("gemini_service") self.model = model or DEFAULT_GEMINI_MODEL self._client = None self._initialized = False def _ensure_initialized(self) -> bool: """確保 SDK 已初始化""" if self._initialized: return True if not is_gemini_fallback_enabled("gemini_service"): logger.info(gemini_disabled_message("gemini_service")) return False if not self.api_key: self.api_key = get_gemini_api_key("gemini_service") if not self.api_key: logger.warning("Gemini API Key 未設定") return False try: import google.generativeai as genai genai.configure(api_key=self.api_key) self._initialized = True logger.info("Gemini SDK 初始化成功") return True except ImportError: logger.error("請安裝 google-generativeai: pip install google-generativeai") return False except Exception as e: logger.error(f"Gemini SDK 初始化失敗: {e}") return False def check_connection(self) -> bool: """檢查 Gemini 服務是否可用（含快取）""" now = time.time() # 使用快取避免頻繁檢查 if (GeminiService._connection_cache['status'] is not None and now - GeminiService._connection_cache['timestamp'] < GeminiService._CACHE_TTL): return GeminiService._connection_cache['status'] if not self._ensure_initialized(): GeminiService._connection_cache = {'status': False, 'timestamp': now, 'models': []} return False try: import google.generativeai as genai # 嘗試列出模型驗證連線 models = list(genai.list_models()) available_models = [m.name for m in models if 'generateContent' in m.supported_generation_methods] logger.info(f"Gemini 連線成功，可用模型數: {len(available_models)}") GeminiService._connection_cache = { 'status': True, 'timestamp': now, 'models': available_models } return True except Exception as e: logger.error(f"Gemini 連線失敗: {e}") GeminiService._connection_cache = {'status': False, 'timestamp': now, 'models': []} return False def list_models(self) -> List[Dict]: """列出可用模型（供 UI 選擇）""" return AVAILABLE_GEMINI_MODELS @staticmethod def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> Dict[str, float]: """ 計算 API 費用 (USD) Args: model: 模型名稱 input_tokens: 輸入 token 數量 output_tokens: 輸出 token 數量 Returns: dict: {'input_cost': float, 'output_cost': float, 'total_cost': float} """ # 查找定價（支援完整模型名稱和簡短名稱） pricing = None for model_key, price in GEMINI_PRICING.items(): if model_key in model or model in model_key: pricing = price break if not pricing: # 使用預設定價（最便宜的） pricing = GEMINI_PRICING.get('gemini-1.5-flash', {'input': 0.075, 'output': 0.30}) # 計算費用（per 1M tokens） input_cost = (input_tokens / 1_000_000) * pricing['input'] output_cost = (output_tokens / 1_000_000) * pricing['output'] total_cost = input_cost + output_cost return { 'input_cost': round(input_cost, 6), 'output_cost': round(output_cost, 6), 'total_cost': round(total_cost, 6) } def generate(self, prompt: str, model: str = None, system_prompt: str = None, temperature: float = 0.7, timeout: int = None) -> GeminiResponse: """ 生成文字 Args: prompt: 使用者提示 model: 模型名稱（預設使用 self.model） system_prompt: 系統提示 temperature: 創意度 (0-2) timeout: 自訂超時時間（秒） Returns: GeminiResponse """ model_name = model or self.model request_timeout = timeout or GEMINI_TIMEOUT if not is_gemini_fallback_enabled("gemini_service.generate"): return GeminiResponse( success=False, content='', model=model_name, error=gemini_disabled_message("gemini_service.generate") ) if not self._ensure_initialized(): return GeminiResponse( success=False, content='', model=model_name, error="Gemini SDK 未初始化，請檢查 API Key" ) try: import google.generativeai as genai start_time = time.time() # 建立模型 generation_config = genai.types.GenerationConfig( temperature=temperature, max_output_tokens=2048, ) gen_model = genai.GenerativeModel( model_name=model_name, generation_config=generation_config, system_instruction=system_prompt if system_prompt else None ) logger.info(f"[Gemini] 開始生成，模型: {model_name}，超時: {request_timeout}秒") # 生成內容 response = gen_model.generate_content( prompt, request_options={'timeout': request_timeout} ) end_time = time.time() duration = end_time - start_time # 解析回應 content = response.text if response.text else '' # 取得 token 用量 input_tokens = 0 output_tokens = 0 if hasattr(response, 'usage_metadata'): input_tokens = getattr(response.usage_metadata, 'prompt_token_count', 0) output_tokens = getattr(response.usage_metadata, 'candidates_token_count', 0) # 計算費用 costs = self.calculate_cost(model_name, input_tokens, output_tokens) logger.info(f"[Gemini] 生成完成，耗時: {duration:.2f}秒，" f"tokens: {input_tokens}+{output_tokens}={input_tokens+output_tokens}，" f"費用: ${costs['total_cost']:.6f}") return GeminiResponse( success=True, content=content, model=model_name, total_duration=duration, input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, input_cost=costs['input_cost'], output_cost=costs['output_cost'], total_cost=costs['total_cost'] ) except Exception as e: logger.error(f"Gemini 生成錯誤: {e}") return GeminiResponse( success=False, content='', model=model_name, error=str(e) ) def generate_sales_copy(self, product_name: str, trend_keywords: List[str] = None, style: str = "吸睛", upcoming_holidays: List[Dict] = None, bestseller_products: List[Dict] = None, model: str = None) -> GeminiResponse: """ 生成銷售文案 Args: product_name: 商品名稱 trend_keywords: 相關趨勢關鍵字 style: 文案風格 (吸睛/專業/溫馨/急迫) upcoming_holidays: 即將到來的假期 bestseller_products: 競品熱銷商品 model: 指定模型（預設使用 self.model） Returns: GeminiResponse """ style_prompts = { "吸睛": "使用吸引眼球的標題和表情符號", "專業": "使用專業術語，強調成分和功效", "溫馨": "使用溫暖的語氣，強調呵護和關愛", "急迫": "使用限時優惠的語氣，創造緊迫感" } # 趨勢關鍵字 trend_context = "" if trend_keywords: trend_context = f"\n目前的熱門趨勢關鍵字：{', '.join(trend_keywords)}。請嘗試將這些趨勢融入文案中。" # 即將到來的假期 holiday_context = "" if upcoming_holidays: holidays_text = [] for h in upcoming_holidays[:3]: name = h.get('name', '') days = h.get('days_until', 0) if days == 0: holidays_text.append(f"{name}（今天）") elif days == 1: holidays_text.append(f"{name}（明天）") else: holidays_text.append(f"{name}（{days}天後）") if holidays_text: holiday_context = f"\n即將到來的假期：{', '.join(holidays_text)}。可以考慮結合節慶氛圍或送禮情境。" # 競品熱銷參考 bestseller_context = "" if bestseller_products: products_text = [f"{p.get('name', '')}（${p.get('price', '')}）" for p in bestseller_products[:3]] if products_text: bestseller_context = f"\n市場熱銷參考：{', '.join(products_text)}。可參考熱銷趨勢但要突出自家商品特色。" system_prompt = """你是一位專業的電商銷售文案寫手和行銷策略專家，專門為台灣電商平台撰寫商品文案。 你的文案特點： - 使用繁體中文 - 善用表情符號增加吸引力 - 強調商品賣點和消費者利益 - 適時使用行動呼籲 (CTA) - 若有即將到來的節日，可適度融入節慶元素 - 提供完整的行銷建議""" prompt = f"""請為以下商品撰寫完整的銷售文案套組： 商品名稱：{product_name} 文案風格：{style_prompts.get(style, style_prompts['吸睛'])} {trend_context}{holiday_context}{bestseller_context} 請按照以下格式生成完整的銷售文案套組： 【大標題】 （15字以內的主打標語，吸引眼球，適合用於廣告Banner） 【中標題】 （30字以內的副標題，補充說明賣點） 【小標題】 （20字以內的精簡標語，適合用於社群貼文） 【詳細文案】 （100-150字的完整銷售文案，包含商品特色、使用情境、行動呼籲） 【推廣建議】 • 社群推廣：（Facebook/Instagram/LINE 等社群平台的建議策略） • 影音內容：（短影音/直播/開箱影片等建議） • 其他建議：（EDM、部落格、KOL合作等專業建議） 請確保所有內容使用繁體中文，風格一致，並突出商品價值：""" return self.generate( prompt, model=model, system_prompt=system_prompt, temperature=0.8, timeout=90 ) def extract_keywords(self, text: str, max_keywords: int = 10, model: str = None) -> GeminiResponse: """ 從文字中提取關鍵字 Args: text: 要分析的文字 max_keywords: 最大關鍵字數量 model: 指定模型 Returns: GeminiResponse（content 為逗號分隔的關鍵字） """ system_prompt = "你是一位關鍵字提取專家。請從給定的文字中提取最重要的關鍵字。" prompt = f"""請從以下文字中提取最多 {max_keywords} 個關鍵字，這些關鍵字應該能代表文章的主題和重點。 文字內容： {text} 請只輸出關鍵字，用逗號分隔，不要輸出其他內容：""" return self.generate(prompt, model=model, system_prompt=system_prompt, temperature=0.3) def search_product_insights(self, product_name: str, include_competitors: bool = True, include_trends: bool = True, model: str = None) -> GeminiResponse: """ 搜尋商品相關的市場洞察 Args: product_name: 商品名稱 include_competitors: 是否包含競品分析 include_trends: 是否包含趨勢分析 model: 指定模型 Returns: GeminiResponse """ system_prompt = """你是一位資深的電商市場分析師，專精於台灣市場。 你擅長分析商品的市場定位、競爭對手、以及銷售趨勢。 請提供全面但簡潔的市場洞察，使用繁體中文。""" analysis_parts = ["市場定位分析"] if include_competitors: analysis_parts.append("主要競爭對手分析") if include_trends: analysis_parts.append("市場趨勢分析") competitors_json = '"competitors": [{"name": "競品名稱", "strength": "優勢", "weakness": "劣勢"}],' if include_competitors else "" trends_json = '"trends": {"current": "當前趨勢", "forecast": "趨勢預測", "seasonality": "季節性因素"},' if include_trends else "" analysis_list = '\n'.join([f'{i+1}. {part}' for i, part in enumerate(analysis_parts)]) prompt = f"""請為以下商品提供市場洞察分析： 商品名稱：{product_name} 請分析以下面向： {analysis_list} 請用以下 JSON 格式回覆（務必輸出有效的 JSON）： {{ "product_name": "{product_name}", "market_position": {{ "target_audience": "目標客群描述", "price_range": "價格區間建議", "positioning": "市場定位建議" }}, {competitors_json} {trends_json} "recommendations": ["銷售建議1", "銷售建議2", "銷售建議3"], "keywords": ["行銷關鍵字1", "關鍵字2", "關鍵字3"] }}""" return self.generate(prompt, model=model, system_prompt=system_prompt, temperature=0.6, timeout=90) # 建立全域服務實例 gemini_service = GeminiService() if __name__ == "__main__": # 測試程式碼 logging.basicConfig(level=logging.INFO) service = GeminiService() # 測試連線 print("測試 Gemini 連線...") if service.check_connection(): print(f"連線成功！") # 測試文案生成 print("\n測試文案生成...") result = service.generate_sales_copy( "玻尿酸保濕面膜", trend_keywords=["換季保養", "敏感肌"], style="吸睛" ) if result.success: print(f"生成結果:\n{result.content}") print(f"\n耗時: {result.total_duration:.2f} 秒") print(f"Token 用量: 輸入 {result.input_tokens}, 輸出 {result.output_tokens}") print(f"費用: ${result.total_cost:.6f} USD") else: print(f"生成失敗: {result.error}") else: print("連線失敗（請確認 GEMINI_API_KEY 環境變數）")