302 lines
11 KiB
Python
302 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Dify API 客戶端服務
|
||
|
||
Author: PANJIT IT Team
|
||
Created: 2024-01-28
|
||
Modified: 2024-01-28
|
||
"""
|
||
|
||
import time
|
||
import requests
|
||
from typing import Dict, Any, Optional
|
||
from flask import current_app
|
||
from app.utils.logger import get_logger
|
||
from app.utils.exceptions import APIError
|
||
from app.models.stats import APIUsageStats
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class DifyClient:
|
||
"""Dify API 客戶端"""
|
||
|
||
def __init__(self):
|
||
self.base_url = current_app.config.get('DIFY_API_BASE_URL', '')
|
||
self.api_key = current_app.config.get('DIFY_API_KEY', '')
|
||
self.timeout = (10, 60) # (連接超時, 讀取超時)
|
||
self.max_retries = 3
|
||
self.retry_delay = 1.6 # 指數退避基數
|
||
|
||
if not self.base_url or not self.api_key:
|
||
logger.warning("Dify API configuration is incomplete")
|
||
|
||
def _make_request(self, method: str, endpoint: str, data: Dict[str, Any] = None,
|
||
user_id: int = None, job_id: int = None) -> Dict[str, Any]:
|
||
"""發送 HTTP 請求到 Dify API"""
|
||
|
||
if not self.base_url or not self.api_key:
|
||
raise APIError("Dify API 未配置完整")
|
||
|
||
url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
|
||
|
||
headers = {
|
||
'Authorization': f'Bearer {self.api_key}',
|
||
'Content-Type': 'application/json',
|
||
'User-Agent': 'PANJIT-Document-Translator/1.0'
|
||
}
|
||
|
||
# 重試邏輯
|
||
last_exception = None
|
||
start_time = time.time()
|
||
|
||
for attempt in range(self.max_retries):
|
||
try:
|
||
logger.debug(f"Making Dify API request: {method} {url} (attempt {attempt + 1})")
|
||
|
||
if method.upper() == 'GET':
|
||
response = requests.get(url, headers=headers, timeout=self.timeout, params=data)
|
||
else:
|
||
response = requests.post(url, headers=headers, timeout=self.timeout, json=data)
|
||
|
||
# 計算響應時間
|
||
response_time_ms = int((time.time() - start_time) * 1000)
|
||
|
||
# 檢查響應狀態
|
||
response.raise_for_status()
|
||
|
||
# 解析響應
|
||
result = response.json()
|
||
|
||
# 記錄 API 使用統計
|
||
if user_id:
|
||
self._record_api_usage(
|
||
user_id=user_id,
|
||
job_id=job_id,
|
||
endpoint=endpoint,
|
||
response_data=result,
|
||
response_time_ms=response_time_ms,
|
||
success=True
|
||
)
|
||
|
||
logger.debug(f"Dify API request successful: {response_time_ms}ms")
|
||
return result
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
last_exception = e
|
||
response_time_ms = int((time.time() - start_time) * 1000)
|
||
|
||
# 記錄失敗的 API 調用
|
||
if user_id:
|
||
self._record_api_usage(
|
||
user_id=user_id,
|
||
job_id=job_id,
|
||
endpoint=endpoint,
|
||
response_data={},
|
||
response_time_ms=response_time_ms,
|
||
success=False,
|
||
error_message=str(e)
|
||
)
|
||
|
||
logger.warning(f"Dify API request failed (attempt {attempt + 1}): {str(e)}")
|
||
|
||
# 如果是最後一次嘗試，拋出異常
|
||
if attempt == self.max_retries - 1:
|
||
break
|
||
|
||
# 指數退避
|
||
delay = self.retry_delay ** attempt
|
||
logger.debug(f"Retrying in {delay} seconds...")
|
||
time.sleep(delay)
|
||
|
||
# 所有重試都失敗了
|
||
error_msg = f"Dify API request failed after {self.max_retries} attempts: {str(last_exception)}"
|
||
logger.error(error_msg)
|
||
raise APIError(error_msg)
|
||
|
||
def _record_api_usage(self, user_id: int, job_id: Optional[int], endpoint: str,
|
||
response_data: Dict, response_time_ms: int, success: bool,
|
||
error_message: str = None):
|
||
"""記錄 API 使用統計"""
|
||
try:
|
||
# 從響應中提取使用量資訊
|
||
metadata = response_data.get('metadata', {})
|
||
|
||
# 如果 job_id 無效，則設為 None 以避免外鍵約束錯誤
|
||
APIUsageStats.record_api_call(
|
||
user_id=user_id,
|
||
job_id=job_id, # 已經是 Optional，如果無效會被設為 NULL
|
||
api_endpoint=endpoint,
|
||
metadata=metadata,
|
||
response_time_ms=response_time_ms,
|
||
success=success,
|
||
error_message=error_message
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"Failed to record API usage: {str(e)}")
|
||
|
||
def translate_text(self, text: str, source_language: str, target_language: str,
|
||
user_id: int = None, job_id: int = None) -> Dict[str, Any]:
|
||
"""翻譯文字"""
|
||
|
||
if not text.strip():
|
||
raise APIError("翻譯文字不能為空")
|
||
|
||
# 構建標準翻譯 prompt（英文指令格式）
|
||
language_names = {
|
||
'zh-tw': 'Traditional Chinese',
|
||
'zh-cn': 'Simplified Chinese',
|
||
'en': 'English',
|
||
'ja': 'Japanese',
|
||
'ko': 'Korean',
|
||
'vi': 'Vietnamese',
|
||
'th': 'Thai',
|
||
'id': 'Indonesian',
|
||
'ms': 'Malay',
|
||
'es': 'Spanish',
|
||
'fr': 'French',
|
||
'de': 'German',
|
||
'ru': 'Russian',
|
||
'ar': 'Arabic'
|
||
}
|
||
|
||
source_lang_name = language_names.get(source_language, source_language)
|
||
target_lang_name = language_names.get(target_language, target_language)
|
||
|
||
query = f"""Task: Translate ONLY into {target_lang_name} from {source_lang_name}.
|
||
|
||
Rules:
|
||
- Output translation text ONLY (no source text, no notes, no questions, no language-detection remarks).
|
||
- Preserve original line breaks.
|
||
- Do NOT wrap in quotes or code blocks.
|
||
- Maintain original formatting and structure.
|
||
|
||
{text.strip()}"""
|
||
|
||
# 構建請求資料 - 使用成功版本的格式
|
||
request_data = {
|
||
'inputs': {},
|
||
'response_mode': 'blocking',
|
||
'user': f"user_{user_id}" if user_id else "doc-translator-user",
|
||
'query': query
|
||
}
|
||
|
||
try:
|
||
response = self._make_request(
|
||
method='POST',
|
||
endpoint='/chat-messages',
|
||
data=request_data,
|
||
user_id=user_id,
|
||
job_id=job_id
|
||
)
|
||
|
||
# 從響應中提取翻譯結果 - 使用成功版本的方式
|
||
answer = response.get('answer')
|
||
|
||
if not isinstance(answer, str) or not answer.strip():
|
||
raise APIError("Dify API 返回空的翻譯結果")
|
||
|
||
return {
|
||
'success': True,
|
||
'translated_text': answer,
|
||
'source_text': text,
|
||
'source_language': source_language,
|
||
'target_language': target_language,
|
||
'metadata': response.get('metadata', {})
|
||
}
|
||
|
||
except APIError:
|
||
raise
|
||
except Exception as e:
|
||
error_msg = f"翻譯請求處理錯誤: {str(e)}"
|
||
logger.error(error_msg)
|
||
raise APIError(error_msg)
|
||
|
||
def test_connection(self) -> bool:
|
||
"""測試 Dify API 連接"""
|
||
try:
|
||
# 發送簡單的測試請求
|
||
test_data = {
|
||
'inputs': {'text': 'test'},
|
||
'response_mode': 'blocking',
|
||
'user': 'health_check'
|
||
}
|
||
|
||
response = self._make_request(
|
||
method='POST',
|
||
endpoint='/chat-messages',
|
||
data=test_data
|
||
)
|
||
|
||
return response is not None
|
||
|
||
except Exception as e:
|
||
logger.error(f"Dify API connection test failed: {str(e)}")
|
||
return False
|
||
|
||
def get_app_info(self) -> Dict[str, Any]:
|
||
"""取得 Dify 應用資訊"""
|
||
try:
|
||
response = self._make_request(
|
||
method='GET',
|
||
endpoint='/parameters'
|
||
)
|
||
|
||
return {
|
||
'success': True,
|
||
'app_info': response
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"Failed to get Dify app info: {str(e)}")
|
||
return {
|
||
'success': False,
|
||
'error': str(e)
|
||
}
|
||
|
||
@classmethod
|
||
def load_config_from_file(cls, file_path: str = 'api.txt'):
|
||
"""從檔案載入 Dify API 配置"""
|
||
try:
|
||
import os
|
||
from pathlib import Path
|
||
|
||
config_file = Path(file_path)
|
||
|
||
if not config_file.exists():
|
||
logger.warning(f"Dify config file not found: {file_path}")
|
||
return
|
||
|
||
with open(config_file, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line.startswith('base_url:'):
|
||
base_url = line.split(':', 1)[1].strip()
|
||
current_app.config['DIFY_API_BASE_URL'] = base_url
|
||
elif line.startswith('api:'):
|
||
api_key = line.split(':', 1)[1].strip()
|
||
current_app.config['DIFY_API_KEY'] = api_key
|
||
|
||
logger.info("Dify API config loaded from file")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Failed to load Dify config from file: {str(e)}")
|
||
|
||
|
||
def init_dify_config(app):
|
||
"""初始化 Dify 配置"""
|
||
with app.app_context():
|
||
# 從 api.txt 載入配置
|
||
DifyClient.load_config_from_file()
|
||
|
||
# 檢查配置完整性
|
||
base_url = app.config.get('DIFY_API_BASE_URL')
|
||
api_key = app.config.get('DIFY_API_KEY')
|
||
|
||
if base_url and api_key:
|
||
logger.info("Dify API configuration loaded successfully")
|
||
else:
|
||
logger.warning("Dify API configuration is incomplete")
|
||
logger.warning(f"Base URL: {'✓' if base_url else '✗'}")
|
||
logger.warning(f"API Key: {'✓' if api_key else '✗'}") |