import datetime import json import re from loguru import logger from konabot.common.apis.ali_content_safety import AlibabaGreen from konabot.common.llm import get_llm SYSTEM_PROMPT = """你是一个专门解析提醒请求的助手。请分析用户输入,识别其中是否包含提醒信息,并输出标准化的JSON格式结果。 输入格式通常是:"xxxx提醒我yyyy",其中: - xxxx 是用户提供的时间信息 - yyyy 是提醒内容。有些时候用户会有一些需求。你可以在合理的范围进行衍生。 输出要求: - 必须是有效的JSON对象 - 包含以下字段: * datetime: 如果是绝对时间,填入ISO 8601格式的日期时间字符串;否则为null * datetime_delta: 如果是相对时间,填入ISO 8601持续时间格式;否则为null * datetime_delta_minus: 如果时间偏移量是负数,则此项为 true,否则为 false * content: 提醒内容的字符串 * is_notice: 布尔值,表示这是否是真正的提醒请求 时间处理规则: - 绝对时间示例:如果 xxxx 输入了非常明确的时间点,如"2024年12月25日" → 转换为具体datetime - 相对时间示例:如果 xxxx 没有输入非常明确的时间点,如"10分钟后"、"2小时后"、"3天后" → 转换为datetime_delta - 如果用户输入了需要计算的时间,你需要计算出正确的结果,如"10分钟后的8分钟前" → 转换为 “PT2M” - zzzz 是系统提供的时间,每句话肯定都有,这不是你判断相对或绝对时间的依据,需严格按照 xxx 来判断 - datetime和datetime_delta有且仅有一个不为null 时间格式要求: - datetime: "YYYY-MM-DDTHH:MM:SS" (ISO 8601) - datetime_delta: "PxYxMxDTxHxMxS" 格式 (如"PT1H30M"表示1小时30分钟,"P3DT4H"表示三天四小时,"P5MT2M"表示五个月两分钟) 判断标准: - is_notice=true: 明确包含时间+提醒内容的请求 - is_notice=false: 闲聊、疑问句、或不符合提醒格式的内容 示例: 用户:"明天下午2点提醒我开会" 输出:{"datetime": "2024-01-16T14:00:00", "datetime_delta": null, "datetime_delta_minus": false, "content": "开会", "is_notice": true} 用户:"5分钟后提醒我关火" 输出:{"datetime": null, "datetime_delta": "PT5M", "datetime_delta_minus": false, "content": "关火", "is_notice": true} 用户:"5分钟前提醒我关火" 输出:{"datetime": null, "datetime_delta": "PT5M", "datetime_delta_minus": true, "content": "关火", "is_notice": true} 用户:"五百年后提醒我关火" 输出:{"datetime": null, "datetime_delta": "P500Y", "datetime_delta_minus": false, "content": "关火", "is_notice": true} 用户:"昨天提醒我关火" 输出:{"datetime": null, "datetime_delta": "P1D", "datetime_delta_minus": true, "content": "关火", "is_notice": true} 用户:"什么是提醒功能?" 输出:{"datetime": null, "datetime_delta": null, "datetime_delta_minus": false, "content": "", "is_notice": false} 用户:"过一会会,用可爱的语气提醒我该睡觉了" 输出:{"datetime": null, "datetime_delta": "PT10M", "datetime_delta_minus": false, "content": "呼呼!该睡觉了哦!ヾ(•ω•`)o", "is_notice": true} 请严格按照上述格式输出JSON,不要添加任何其他文字说明。现在是 DATETIME""" pt_pattern = re.compile( r"^P" r"((?P\d+)Y)?" r"((?P\d+)M)?" r"((?P\d+)D)?" r"(T((?P\d+)H)?" r"((?P\d+)M)?" r"((?P\d+)S)?)?$" ) def tryint(s: str | None): if s: if re.match(r"^\d+$", s): return int(s) return 0 async def ask_ai(expression: str, now: datetime.datetime | None = None) -> tuple[datetime.datetime | None, str]: if now is None: now = datetime.datetime.now() prompt = SYSTEM_PROMPT.replace("DATETIME", f"{now}, 星期 {now.weekday() + 1}") is_safe = await AlibabaGreen.detect(expression) if not is_safe: logger.info(f"提醒功能:消息被阿里绿网拦截 message={expression}") return None, "" llm = get_llm() message = await llm.chat([ { "role": "system", "content": prompt }, { "role": "user", "content": expression }, ], extra_body={"enable_thinking": False}) result = message.content if result is None: return (None, "") try: data = json.loads(result) except json.JSONDecodeError: logger.info(f"提醒功能:解析 AI 返回值时出现问题 raw={result}") return (None, "") datetime_absolute = data.get("datetime", None) datetime_delta = data.get("datetime_delta", None) is_minus = data.get("datetime_delta_minus", False) content = data.get("content", "") is_notice = data.get("is_notice", False) if not is_notice: return (None, "") if datetime_absolute: try: res = datetime.datetime.strptime(datetime_absolute, "%Y-%m-%dT%H:%M:%S"), content logger.info(f"提醒功能:使用绝对时间解析 AI 返回值 raw={result} target={res[0]}") return res except ValueError: pass if datetime_delta and (match := pt_pattern.match(datetime_delta)): years = tryint(match.group("year")) months = tryint(match.group("month")) days = tryint(match.group("day")) hours = tryint(match.group("hour")) minutes = tryint(match.group("minute")) seconds = tryint(match.group("second")) dt = datetime.timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds) if is_minus: dt = -dt if is_minus: now2 = now.replace(year=now.year - years) m = now2.month if (months - m) >= 0: neg_months = -(m - months - 1) neg_years = (neg_months + 11) // 12 target_month = 12 - ((neg_months - 1) % 12) now2 = now2.replace(year=now2.year - neg_years, month=target_month) else: now2 = now2.replace(month=m - months) else: now2 = now.replace(year=now.year + years) m = now2.month now2 = now2.replace( year=now2.year + (m + months - 1) // 12, month=(m + months - 1) % 12 + 1 ) logger.info(f"提醒功能:使用相对时间解析 AI 返回值 raw={result} target={now2+dt}") return (now2 + dt, content) logger.warning(f"提醒功能:解析 AI 返回值时没有找到解析方法 raw={result}") return (None, "")