48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
"""
|
|
通用工具函数
|
|
"""
|
|
import os
|
|
import json
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Dict
|
|
|
|
|
|
def get_current_timestamp():
|
|
"""获取当前时间戳"""
|
|
return int(datetime.utcnow().timestamp())
|
|
|
|
|
|
def generate_unique_id():
|
|
"""生成唯一ID"""
|
|
return str(uuid.uuid4())
|
|
|
|
|
|
def format_api_response(content: str, conversation_id: str = None, model: str = "qwen-plus"):
|
|
"""格式化API响应"""
|
|
return {
|
|
"id": generate_unique_id(),
|
|
"conversationId": conversation_id or generate_unique_id(),
|
|
"content": content,
|
|
"model": model,
|
|
"createdAt": get_current_timestamp()
|
|
}
|
|
|
|
|
|
def log_request(method: str, path: str, client_ip: str = "unknown"):
|
|
"""记录请求日志"""
|
|
print(f"[INFO] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - "
|
|
f"HTTP {method} {path} - IP: {client_ip}")
|
|
|
|
|
|
def log_response(status_code: int, process_time: float):
|
|
"""记录响应日志"""
|
|
print(f"[INFO] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - "
|
|
f"Response {status_code}, Process Time: {process_time:.2f}ms")
|
|
|
|
|
|
def extract_delta_content(full_content: str, previous_content: str) -> str:
|
|
"""提取增量内容"""
|
|
if len(full_content) > len(previous_content):
|
|
return full_content[len(previous_content):]
|
|
return "" |