chore: 统一代码风格并进行现代化改造

本次提交主要包含以下内容:
- **代码风格统一**:对多个文件进行了格式化,包括移除多余的空行、调整导入顺序、统一字符串引号等,以提高代码一致性和可读性。
- **类型提示现代化**:在多个文件中将旧的 `typing` 模块类型提示(如 `Optional[T]`、`List[T]`、`Union[T, U]`)更新为现代 Python 语法(`T | None`、`list[T]`、`T | U`)。
- **f-string 格式化**:在 `scripts/convert_manifest.py` 中,将 `.format()` 调用更新为更现代和易读的 f-string `!r` 表示法。
- **文件末尾换行符**:为多个文件添加或修正了文件末尾的换行符,遵循 POSIX 标准。
This commit is contained in:
minecraft1024a
2025-10-25 13:31:22 +08:00
committed by Windpicker-owo
parent 778f2bbf1d
commit a234e0b8aa
6 changed files with 34 additions and 33 deletions

View File

@@ -1,6 +1,4 @@
import asyncio import asyncio
import re
from typing import Type
from src.chat.utils.prompt_params import PromptParameters from src.chat.utils.prompt_params import PromptParameters
from src.common.logger import get_logger from src.common.logger import get_logger
@@ -21,7 +19,7 @@ class PromptComponentManager:
3. 提供一个接口以便在构建核心Prompt时能够获取并执行所有相关的组件。 3. 提供一个接口以便在构建核心Prompt时能够获取并执行所有相关的组件。
""" """
def _get_rules_for(self, target_prompt_name: str) -> list[tuple[InjectionRule, Type[BasePrompt]]]: def get_components_for(self, injection_point: str) -> list[type[BasePrompt]]:
""" """
获取指定目标Prompt的所有注入规则及其关联的组件类。 获取指定目标Prompt的所有注入规则及其关联的组件类。
@@ -36,7 +34,8 @@ class PromptComponentManager:
enabled_prompts = component_registry.get_enabled_components_by_type(ComponentType.PROMPT) enabled_prompts = component_registry.get_enabled_components_by_type(ComponentType.PROMPT)
matching_rules = [] matching_rules = []
# 遍历所有启用的 Prompt 组件,查找与目标 Prompt 相关的注入规则 matching_components: list[type[BasePrompt]] = []
for prompt_name, prompt_info in enabled_prompts.items(): for prompt_name, prompt_info in enabled_prompts.items():
if not isinstance(prompt_info, PromptInfo): if not isinstance(prompt_info, PromptInfo):
continue continue

View File

