280 lines
7.2 KiB
Python
280 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
輔助工具模組
|
|
|
|
Author: PANJIT IT Team
|
|
Created: 2024-01-28
|
|
Modified: 2024-01-28
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
import shutil
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from werkzeug.utils import secure_filename
|
|
from flask import current_app
|
|
|
|
|
|
def generate_filename(original_filename, job_uuid, file_type='original', language_code=None):
|
|
"""生成安全的檔案名稱"""
|
|
# 取得檔案副檔名
|
|
file_ext = Path(original_filename).suffix.lower()
|
|
|
|
# 清理原始檔名
|
|
clean_name = Path(original_filename).stem
|
|
clean_name = secure_filename(clean_name)[:50] # 限制長度
|
|
|
|
if file_type == 'original':
|
|
return f"original_{clean_name}_{job_uuid[:8]}{file_ext}"
|
|
elif file_type == 'translated':
|
|
return f"translated_{clean_name}_{language_code}_{job_uuid[:8]}{file_ext}"
|
|
else:
|
|
return f"{file_type}_{clean_name}_{job_uuid[:8]}{file_ext}"
|
|
|
|
|
|
def create_job_directory(job_uuid):
|
|
"""建立任務專用目錄"""
|
|
upload_folder = current_app.config.get('UPLOAD_FOLDER')
|
|
job_dir = Path(upload_folder) / job_uuid
|
|
|
|
# 建立目錄
|
|
job_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
return job_dir
|
|
|
|
|
|
def save_uploaded_file(file_obj, job_uuid):
|
|
"""儲存上傳的檔案"""
|
|
try:
|
|
# 建立任務目錄
|
|
job_dir = create_job_directory(job_uuid)
|
|
|
|
# 生成檔案名稱
|
|
filename = generate_filename(file_obj.filename, job_uuid, 'original')
|
|
file_path = job_dir / filename
|
|
|
|
# 儲存檔案
|
|
file_obj.save(str(file_path))
|
|
|
|
# 取得檔案大小
|
|
file_size = file_path.stat().st_size
|
|
|
|
return {
|
|
'success': True,
|
|
'filename': filename,
|
|
'file_path': str(file_path),
|
|
'file_size': file_size
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'error': str(e)
|
|
}
|
|
|
|
|
|
def cleanup_job_directory(job_uuid):
|
|
"""清理任務目錄"""
|
|
try:
|
|
upload_folder = current_app.config.get('UPLOAD_FOLDER')
|
|
job_dir = Path(upload_folder) / job_uuid
|
|
|
|
if job_dir.exists() and job_dir.is_dir():
|
|
shutil.rmtree(job_dir)
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def format_file_size(size_bytes):
|
|
"""格式化檔案大小"""
|
|
if size_bytes == 0:
|
|
return "0 B"
|
|
|
|
size_names = ["B", "KB", "MB", "GB", "TB"]
|
|
i = 0
|
|
while size_bytes >= 1024 and i < len(size_names) - 1:
|
|
size_bytes /= 1024.0
|
|
i += 1
|
|
|
|
return f"{size_bytes:.1f} {size_names[i]}"
|
|
|
|
|
|
def get_file_icon(file_extension):
|
|
"""根據副檔名取得檔案圖示"""
|
|
icon_map = {
|
|
'.docx': 'file-word',
|
|
'.doc': 'file-word',
|
|
'.pptx': 'file-powerpoint',
|
|
'.ppt': 'file-powerpoint',
|
|
'.xlsx': 'file-excel',
|
|
'.xls': 'file-excel',
|
|
'.pdf': 'file-pdf'
|
|
}
|
|
|
|
return icon_map.get(file_extension.lower(), 'file')
|
|
|
|
|
|
def calculate_processing_time(start_time, end_time=None):
|
|
"""計算處理時間"""
|
|
if not start_time:
|
|
return None
|
|
|
|
if not end_time:
|
|
end_time = datetime.utcnow()
|
|
|
|
if isinstance(start_time, str):
|
|
start_time = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
|
|
|
|
if isinstance(end_time, str):
|
|
end_time = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
|
|
|
|
duration = end_time - start_time
|
|
|
|
# 轉換為秒
|
|
total_seconds = int(duration.total_seconds())
|
|
|
|
if total_seconds < 60:
|
|
return f"{total_seconds}秒"
|
|
elif total_seconds < 3600:
|
|
minutes = total_seconds // 60
|
|
seconds = total_seconds % 60
|
|
return f"{minutes}分{seconds}秒"
|
|
else:
|
|
hours = total_seconds // 3600
|
|
minutes = (total_seconds % 3600) // 60
|
|
return f"{hours}小時{minutes}分"
|
|
|
|
|
|
def generate_download_token(job_uuid, language_code, user_id):
|
|
"""生成下載令牌"""
|
|
import hashlib
|
|
import time
|
|
|
|
# 組合資料
|
|
data = f"{job_uuid}:{language_code}:{user_id}:{int(time.time())}"
|
|
|
|
# 加上應用程式密鑰
|
|
secret_key = current_app.config.get('SECRET_KEY', 'default_secret')
|
|
data_with_secret = f"{data}:{secret_key}"
|
|
|
|
# 生成 hash
|
|
token = hashlib.sha256(data_with_secret.encode()).hexdigest()
|
|
|
|
return token
|
|
|
|
|
|
def verify_download_token(token, job_uuid, language_code, user_id, max_age=3600):
|
|
"""驗證下載令牌"""
|
|
import time
|
|
|
|
try:
|
|
# 取得當前時間戳
|
|
current_time = int(time.time())
|
|
|
|
# 在有效時間範圍內嘗試匹配令牌
|
|
for i in range(max_age):
|
|
timestamp = current_time - i
|
|
expected_token = generate_download_token_with_timestamp(
|
|
job_uuid, language_code, user_id, timestamp
|
|
)
|
|
|
|
if token == expected_token:
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def generate_download_token_with_timestamp(job_uuid, language_code, user_id, timestamp):
|
|
"""使用指定時間戳生成下載令牌"""
|
|
import hashlib
|
|
|
|
data = f"{job_uuid}:{language_code}:{user_id}:{timestamp}"
|
|
secret_key = current_app.config.get('SECRET_KEY', 'default_secret')
|
|
data_with_secret = f"{data}:{secret_key}"
|
|
|
|
return hashlib.sha256(data_with_secret.encode()).hexdigest()
|
|
|
|
|
|
def get_supported_languages():
|
|
"""取得支援的語言列表"""
|
|
return {
|
|
'auto': '自動偵測',
|
|
'zh-CN': '簡體中文',
|
|
'zh-TW': '繁體中文',
|
|
'en': '英文',
|
|
'ja': '日文',
|
|
'ko': '韓文',
|
|
'vi': '越南文',
|
|
'th': '泰文',
|
|
'id': '印尼文',
|
|
'ms': '馬來文',
|
|
'es': '西班牙文',
|
|
'fr': '法文',
|
|
'de': '德文',
|
|
'ru': '俄文'
|
|
}
|
|
|
|
|
|
def parse_json_field(json_str):
|
|
"""安全解析JSON欄位"""
|
|
import json
|
|
|
|
if not json_str:
|
|
return None
|
|
|
|
try:
|
|
if isinstance(json_str, str):
|
|
return json.loads(json_str)
|
|
return json_str
|
|
except (json.JSONDecodeError, TypeError):
|
|
return None
|
|
|
|
|
|
def format_datetime(dt, format_type='full'):
|
|
"""格式化日期時間"""
|
|
if not dt:
|
|
return None
|
|
|
|
if isinstance(dt, str):
|
|
try:
|
|
dt = datetime.fromisoformat(dt.replace('Z', '+00:00'))
|
|
except ValueError:
|
|
return dt
|
|
|
|
if format_type == 'date':
|
|
return dt.strftime('%Y-%m-%d')
|
|
elif format_type == 'time':
|
|
return dt.strftime('%H:%M:%S')
|
|
elif format_type == 'short':
|
|
return dt.strftime('%Y-%m-%d %H:%M')
|
|
else: # full
|
|
return dt.strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
|
|
def create_response(success=True, data=None, message=None, error=None, error_code=None):
|
|
"""建立統一的API回應格式"""
|
|
response = {
|
|
'success': success
|
|
}
|
|
|
|
if data is not None:
|
|
response['data'] = data
|
|
|
|
if message:
|
|
response['message'] = message
|
|
|
|
if error:
|
|
response['error'] = error_code or 'ERROR'
|
|
if not message:
|
|
response['message'] = error
|
|
|
|
return response |