GithubBackup/gemini-balance
1
0
Fork 0
mirror of https://github.com/snailyp/gemini-balance.git synced 2026-08-28 19:47:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 改进错误日志功能并优化应用初始化流程

Browse Source 
本次提交主要包含以下更新：

- **错误日志页面增强**:
    - 重构了 [`app/static/js/error_logs.js`](app/static/js/error_logs.js) 中的分页逻辑，将样式控制移至 CSS，简化了 JavaScript 代码。
    - 更新了 [`app/templates/error_logs.html`](app/templates/error_logs.html) 中的分页样式，使其与 `keys_status.html` 保持一致，提升了视觉统一性。
    - 在错误日志页面新增了“清空全部”按钮，方便用户一键清除所有错误记录。
    - 调整了错误日志表格头部的文本颜色为白色，以改善深色主题下的可读性。

- **应用初始化与配置优化**:
    - 调整了 [`app/config/config.py`](app/config/config.py) 中日志记录器的获取方式，确保在配置加载早期即可用。
    - 在 [`app/core/application.py`](app/core/application.py) 中引入了更明确的数据库连接管理（连接、断开、初始化）逻辑。
    - 优化了 [`app/utils/helpers.py`](app/utils/helpers.py) 中项目路径和版本文件路径的定义方式，使其在模块级别初始化。

- **依赖清理**:
    - 从 [`requirements.txt`](requirements.txt) 中移除了不必要的注释。

