From 5fc9d1b9da9c53642741659d251ce418068f2888 Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Sat, 25 Oct 2025 13:29:49 +0800 Subject: [PATCH] =?UTF-8?q?refactor(schedule=5Fapi):=20=E9=87=8D=E6=9E=84?= =?UTF-8?q?=E6=97=A5=E7=A8=8B=E4=B8=8E=E8=AE=A1=E5=88=92API=E4=B8=BA?= =?UTF-8?q?=E5=8F=AA=E8=AF=BB=E6=95=B0=E6=8D=AE=E5=BA=93=E6=9F=A5=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将日程API从依赖`schedule_manager`的内存状态改为直接从数据库异步查询。这提高了数据一致性，并解除了模块间的紧密耦合。 主要变更： - **解耦**: 移除对`schedule_manager`的依赖，所有数据直接来自数据库。 - **只读设计**: 移除了`regenerate_schedule`, `ensure_monthly_plans`, `archive_monthly_plans`等写操作API，使API职责更清晰，专注于查询。 - **功能增强**: - `get_schedule` (原`get_today_schedule`) 现在支持查询任意日期的日程。 - `get_monthly_plans` 新增随机抽样功能。 - 新增`get_activities_between`用于查询特定时间范围的活动。 - 新增`count_monthly_plans`用于统计月度计划数量。 - **格式化输出**: 所有查询函数均增加了`formatted`参数，方便插件直接获取格式化后的字符串。 - **文档更新**: 全面更新了模块和函数的文档字符串，以反映新的API设计和用法。 --- src/plugin_system/apis/schedule_api.py | 336 ++++++++++++++++++------- 1 file changed, 243 insertions(+), 93 deletions(-) diff --git a/src/plugin_system/apis/schedule_api.py b/src/plugin_system/apis/schedule_api.py index 61c5d13f4..4273a8c64 100644 --- a/src/plugin_system/apis/schedule_api.py +++ b/src/plugin_system/apis/schedule_api.py @@ -1,180 +1,330 @@ """ -日程表与月度计划API模块 +日程表与月度计划查询API模块 -专门负责日程和月度计划信息的查询与管理，采用标准Python包设计模式 -所有对外接口均为异步函数，以便于插件开发者在异步环境中使用。 +本模块提供了一系列用于查询日程和月度计划的只读接口。 +所有对外接口均为异步函数，专为插件开发者设计，以便在异步环境中无缝集成。 + +核心功能： +- 查询指定日期的日程安排。 +- 获取当前正在进行的活动。 +- 筛选特定时间范围内的活动。 +- 查询月度计划，支持随机抽样和计数。 +- 所有查询接口均提供格式化输出选项。 使用方式： import asyncio from src.plugin_system.apis import schedule_api async def main(): - # 获取今日日程 - today_schedule = await schedule_api.get_today_schedule() + # 获取今天的日程（原始数据） + today_schedule = await schedule_api.get_schedule() if today_schedule: print("今天的日程:", today_schedule) + # 获取昨天的日程，并格式化为字符串 + from datetime import datetime, timedelta + yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") + formatted_schedule = await schedule_api.get_schedule(date=yesterday, formatted=True) + if formatted_schedule: + print(f"\\n昨天的日程 (格式化):\\n{formatted_schedule}") + # 获取当前活动 current_activity = await schedule_api.get_current_activity() if current_activity: - print("当前活动:", current_activity) + print(f"\\n当前活动: {current_activity.get('activity')}") - # 获取本月月度计划 - from datetime import datetime - this_month = datetime.now().strftime("%Y-%m") - plans = await schedule_api.get_monthly_plans(this_month) - if plans: - print(f"{this_month} 的月度计划:", [p.plan_text for p in plans]) + # 获取本月月度计划总数 + plan_count = await schedule_api.count_monthly_plans() + print(f"\\n本月月度计划总数: {plan_count}") + + # 随机获取本月的2个计划 + random_plans = await schedule_api.get_monthly_plans(random_count=2) + if random_plans: + print("\\n随机的2个计划:", [p.plan_text for p in random_plans]) asyncio.run(main()) """ -from datetime import datetime -from typing import Any +import random +from datetime import datetime, time +from typing import Any, List, Optional, Union -from src.common.database.sqlalchemy_models import MonthlyPlan +import orjson +from sqlalchemy import func, select + +from src.common.database.sqlalchemy_models import MonthlyPlan, Schedule, get_db_session from src.common.logger import get_logger from src.schedule.database import get_active_plans_for_month -from src.schedule.schedule_manager import schedule_manager logger = get_logger("schedule_api") +# --- 内部辅助函数 --- + +def _format_schedule_list( + items: Union[List[dict[str, Any]], List[MonthlyPlan]], + template: str, + item_type: str, +) -> str: + """将日程或计划列表格式化为字符串""" + if not items: + return "无" + + lines = [] + for item in items: + if item_type == "schedule" and isinstance(item, dict): + lines.append(template.format(time_range=item.get("time_range", ""), activity=item.get("activity", ""))) + elif item_type == "plan" and isinstance(item, MonthlyPlan): + lines.append(template.format(plan_text=item.plan_text)) + return "\\n".join(lines) + + +async def _get_schedule_from_db(date_str: str) -> Optional[List[dict[str, Any]]]: + """从数据库中获取并解析指定日期的日程""" + async with get_db_session() as session: + result = await session.execute(select(Schedule).filter(Schedule.date == date_str)) + schedule_record = result.scalars().first() + if schedule_record and schedule_record.schedule_data: + try: + return orjson.loads(str(schedule_record.schedule_data)) + except orjson.JSONDecodeError: + logger.warning(f"无法解析数据库中的日程数据 (日期: {date_str})") + return None + + +# --- API实现 --- + + class ScheduleAPI: - """日程表与月度计划API - 负责日程和计划信息的查询与管理""" + """日程表与月度计划查询API""" @staticmethod - async def get_today_schedule() -> list[dict[str, Any]] | None: - """(异步) 获取今天的日程安排 + async def get_schedule( + date: Optional[str] = None, + formatted: bool = False, + format_template: str = "{time_range}: {activity}", + ) -> Union[List[dict[str, Any]], str, None]: + """ + (异步) 获取指定日期的日程安排。 + + Args: + date (Optional[str]): 目标日期，格式 "YYYY-MM-DD"。如果为None，则使用当前日期。 + formatted (bool): 如果为True，返回格式化的字符串；否则返回原始数据列表。 + format_template (str): 当 formatted=True 时使用的格式化模板。 Returns: - Optional[List[Dict[str, Any]]]: 今天的日程列表，如果未生成或未启用则返回None + Union[List[Dict[str, Any]], str, None]: 日程数据或None。 """ + target_date = date or datetime.now().strftime("%Y-%m-%d") try: - logger.debug("[ScheduleAPI] 正在获取今天的日程安排...") - return schedule_manager.today_schedule + logger.debug(f"[ScheduleAPI] 正在获取 {target_date} 的日程安排...") + schedule_data = await _get_schedule_from_db(target_date) + if schedule_data is None: + return None + if formatted: + return _format_schedule_list(schedule_data, format_template, "schedule") + return schedule_data except Exception as e: - logger.error(f"[ScheduleAPI] 获取今日日程失败: {e}") + logger.error(f"[ScheduleAPI] 获取 {target_date} 日程失败: {e}") return None @staticmethod - async def get_current_activity() -> str | None: - """(异步) 获取当前正在进行的活动 + async def get_current_activity( + formatted: bool = False, + format_template: str = "{time_range}: {activity}", + ) -> Union[dict[str, Any], str, None]: + """ + (异步) 获取当前正在进行的活动。 + + Args: + formatted (bool): 如果为True，返回格式化的字符串；否则返回活动字典。 + format_template (str): 当 formatted=True 时使用的格式化模板。 Returns: - Optional[str]: 当前活动名称，如果没有则返回None + Union[Dict[str, Any], str, None]: 当前活动数据或None。 """ try: logger.debug("[ScheduleAPI] 正在获取当前活动...") - return schedule_manager.get_current_activity() + today_schedule = await _get_schedule_from_db(datetime.now().strftime("%Y-%m-%d")) + if not today_schedule: + return None + + now = datetime.now().time() + for event in today_schedule: + time_range = event.get("time_range") + if not time_range: + continue + try: + start_str, end_str = time_range.split("-") + start_time = datetime.strptime(start_str.strip(), "%H:%M").time() + end_time = datetime.strptime(end_str.strip(), "%H:%M").time() + if (start_time <= now < end_time) or \ + (end_time < start_time and (now >= start_time or now < end_time)): + if formatted: + return _format_schedule_list([event], format_template, "schedule") + return event + except (ValueError, KeyError): + continue + return None except Exception as e: logger.error(f"[ScheduleAPI] 获取当前活动失败: {e}") return None @staticmethod - async def regenerate_schedule() -> bool: - """(异步) 触发后台重新生成今天的日程 - - Returns: - bool: 是否成功触发 + async def get_activities_between( + start_time: str, + end_time: str, + date: Optional[str] = None, + formatted: bool = False, + format_template: str = "{time_range}: {activity}", + ) -> Union[List[dict[str, Any]], str, None]: """ - try: - logger.info("[ScheduleAPI] 正在触发后台重新生成日程...") - await schedule_manager.generate_and_save_schedule() - return True - except Exception as e: - logger.error(f"[ScheduleAPI] 触发日程重新生成失败: {e}") - return False - - @staticmethod - async def get_monthly_plans(target_month: str | None = None) -> list[MonthlyPlan]: - """(异步) 获取指定月份的有效月度计划 + (异步) 获取指定日期和时间范围内的所有活动。 Args: - target_month (Optional[str]): 目标月份，格式为 "YYYY-MM"。如果为None，则使用当前月份。 + start_time (str): 开始时间，格式 "HH:MM"。 + end_time (str): 结束时间，格式 "HH:MM"。 + date (Optional[str]): 目标日期，格式 "YYYY-MM-DD"。如果为None，则使用当前日期。 + formatted (bool): 如果为True，返回格式化的字符串；否则返回活动列表。 + format_template (str): 当 formatted=True 时使用的格式化模板。 Returns: - List[MonthlyPlan]: 月度计划对象列表 + Union[List[Dict[str, Any]], str, None]: 在时间范围内的活动列表或None。 """ - if target_month is None: - target_month = datetime.now().strftime("%Y-%m") + target_date = date or datetime.now().strftime("%Y-%m-%d") try: - logger.debug(f"[ScheduleAPI] 正在获取 {target_month} 的月度计划...") - return await get_active_plans_for_month(target_month) + logger.debug(f"[ScheduleAPI] 正在获取 {target_date} 从 {start_time} 到 {end_time} 的活动...") + schedule_data = await _get_schedule_from_db(target_date) + if not schedule_data: + return None + + start = datetime.strptime(start_time, "%H:%M").time() + end = datetime.strptime(end_time, "%H:%M").time() + activities_in_range = [] + + for event in schedule_data: + time_range = event.get("time_range") + if not time_range: + continue + try: + event_start_str, event_end_str = time_range.split("-") + event_start = datetime.strptime(event_start_str.strip(), "%H:%M").time() + if start <= event_start < end: + activities_in_range.append(event) + except (ValueError, KeyError): + continue + + if formatted: + return _format_schedule_list(activities_in_range, format_template, "schedule") + return activities_in_range except Exception as e: - logger.error(f"[ScheduleAPI] 获取 {target_month} 月度计划失败: {e}") - return [] + logger.error(f"[ScheduleAPI] 获取时间段内活动失败: {e}") + return None @staticmethod - async def ensure_monthly_plans(target_month: str | None = None) -> bool: - """(异步) 确保指定月份存在月度计划，如果不存在则触发生成 + async def get_monthly_plans( + target_month: Optional[str] = None, + random_count: Optional[int] = None, + formatted: bool = False, + format_template: str = "- {plan_text}", + ) -> Union[List[MonthlyPlan], str, None]: + """ + (异步) 获取指定月份的有效月度计划。 Args: - target_month (Optional[str]): 目标月份，格式为 "YYYY-MM"。如果为None，则使用当前月份。 + target_month (Optional[str]): 目标月份，格式 "YYYY-MM"。如果为None，则使用当前月份。 + random_count (Optional[int]): 如果设置，将随机返回指定数量的计划。 + formatted (bool): 如果为True，返回格式化的字符串；否则返回对象列表。 + format_template (str): 当 formatted=True 时使用的格式化模板。 Returns: - bool: 操作是否成功 (如果已存在或成功生成) + Union[List[MonthlyPlan], str, None]: 月度计划列表、格式化字符串或None。 """ - if target_month is None: - target_month = datetime.now().strftime("%Y-%m") + month = target_month or datetime.now().strftime("%Y-%m") try: - logger.info(f"[ScheduleAPI] 正在确保 {target_month} 的月度计划存在...") - return await schedule_manager.plan_manager.ensure_and_generate_plans_if_needed(target_month) + logger.debug(f"[ScheduleAPI] 正在获取 {month} 的月度计划...") + plans = await get_active_plans_for_month(month) + if not plans: + return None + + if random_count is not None and random_count > 0 and len(plans) > random_count: + plans = random.sample(plans, random_count) + + if formatted: + return _format_schedule_list(plans, format_template, "plan") + return plans except Exception as e: - logger.error(f"[ScheduleAPI] 确保 {target_month} 月度计划失败: {e}") - return False + logger.error(f"[ScheduleAPI] 获取 {month} 月度计划失败: {e}") + return None @staticmethod - async def archive_monthly_plans(target_month: str | None = None) -> bool: - """(异步) 归档指定月份的月度计划 + async def count_monthly_plans(target_month: Optional[str] = None) -> int: + """ + (异步) 获取指定月份的有效月度计划总数。 Args: - target_month (Optional[str]): 目标月份，格式为 "YYYY-MM"。如果为None，则使用当前月份。 + target_month (Optional[str]): 目标月份，格式 "YYYY-MM"。如果为None，则使用当前月份。 Returns: - bool: 操作是否成功 + int: 有效月度计划的数量。 """ - if target_month is None: - target_month = datetime.now().strftime("%Y-%m") + month = target_month or datetime.now().strftime("%Y-%m") try: - logger.info(f"[ScheduleAPI] 正在归档 {target_month} 的月度计划...") - await schedule_manager.plan_manager.archive_current_month_plans(target_month) - return True + logger.debug(f"[ScheduleAPI] 正在统计 {month} 的月度计划数量...") + async with get_db_session() as session: + result = await session.execute( + select(func.count(MonthlyPlan.id)).where( + MonthlyPlan.target_month == month, MonthlyPlan.status == "active" + ) + ) + return result.scalar_one() or 0 except Exception as e: - logger.error(f"[ScheduleAPI] 归档 {target_month} 月度计划失败: {e}") - return False + logger.error(f"[ScheduleAPI] 统计 {month} 月度计划数量失败: {e}") + return 0 # ============================================================================= # 模块级别的便捷函数 (全部为异步) # ============================================================================= - -async def get_today_schedule() -> list[dict[str, Any]] | None: - """(异步) 获取今天的日程安排的便捷函数""" - return await ScheduleAPI.get_today_schedule() +async def get_schedule( + date: Optional[str] = None, + formatted: bool = False, + format_template: str = "{time_range}: {activity}", +) -> Union[List[dict[str, Any]], str, None]: + """(异步) 获取指定日期的日程安排的便捷函数。""" + return await ScheduleAPI.get_schedule(date, formatted, format_template) -async def get_current_activity() -> str | None: - """(异步) 获取当前正在进行的活动的便捷函数""" - return await ScheduleAPI.get_current_activity() +async def get_current_activity( + formatted: bool = False, + format_template: str = "{time_range}: {activity}", +) -> Union[dict[str, Any], str, None]: + """(异步) 获取当前正在进行的活动的便捷函数。""" + return await ScheduleAPI.get_current_activity(formatted, format_template) -async def regenerate_schedule() -> bool: - """(异步) 触发后台重新生成今天的日程的便捷函数""" - return await ScheduleAPI.regenerate_schedule() +async def get_activities_between( + start_time: str, + end_time: str, + date: Optional[str] = None, + formatted: bool = False, + format_template: str = "{time_range}: {activity}", +) -> Union[List[dict[str, Any]], str, None]: + """(异步) 获取指定时间范围内活动的便捷函数。""" + return await ScheduleAPI.get_activities_between(start_time, end_time, date, formatted, format_template) -async def get_monthly_plans(target_month: str | None = None) -> list[MonthlyPlan]: - """(异步) 获取指定月份的有效月度计划的便捷函数""" - return await ScheduleAPI.get_monthly_plans(target_month) +async def get_monthly_plans( + target_month: Optional[str] = None, + random_count: Optional[int] = None, + formatted: bool = False, + format_template: str = "- {plan_text}", +) -> Union[List[MonthlyPlan], str, None]: + """(异步) 获取月度计划的便捷函数。""" + return await ScheduleAPI.get_monthly_plans(target_month, random_count, formatted, format_template) -async def ensure_monthly_plans(target_month: str | None = None) -> bool: - """(异步) 确保指定月份存在月度计划的便捷函数""" - return await ScheduleAPI.ensure_monthly_plans(target_month) - - -async def archive_monthly_plans(target_month: str | None = None) -> bool: - """(异步) 归档指定月份的月度计划的便捷函数""" - return await ScheduleAPI.archive_monthly_plans(target_month) +async def count_monthly_plans(target_month: Optional[str] = None) -> int: + """(异步) 获取月度计划总数的便捷函数。""" + return await ScheduleAPI.count_monthly_plans(target_month)