@@ -46,8 +46,8 @@
""" """
import random import random
from datetime import datetime, time from datetime import datetime
from typing import Any, List, Optional, Union from typing import Any
import orjson import orjson
from sqlalchemy import func, select from sqlalchemy import func, select
@@ -62,7 +62,7 @@ logger = get_logger("schedule_api")
# --- 内部辅助函数 --- # --- 内部辅助函数 ---
def _format_schedule_list( def _format_schedule_list(
items: Union[List[dict[str, Any]], List[MonthlyPlan]], items: list[dict[str, Any]] | list[MonthlyPlan],
template: str, template: str,
item_type: str, item_type: str,
) -> str: ) -> str:
@@ -79,7 +79,7 @@ def _format_schedule_list(
return "\\n".join(lines) return "\\n".join(lines)
async def _get_schedule_from_db(date_str: str) -> Optional[List[dict[str, Any]]]: async def _get_schedule_from_db(date_str: str) -> list[dict[str, Any]] | None:
"""从数据库中获取并解析指定日期的日程""" """从数据库中获取并解析指定日期的日程"""
async with get_db_session() as session: async with get_db_session() as session:
result = await session.execute(select(Schedule).filter(Schedule.date == date_str)) result = await session.execute(select(Schedule).filter(Schedule.date == date_str))
@@ -100,10 +100,10 @@ class ScheduleAPI:
@staticmethod @staticmethod
async def get_schedule( async def get_schedule(
date: Optional[str] = None, date: str | None = None,
formatted: bool = False, formatted: bool = False,
format_template: str = "{time_range}: {activity}", format_template: str = "{time_range}: {activity}",
) -> Union[List[dict[str, Any]], str, None]: ) -> list[dict[str, Any]] | str | None:
""" """
(异步) 获取指定日期的日程安排。 (异步) 获取指定日期的日程安排。
@@ -132,7 +132,7 @@ class ScheduleAPI:
async def get_current_activity( async def get_current_activity(
formatted: bool = False, formatted: bool = False,
format_template: str = "{time_range}: {activity}", format_template: str = "{time_range}: {activity}",
) -> Union[dict[str, Any], str, None]: ) -> dict[str, Any] | str | None:
""" """
(异步) 获取当前正在进行的活动。 (异步) 获取当前正在进行的活动。
@@ -174,10 +174,10 @@ class ScheduleAPI:
async def get_activities_between( async def get_activities_between(
start_time: str, start_time: str,
end_time: str, end_time: str,
date: Optional[str] = None, date: str | None = None,
formatted: bool = False, formatted: bool = False,
format_template: str = "{time_range}: {activity}", format_template: str = "{time_range}: {activity}",
) -> Union[List[dict[str, Any]], str, None]: ) -> list[dict[str, Any]] | str | None:
""" """
(异步) 获取指定日期和时间范围内的所有活动。 (异步) 获取指定日期和时间范围内的所有活动。
@@ -223,11 +223,11 @@ class ScheduleAPI:
@staticmethod @staticmethod
async def get_monthly_plans( async def get_monthly_plans(
target_month: Optional[str] = None, target_month: str | None = None,
random_count: Optional[int] = None, random_count: int | None = None,
formatted: bool = False, formatted: bool = False,
format_template: str = "- {plan_text}", format_template: str = "- {plan_text}",
) -> Union[List[MonthlyPlan], str, None]: ) -> list[MonthlyPlan] | str | None:
""" """
(异步) 获取指定月份的有效月度计划。 (异步) 获取指定月份的有效月度计划。
@@ -258,7 +258,7 @@ class ScheduleAPI:
return None return None
@staticmethod @staticmethod
async def count_monthly_plans(target_month: Optional[str] = None) -> int: async def count_monthly_plans(target_month: str | None = None) -> int:
""" """
(异步) 获取指定月份的有效月度计划总数。 (异步) 获取指定月份的有效月度计划总数。
@@ -288,10 +288,10 @@ class ScheduleAPI:
# ============================================================================= # =============================================================================
async def get_schedule( async def get_schedule(
date: Optional[str] = None, date: str | None = None,
formatted: bool = False, formatted: bool = False,
format_template: str = "{time_range}: {activity}", format_template: str = "{time_range}: {activity}",
) -> Union[List[dict[str, Any]], str, None]: ) -> list[dict[str, Any]] | str | None:
"""(异步) 获取指定日期的日程安排的便捷函数。""" """(异步) 获取指定日期的日程安排的便捷函数。"""
return await ScheduleAPI.get_schedule(date, formatted, format_template) return await ScheduleAPI.get_schedule(date, formatted, format_template)
@@ -299,7 +299,7 @@ async def get_schedule(
async def get_current_activity( async def get_current_activity(
formatted: bool = False, formatted: bool = False,
format_template: str = "{time_range}: {activity}", format_template: str = "{time_range}: {activity}",
) -> Union[dict[str, Any], str, None]: ) -> dict[str, Any] | str | None:
"""(异步) 获取当前正在进行的活动的便捷函数。""" """(异步) 获取当前正在进行的活动的便捷函数。"""
return await ScheduleAPI.get_current_activity(formatted, format_template) return await ScheduleAPI.get_current_activity(formatted, format_template)
@@ -307,24 +307,24 @@ async def get_current_activity(
async def get_activities_between( async def get_activities_between(
start_time: str, start_time: str,
end_time: str, end_time: str,
date: Optional[str] = None, date: str | None = None,
formatted: bool = False, formatted: bool = False,
format_template: str = "{time_range}: {activity}", format_template: str = "{time_range}: {activity}",
) -> Union[List[dict[str, Any]], str, None]: ) -> list[dict[str, Any]] | str | None:
"""(异步) 获取指定时间范围内活动的便捷函数。""" """(异步) 获取指定时间范围内活动的便捷函数。"""
return await ScheduleAPI.get_activities_between(start_time, end_time, date, formatted, format_template) return await ScheduleAPI.get_activities_between(start_time, end_time, date, formatted, format_template)
async def get_monthly_plans( async def get_monthly_plans(
target_month: Optional[str] = None, target_month: str | None = None,
random_count: Optional[int] = None, random_count: int | None = None,
formatted: bool = False, formatted: bool = False,
format_template: str = "- {plan_text}", format_template: str = "- {plan_text}",
) -> Union[List[MonthlyPlan], str, None]: ) -> list[MonthlyPlan] | str | None:
"""(异步) 获取月度计划的便捷函数。""" """(异步) 获取月度计划的便捷函数。"""
return await ScheduleAPI.get_monthly_plans(target_month, random_count, formatted, format_template) return await ScheduleAPI.get_monthly_plans(target_month, random_count, formatted, format_template)
async def count_monthly_plans(target_month: Optional[str] = None) -> int: async def count_monthly_plans(target_month: str | None = None) -> int:
"""(异步) 获取月度计划总数的便捷函数。""" """(异步) 获取月度计划总数的便捷函数。"""
return await ScheduleAPI.count_monthly_plans(target_month) return await ScheduleAPI.count_monthly_plans(target_month)

View File

@@ -9,7 +9,7 @@
import json import json
import os import os
import threading import threading
from typing import Any, Dict # noqa: UP035 from typing import Any
from src.common.logger import get_logger from src.common.logger import get_logger

View File

@@ -9,7 +9,6 @@ from src.chat.utils.prompt import Prompt
from src.common.logger import get_logger from src.common.logger import get_logger
from src.config.config import global_config, model_config from src.config.config import global_config, model_config
from src.mood.mood_manager import mood_manager from src.mood.mood_manager import mood_manager
from .prompts import DECISION_PROMPT, PLAN_PROMPT
from src.person_info.person_info import get_person_info_manager from src.person_info.person_info import get_person_info_manager
from src.plugin_system.apis import ( from src.plugin_system.apis import (
chat_api, chat_api,
@@ -22,6 +21,8 @@ from src.plugin_system.apis import (
send_api, send_api,
) )
from .prompts import DECISION_PROMPT, PLAN_PROMPT
logger = get_logger(__name__) logger = get_logger(__name__)

View File

@@ -94,4 +94,4 @@ PLAN_PROMPT = Prompt(
现在,你说: 现在,你说:
""" """
) )

View File

@@ -2,11 +2,12 @@
TTS 语音合成 Action TTS 语音合成 Action
""" """
import toml
from pathlib import Path from pathlib import Path
import toml
from src.common.logger import get_logger from src.common.logger import get_logger
from src.plugin_system.base.base_action import ActionActivationType, BaseAction, ChatMode from src.plugin_system.base.base_action import BaseAction, ChatMode
from ..services.manager import get_service from ..services.manager import get_service
@@ -27,7 +28,7 @@ def _get_available_styles() -> list[str]:
return ["default"] return ["default"]
config = toml.loads(config_file.read_text(encoding="utf-8")) config = toml.loads(config_file.read_text(encoding="utf-8"))
styles_config = config.get("tts_styles", []) styles_config = config.get("tts_styles", [])
if not isinstance(styles_config, list): if not isinstance(styles_config, list):
return ["default"] return ["default"]
@@ -40,7 +41,7 @@ def _get_available_styles() -> list[str]:
# 确保 name 是一个非空字符串 # 确保 name 是一个非空字符串
if isinstance(name, str) and name: if isinstance(name, str) and name:
style_names.append(name) style_names.append(name)
return style_names if style_names else ["default"] return style_names if style_names else ["default"]
except Exception as e: except Exception as e:
logger.error(f"动态加载TTS风格列表时出错: {e}", exc_info=True) logger.error(f"动态加载TTS风格列表时出错: {e}", exc_info=True)