96 lines
4.4 KiB
Python
96 lines
4.4 KiB
Python
import random
|
||
from datetime import datetime
|
||
from framework.core.base_api import BaseAPI
|
||
from framework.config.settings import Config
|
||
from framework.core.logger import get_logger
|
||
|
||
logger = get_logger("MealReminderService")
|
||
|
||
class MealReminderService(BaseAPI):
|
||
MEAL_PERIODS = {
|
||
"breakfast": {
|
||
"range": (6, 9), "emoji": "🌅", "label": "早餐", "keywords": "早餐|早点|咖啡",
|
||
"greetings": ["早安！中关村的早晨从热腾腾的早点开始 ☀️", "美好的早晨，看看附近有什么好吃的早餐 ☕"]
|
||
},
|
||
"lunch": {
|
||
"range": (10, 13), "emoji": "☀️", "label": "午餐", "keywords": "餐厅|快餐|中餐",
|
||
"greetings": ["午饭时间到！放下代码，看看公司附近吃什么 🍚", "到点干饭了，为您搜罗了附近的优质午餐 🤔"]
|
||
},
|
||
"afternoon_tea": {
|
||
"range": (14, 16), "emoji": "🍰", "label": "下午茶", "keywords": "甜品|饮品|咖啡",
|
||
"greetings": ["下午茶时间～来点甜的补充能量 🧁", "三点几嚟，饮茶先啦！🫖"]
|
||
},
|
||
"dinner": {
|
||
"range": (17, 20), "emoji": "🌆", "label": "晚餐", "keywords": "特色菜|火锅|烧烤|餐厅",
|
||
"greetings": ["下班啦！中关村的夜晚，吃顿好的犒劳自己 🎉", "晚餐时间到，看看壹号周边有哪些人气餐厅 🍽️"]
|
||
},
|
||
"late_night": {
|
||
"range": (21, 5), "emoji": "🌙", "label": "夜宵", "keywords": "烧烤|宵夜|快餐",
|
||
"greetings": ["深夜食堂正式营业！🏮", "加班辛苦了，看看附近有什么深夜美食 🌙"]
|
||
},
|
||
}
|
||
|
||
def __init__(self):
|
||
super().__init__("https://restapi.amap.com") # 高德 API 基地址
|
||
|
||
def get_current_period(self) -> dict:
|
||
hour = datetime.now().hour
|
||
for period_info in self.MEAL_PERIODS.values():
|
||
start, end = period_info["range"]
|
||
if start <= end:
|
||
if start <= hour <= end: return period_info
|
||
else:
|
||
if hour >= start or hour <= end: return period_info
|
||
return self.MEAL_PERIODS["lunch"]
|
||
|
||
def fetch_nearby_restaurants(self, keywords: str) -> list:
|
||
endpoint = "/v3/place/around"
|
||
params = {
|
||
"key": Config.AMAP_KEY,
|
||
"location": Config.AMAP_COORDINATES,
|
||
"keywords": keywords,
|
||
"types": "050000", # 餐饮服务
|
||
"radius": 1000,
|
||
"sortrule": "weight",
|
||
"offset": 10,
|
||
"page": 1,
|
||
"extensions": "all"
|
||
}
|
||
try:
|
||
# BaseAPI.request 默认会拼 BASE_URL，这里高德是不同的地址，手动处理或实例化时指定
|
||
import requests # 临时直接用 requests，或者修改 BaseAPI 支持绝对路径
|
||
resp = requests.get(f"{self.base_url}{endpoint}", params=params, timeout=5)
|
||
data = resp.json()
|
||
if data.get("status") == "1":
|
||
return data.get("pois", [])
|
||
except Exception as e:
|
||
logger.error(f"高德 API 调用失败: {e}")
|
||
return []
|
||
|
||
def build_card(self) -> dict:
|
||
period = self.get_current_period()
|
||
greeting = random.choice(period["greetings"])
|
||
pois = self.fetch_nearby_restaurants(period["keywords"])
|
||
|
||
elements = [{"tag": "div", "text": {"tag": "lark_md", "content": f"**{greeting}**"}}, {"tag": "hr"}]
|
||
|
||
if pois:
|
||
for poi in pois[:5]:
|
||
name = poi.get("name", "未知餐厅")
|
||
biz_ext = poi.get("biz_ext", {})
|
||
rating = biz_ext.get("rating") if isinstance(biz_ext, dict) else "暂无"
|
||
distance = poi.get("distance", "未知")
|
||
elements.append({
|
||
"tag": "div",
|
||
"text": {"tag": "lark_md", "content": f"🏠 **{name}**\n⭐️ 评分: {rating} | 📍 距离: {distance}m"}
|
||
})
|
||
else:
|
||
elements.append({"tag": "div", "text": {"tag": "lark_md", "content": "⚠️ 暂时没搜到附近的店"}})
|
||
|
||
elements.extend([{"tag": "hr"}, {"tag": "note", "elements": [{"tag": "plain_text", "content": f"🕒 {datetime.now().strftime('%H:%M')}"}]}])
|
||
|
||
return {
|
||
"header": {"title": {"tag": "plain_text", "content": f"{period['emoji']} 饭否 · {period['label']}实时推荐"}, "template": "orange"},
|
||
"elements": elements
|
||
}
|