这些更改旨在提升错误日志模块的用户体验和功能性，并优化应用程序的启动和配置管理流程。
This commit is contained in:
snaily
2025-05-14 14:25:04 +08:00
parent 67f87989db
commit 4becc8d4d4
27 changed files with 1116 additions and 1000 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/service/chat/gemini_chat_service.py
+13 -20
View File
@@ -144,7 +144,7 @@ class GeminiChatService:
"""生成内容"""
payload = _build_payload(model, request)
start_time = time.perf_counter()
request_datetime = datetime.datetime.now() # Record request time
request_datetime = datetime.datetime.now()
is_success = False
status_code = None
response = None
@@ -152,20 +152,18 @@ class GeminiChatService:
try:
response = await self.api_client.generate_content(payload, model, api_key)
is_success = True
status_code = 200 # Assume 200 on success
status_code = 200
return self.response_handler.handle_response(response, model, stream=False)
except Exception as e:
is_success = False
error_log_msg = str(e)
logger.error(f"Normal API call failed with error: {error_log_msg}")
# Try to parse status code from exception
match = re.search(r"status code (\d+)", error_log_msg)
if match:
status_code = int(match.group(1))
else:
status_code = 500 # Default to 500 if parsing fails
status_code = 500
# Log error to error log table
await add_error_log(
gemini_key=api_key,
model_name=model,
@@ -174,11 +172,10 @@ class GeminiChatService:
error_code=status_code,
request_msg=payload
)
raise e # Re-throw exception for upstream handling
raise e
finally:
end_time = time.perf_counter()
latency_ms = int((end_time - start_time) * 1000)
# Log request to request log table
await add_request_log(
model_name=model,
api_key=api_key,
@@ -240,16 +237,14 @@ class GeminiChatService:
logger.warning(
f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}"
)
# Parse error code for logging
match = re.search(r"status code (\d+)", error_log_msg)
if match:
status_code = int(match.group(1))
else:
status_code = 500
# Log error to error log table
await add_error_log(
gemini_key=current_attempt_key, # Log key used for this failed attempt
gemini_key=current_attempt_key,
model_name=model,
error_type="gemini-chat-stream",
error_log=error_log_msg,
@@ -257,28 +252,26 @@ class GeminiChatService:
request_msg=payload
)
# Attempt to switch API Key
api_key = await self.key_manager.handle_api_failure(current_attempt_key, retries)
if api_key:
logger.info(f"Switched to new API key: {api_key}")
else: # No more keys or retries exceeded by handle_api_failure logic
logger.error(f"No valid API key available after {retries} retries.")
break # Exit loop if no key available
else:
logger.error(f"No valid API key available after {retries} retries.")
break
if retries >= max_retries:
logger.error(
f"Max retries ({max_retries}) reached for streaming."
)
break # Exit loop after max retries
break
finally:
# Log the final outcome of the streaming request
end_time = time.perf_counter()
latency_ms = int((end_time - start_time) * 1000)
await add_request_log(
model_name=model,
api_key=final_api_key, # Log the last key used
is_success=is_success, # Log the final success status
status_code=status_code, # Log the last known status code
latency_ms=latency_ms, # Log total time including retries
api_key=final_api_key,
is_success=is_success,
status_code=status_code,
latency_ms=latency_ms,
request_time=request_datetime
)
app/service/chat/openai_chat_service.py
+1 -5
View File
@@ -173,7 +173,7 @@ class OpenAIChatService:
self, original_chunk: Dict[str, Any], text: str
) -> Dict[str, Any]:
"""创建包含指定文本的OpenAI响应块"""
chunk_copy = json.loads(json.dumps(original_chunk)) # 深拷贝
chunk_copy = json.loads(json.dumps(original_chunk))
if chunk_copy.get("choices") and "delta" in chunk_copy["choices"][0]:
chunk_copy["choices"][0]["delta"]["content"] = text
return chunk_copy
@@ -184,10 +184,8 @@ class OpenAIChatService:
api_key: str,
) -> Union[Dict[str, Any], AsyncGenerator[str, None]]:
"""创建聊天完成"""
# 转换消息格式
messages, instruction = self.message_converter.convert(request.messages)
# 构建请求payload
payload = _build_payload(request, messages, instruction)
if request.stream:
@@ -219,7 +217,6 @@ class OpenAIChatService:
is_success = False
error_log_msg = str(e)
logger.error(f"Normal API call failed with error: {error_log_msg}")
# Try to parse status code from exception
match = re.search(r"status code (\d+)", error_log_msg)
if match:
status_code = int(match.group(1))
@@ -587,7 +584,6 @@ class OpenAIChatService:
error_code=status_code,
request_msg={"image_data_truncated": image_data[:1000]},
)
# Re-raise the exception so the caller knows about the failure
raise e
finally:
end_time = time.perf_counter()
app/service/client/api_client.py
+1 -3
View File
@@ -53,16 +53,14 @@ class GeminiApiClient(ApiClient):
url = f"{self.base_url}/models?key={api_key}"
try:
response = await client.get(url)
response.raise_for_status() # 如果状态码不是 2xx，则引发 HTTPStatusError
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"获取模型列表失败: {e.response.status_code}")
logger.error(e.response.text)
# 返回 None 而不是抛出异常，以便上层处理
return None
except httpx.RequestError as e:
logger.error(f"请求模型列表失败: {e}")
# 返回 None 而不是抛出异常
return None
async def generate_content(self, payload: Dict[str, Any], model: str, api_key: str) -> Dict[str, Any]:
app/service/config/config_service.py
+1 -6
View File
@@ -76,7 +76,6 @@ class ConfigService:
}
if key in existing_keys:
# Preserve original description if not explicitly provided
data["description"] = existing_settings_map[key].get(
"description", description
)
@@ -111,7 +110,7 @@ class ConfigService:
logger.info(f"Updated {len(settings_to_update)} settings.")
except Exception as e:
logger.error(f"Failed to bulk update/insert settings: {str(e)}")
raise # Re-raise the exception after logging
raise
# 重置并重新初始化 KeyManager
try:
@@ -120,8 +119,6 @@ class ConfigService:
logger.info("KeyManager instance re-initialized with updated settings.")
except Exception as e:
logger.error(f"Failed to re-initialize KeyManager: {str(e)}")
# Decide if this error should prevent returning the updated config
# For now, we log the error and continue
return await ConfigService.get_config()
@@ -244,13 +241,11 @@ class ConfigService:
models = await model_service.get_gemini_openai_models(api_key)
return models
except HTTPException as e:
# Re-raise HTTPExceptions directly if they are already specific
raise e
except Exception as e:
logger.error(
f"Failed to fetch models for UI in ConfigService: {e}", exc_info=True
)
# Raise a generic HTTPException for other errors
raise HTTPException(
status_code=500, detail=f"Failed to fetch models for UI: {str(e)}"
)
app/service/embedding/embedding_service.py
+13 -17
View File
@@ -1,15 +1,15 @@
import datetime
import time
import re # For potential status code parsing from generic errors
import re
from typing import List, Union
import openai
from openai import APIStatusError # Import specific error type
from openai import APIStatusError
from openai.types import CreateEmbeddingResponse
from app.config.config import settings
from app.log.logger import get_embeddings_logger
from app.database.services import add_error_log, add_request_log # Import DB logging functions
from app.database.services import add_error_log, add_request_log
logger = get_embeddings_logger()
@@ -26,7 +26,6 @@ class EmbeddingService:
status_code = None
response = None
error_log_msg = ""
# Prepare request message for logging (truncate if list or long string)
if isinstance(input_text, list):
request_msg_log = {"input_truncated": [str(item)[:100] + "..." if len(str(item)) > 100 else str(item) for item in input_text[:5]]}
if len(input_text) > 5:
@@ -46,32 +45,29 @@ class EmbeddingService:
status_code = e.status_code
error_log_msg = f"OpenAI API error: {e}"
logger.error(f"Error creating embedding (APIStatusError): {error_log_msg}")
raise e # Re-raise the specific error
raise e
except Exception as e:
is_success = False
error_log_msg = f"Generic error: {e}"
logger.error(f"Error creating embedding (Exception): {error_log_msg}")
# Try to parse status code from generic error (less reliable)
match = re.search(r"status code (\d+)", str(e))
if match:
status_code = int(match.group(1))
else:
status_code = 500 # Default if parsing fails
raise e # Re-raise the generic error
status_code = 500
raise e
finally:
end_time = time.perf_counter()
latency_ms = int((end_time - start_time) * 1000)
if not is_success:
# Log error to database if it failed
await add_error_log(
gemini_key=api_key, # Using gemini_key parameter name for consistency
model_name=model,
error_type="openai-embedding",
error_log=error_log_msg,
error_code=status_code,
request_msg=request_msg_log
await add_error_log(
gemini_key=api_key,
model_name=model,
error_type="openai-embedding",
error_log=error_log_msg,
error_code=status_code,
request_msg=request_msg_log
)
# Log request outcome to database regardless of success/failure
await add_request_log(
model_name=model,
api_key=api_key,
app/service/error_log/error_log_service.py
+24
View File
@@ -153,3 +153,27 @@ async def process_delete_error_log_by_id(log_id: int) -> bool:
exc_info=True,
)
raise
async def process_delete_all_error_logs() -> int:
"""
处理删除所有错误日志的请求。
返回删除的日志数量。
"""
try:
# 确保数据库已连接 (如果适用，类似于 delete_old_error_logs)
# if not database.is_connected:
# await database.connect()
# logger.info("Database connection established for deleting all error logs.")
deleted_count = await db_services.delete_all_error_logs()
logger.info(
f"Successfully processed request to delete all error logs. Count: {deleted_count}"
)
return deleted_count
except Exception as e:
logger.error(
f"Service error in process_delete_all_error_logs: {e}",
exc_info=True,
)
raise
app/service/image/image_create_service.py
+1 -1
View File
@@ -137,7 +137,7 @@ class ImageCreateService:
)
response_data = {
"created": int(time.time()), # Current timestamp
"created": int(time.time()),
"data": images_data,
}
return response_data
app/service/key/key_manager.py
+8 -44
View File
@@ -60,7 +60,6 @@ class KeyManager:
current_key = await self.get_next_key()
if current_key == initial_key:
# await self.reset_failure_counts() 取消重置
return current_key
async def handle_api_failure(self, api_key: str, retries: int) -> str:
@@ -101,25 +100,12 @@ class KeyManager:
for key in self.key_failure_counts:
if self.key_failure_counts[key] < self.MAX_FAILURES:
return key
# 如果所有 key 都无效，或者列表为空，则尝试返回第一个（如果列表不为空）
# 或者根据具体逻辑处理，这里保持原样，可能在空列表或全无效时需要调整
if self.api_keys:
return self.api_keys[0]
# 如果 api_keys 为空，这里会出问题。实际应用中应有非空保证或更好处理。
# 为了保持接口一致性，如果列表为空，可能应该抛出异常或返回特定值。
# 暂且假设 api_keys 不会为空，或者调用者处理后续的空 key 问题。
# 根据现有代码，如果api_keys为空，self.api_keys[0]会报错。
# 如果没有有效key且列表不空，返回第一个。若列表为空，这里会出IndexError。
# 更安全的做法是：
if not self.api_keys:
logger.warning("API key list is empty, cannot get first valid key.")
# Depending on desired behavior, either raise error or return an indicator like "" or None
# For now, let's allow it to potentially fail if a key is expected by caller
# but it's better to be explicit. Let's return empty string for consistency with handle_api_failure
return ""
return self.api_keys[
0
] # Fallback to the first key if no key is "valid" but list is not empty
return self.api_keys[0]
_singleton_instance = None
@@ -142,20 +128,13 @@ async def get_key_manager_instance(api_keys: list = None) -> KeyManager:
async with _singleton_lock:
if _singleton_instance is None:
if api_keys is None:
# This case needs careful handling. If it's the very first call, api_keys are required.
# If it's after a reset and no api_keys are provided, what should happen?
# The original ValueError was "API keys are required to initialize the KeyManager".
# Let's assume if api_keys is None here, it's an error unless we are restoring from non-None _preserved_old_api_keys_for_reset.
# However, the user's request implies new api_keys will be part of the reset flow.
# For now, stick to a strict requirement for api_keys if _singleton_instance is None.
raise ValueError(
"API keys are required to initialize or re-initialize the KeyManager instance."
)
if not api_keys: # Handle case where api_keys is an empty list
if not api_keys:
logger.warning(
"Initializing KeyManager with an empty list of API keys."
)
# Consider if this should be an error or allowed. Current KeyManager supports it.
_singleton_instance = KeyManager(api_keys)
logger.info(
@@ -164,33 +143,28 @@ async def get_key_manager_instance(api_keys: list = None) -> KeyManager:
# 1. 恢复失败计数
if _preserved_failure_counts:
# Initialize new instance's failure_counts for all new keys to 0
current_failure_counts = {
key: 0 for key in _singleton_instance.api_keys
}
# Inherit counts for keys that exist in both old and new lists
for key, count in _preserved_failure_counts.items():
if key in current_failure_counts:
current_failure_counts[key] = count
_singleton_instance.key_failure_counts = current_failure_counts
logger.info("Inherited failure counts for applicable keys.")
_preserved_failure_counts = None # Clear after use
_preserved_failure_counts = None
# 2. 调整 key_cycle 的起始点
start_key_for_new_cycle = None
if (
_preserved_old_api_keys_for_reset
and _preserved_next_key_in_cycle
and _singleton_instance.api_keys # Ensure new api_keys list is not empty
and _singleton_instance.api_keys
):
try:
# Find the index of the preserved next key in the *old* list
start_idx_in_old = _preserved_old_api_keys_for_reset.index(
_preserved_next_key_in_cycle
)
# Iterate through the old key list (circularly) starting from _preserved_next_key_in_cycle
# Find the first key that also exists in the new api_keys list
for i in range(len(_preserved_old_api_keys_for_reset)):
current_old_key_idx = (start_idx_in_old + i) % len(
_preserved_old_api_keys_for_reset
@@ -214,26 +188,22 @@ async def get_key_manager_instance(api_keys: list = None) -> KeyManager:
if start_key_for_new_cycle and _singleton_instance.api_keys:
try:
# Find the index of the determined start_key in the new api_keys list
target_idx = _singleton_instance.api_keys.index(
start_key_for_new_cycle
)
# Advance the new cycle by calling next() target_idx times
# This positions the cycle so that the *next* call to next() will yield start_key_for_new_cycle
for _ in range(target_idx):
next(_singleton_instance.key_cycle)
logger.info(
f"Key cycle in new instance advanced. Next call to get_next_key() will yield: {start_key_for_new_cycle}"
)
except ValueError:
# This should not happen if start_key_for_new_cycle was correctly found in api_keys
logger.warning(
f"Determined start key '{start_key_for_new_cycle}' not found in new API keys during cycle advancement. "
"New cycle will start from the beginning."
)
except (
StopIteration
): # Should not happen with cycle unless api_keys is empty, handled by _singleton_instance.api_keys check
):
logger.error(
"StopIteration while advancing key cycle, implies empty new API key list previously missed."
)
@@ -254,7 +224,6 @@ async def get_key_manager_instance(api_keys: list = None) -> KeyManager:
# 清理所有保存的状态
_preserved_old_api_keys_for_reset = None
_preserved_next_key_in_cycle = None
# _preserved_failure_counts already cleared
return _singleton_instance
@@ -275,21 +244,16 @@ async def reset_key_manager_instance():
_preserved_old_api_keys_for_reset = _singleton_instance.api_keys.copy()
# 3. 保存 key_cycle 的下一个 key 提示
# This should be the key that get_next_key() would return next.
try:
if (
_singleton_instance.api_keys
): # Only if there are keys to cycle through
# Calling get_next_key() consumes one key and returns it. This is the key
# we want the new cycle to effectively start with.
if _singleton_instance.api_keys:
_preserved_next_key_in_cycle = (
await _singleton_instance.get_next_key()
)
else:
_preserved_next_key_in_cycle = None # No keys, so no next key
_preserved_next_key_in_cycle = None
except (
StopIteration
): # Should be caught by "if _singleton_instance.api_keys"
):
logger.warning(
"Could not preserve next key hint: key cycle was empty or exhausted in old instance."
)
app/service/model/model_service.py
+1 -2
View File
@@ -10,8 +10,7 @@ logger = get_model_logger()
class ModelService:
async def get_gemini_models(self, api_key: str) -> Optional[Dict[str, Any]]:
"""使用 GeminiApiClient 获取并过滤模型列表"""
api_client = GeminiApiClient(base_url=settings.BASE_URL) # 实例化客户端
api_client = GeminiApiClient(base_url=settings.BASE_URL)
gemini_models = await api_client.get_models(api_key)
if gemini_models is None:
app/service/openai_compatiable/openai_compatiable_service.py
-7
View File
@@ -79,7 +79,6 @@ class OpenAICompatiableService:
is_success = False
error_log_msg = str(e)
logger.error(f"Normal API call failed with error: {error_log_msg}")
# Try to parse status code from exception
match = re.search(r"status code (\d+)", error_log_msg)
if match:
status_code = int(match.group(1))
@@ -140,14 +139,12 @@ class OpenAICompatiableService:
logger.warning(
f"Streaming API call failed with error: {error_log_msg}. Attempt {retries} of {max_retries}"
)
# Parse error code for logging
match = re.search(r"status code (\d+)", error_log_msg)
if match:
status_code = int(match.group(1))
else:
status_code = 500
# Log error to error log table
await add_error_log(
gemini_key=current_attempt_key,
model_name=model,
@@ -157,8 +154,6 @@ class OpenAICompatiableService:
request_msg=payload,
)
# Attempt to switch API Key
# Ensure key_manager is available (might need adjustment if not always passed)
if self.key_manager:
api_key = await self.key_manager.handle_api_failure(
current_attempt_key, retries
@@ -178,7 +173,6 @@ class OpenAICompatiableService:
logger.error(f"Max retries ({max_retries}) reached for streaming.")
break
finally:
# Log the final outcome of the streaming request
end_time = time.perf_counter()
latency_ms = int((end_time - start_time) * 1000)
await add_request_log(
@@ -189,7 +183,6 @@ class OpenAICompatiableService:
latency_ms=latency_ms,
request_time=request_datetime,
)
# If the loop finished due to failure, yield error and DONE
if not is_success and retries >= max_retries:
yield f"data: {json.dumps({'error': 'Streaming failed after retries'})}\n\n"
yield "data: [DONE]\n\n"
app/service/request_log/request_log_service.py
+3 -3
View File
@@ -6,12 +6,12 @@ from datetime import datetime, timedelta, timezone
from sqlalchemy import delete
from app import database
from app.database.connection import database
from app.config.config import settings
from app.database.models import RequestLog
from app.log.logger import Logger
from app.log.logger import get_request_log_logger
logger = Logger.setup_logger("request_log_service")
logger = get_request_log_logger()
async def delete_old_request_logs_task():
app/service/stats/stats_service.py
+11 -15
View File
@@ -41,7 +41,7 @@ class StatsService:
),
1,
),
(RequestLog.status_code is None, 1), # type: ignore
(RequestLog.status_code is None, 1),
else_=0,
)
).label("failure"),
@@ -96,7 +96,7 @@ class StatsService:
),
1,
),
(RequestLog.status_code is None, 1), # type: ignore
(RequestLog.status_code is None, 1),
else_=0,
)
).label("failure"),
@@ -166,25 +166,24 @@ class StatsService:
RequestLog.request_time.label("timestamp"),
RequestLog.api_key.label("key"),
RequestLog.model_name.label("model"),
RequestLog.status_code, # We might need to map this to 'success'/'failure' later
RequestLog.status_code,
)
.where(RequestLog.request_time >= start_time)
.order_by(RequestLog.request_time.desc())
) # Order by most recent first
)
results = await database.fetch_all(query)
# Convert results to list of dicts and map status_code
details = []
for row in results:
status = "failure" # 默认状态为 failure，如果 status_code 有效且在 200-299 范围内则更新为 success
if row["status_code"] is not None: # 检查 status_code 是否为空
status = "failure"
if row["status_code"] is not None:
status = "success" if 200 <= row["status_code"] < 300 else "failure"
details.append(
{
"timestamp": row[
"timestamp"
].isoformat(), # Use ISO format for JS compatibility
].isoformat(),
"key": row["key"],
"model": row["model"],
"status": status,
@@ -197,7 +196,6 @@ class StatsService:
except Exception as e:
logger.error(f"Failed to get API call details for period '{period}': {e}")
# Re-raise the exception to be handled by the route
raise
async def get_key_usage_details_last_24h(self, key: str) -> dict | None:
@@ -225,10 +223,10 @@ class StatsService:
.where(
RequestLog.api_key == key,
RequestLog.request_time >= cutoff_time,
RequestLog.model_name.isnot(None), # Ensure model_name is not null
RequestLog.model_name.isnot(None),
)
.group_by(RequestLog.model_name)
.order_by(func.count(RequestLog.id).desc()) # Order by count descending
.order_by(func.count(RequestLog.id).desc())
)
results = await database.fetch_all(query)
@@ -237,7 +235,7 @@ class StatsService:
logger.info(
f"No usage details found for key ending in ...{key[-4:]} in the last 24h."
)
return {} # Return empty dict if no records found
return {}
usage_details = {row["model_name"]: row["call_count"] for row in results}
logger.info(
@@ -250,6 +248,4 @@ class StatsService:
f"Failed to get key usage details for key ending in ...{key[-4:]}: {e}",
exc_info=True,
)
# Depending on requirements, you might return None or raise the exception
# Raising allows the route handler to return a 500 error.
raise # Re-raise the exception
raise
app/service/update/update_service.py
+5 -18
View File
@@ -7,11 +7,7 @@ from app.log.logger import get_update_logger
logger = get_update_logger()
# GitHub repository details are read from settings (defined in app/config/config.py or environment variables)
# GITHUB_API_URL will be constructed inside the function to ensure settings are loaded
VERSION_FILE_PATH = "VERSION" # Path relative to project root
VERSION_FILE_PATH = "VERSION"
async def check_for_updates() -> Tuple[bool, Optional[str], Optional[str]]:
"""
@@ -24,9 +20,6 @@ async def check_for_updates() -> Tuple[bool, Optional[str], Optional[str]]:
- Optional[str]: 如果检查失败，则为错误消息，否则为 None。
"""
try:
# Read current version from VERSION file
# Ensure the path is correct relative to the execution context or use absolute path if needed
# Assuming execution from project root d:/develop/pythonProjects/gemini-balance
with open(VERSION_FILE_PATH, 'r', encoding='utf-8') as f:
current_v = f.read().strip()
if not current_v:
@@ -41,25 +34,22 @@ async def check_for_updates() -> Tuple[bool, Optional[str], Optional[str]]:
logger.info(f"当前应用程序版本 (from {VERSION_FILE_PATH}): {current_v}")
# Check if repository details are configured in settings
if not settings.GITHUB_REPO_OWNER or not settings.GITHUB_REPO_NAME or \
settings.GITHUB_REPO_OWNER == "your_owner" or settings.GITHUB_REPO_NAME == "your_repo":
logger.warning("GitHub repository owner/name not configured in settings. Skipping update check.")
return False, None, "Update check skipped: Repository not configured in settings."
# Construct the API URL inside the function to ensure settings are loaded
github_api_url = f"https://api.github.com/repos/{settings.GITHUB_REPO_OWNER}/{settings.GITHUB_REPO_NAME}/releases/latest"
logger.debug(f"Checking for updates at URL: {github_api_url}") # Log the URL for debugging
logger.debug(f"Checking for updates at URL: {github_api_url}")
try:
async with httpx.AsyncClient(timeout=10.0) as client:
# 添加 User-Agent 头，GitHub API 可能需要
headers = {
"Accept": "application/vnd.github.v3+json",
"User-Agent": f"{settings.GITHUB_REPO_NAME}-UpdateChecker/1.0" # Use repo name from settings for User-Agent
"User-Agent": f"{settings.GITHUB_REPO_NAME}-UpdateChecker/1.0"
}
response = await client.get(github_api_url, headers=headers) # Use the locally constructed URL
response.raise_for_status() # 对错误的 HTTP 状态码（4xx 或 5xx）抛出异常
response = await client.get(github_api_url, headers=headers)
response.raise_for_status()
latest_release = response.json()
latest_v_str = latest_release.get("tag_name")
@@ -68,7 +58,6 @@ async def check_for_updates() -> Tuple[bool, Optional[str], Optional[str]]:
logger.warning("在最新的 GitHub release 响应中找不到 'tag_name'。")
return False, None, "无法从 GitHub 解析最新版本。"
# 移除 tag 名称中可能存在的 'v' 前缀
if latest_v_str.startswith('v'):
latest_v_str = latest_v_str[1:]
@@ -98,8 +87,6 @@ async def check_for_updates() -> Tuple[bool, Optional[str], Optional[str]]:
logger.error(f"检查更新时发生网络错误: {e}")
return False, None, "更新检查期间发生网络错误。"
except version.InvalidVersion:
# Note: latest_v_str might not be defined if the error occurs before fetching it.
# Consider adding a check or default value for logging.
latest_v_str_for_log = latest_v_str if 'latest_v_str' in locals() else 'N/A'
logger.error(f"发现无效的版本格式。当前 (from {VERSION_FILE_PATH}): '{current_v}', 最新: '{latest_v_str_for_log}'")
return False, None, "遇到无效的版本格式。"