mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 14:22:55 +08:00
Update data_source.py
This commit is contained in:
AkashiCoin
GitHub
parent
6a1684d277
commit
b50b934a12
@ -1,124 +1,116 @@
|
|||||||
from httpx import AsyncClient
|
from httpx import AsyncClient
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from nonebot.log import logger
|
from nonebot.log import logger
|
||||||
from nonebot.adapters.cqhttp import Bot, Event, Message, MessageEvent, GroupMessageEvent
|
from nonebot.adapters.cqhttp import Bot, Event, GroupMessageEvent
|
||||||
from configs.config import NICKNAME
|
from configs.config import NICKNAME
|
||||||
|
|
||||||
|
|
||||||
# 获取所有 Epic Game Store 促销游戏
|
# 获取所有 Epic Game Store 促销游戏
|
||||||
# 方法参考：RSSHub /epicgames 路由
|
# 方法参考：RSSHub /epicgames 路由
|
||||||
# https://github.com/DIYgod/RSSHub/blob/master/lib/routes/epicgames/index.js
|
# https://github.com/DIYgod/RSSHub/blob/master/lib/routes/epicgames/index.js
|
||||||
async def get_Epicgame():
|
async def get_epic_game():
|
||||||
epic_url = "https://www.epicgames.com/store/backend/graphql-proxy"
|
epic_url = "https://www.epicgames.com/store/backend/graphql-proxy"
|
||||||
headers = {
|
headers = {
|
||||||
"Referer": "https://www.epicgames.com/store/zh-CN/",
|
"Referer": "https://www.epicgames.com/store/zh-CN/",
|
||||||
"Content-Type": "application/json; charset=utf-8",
|
"Content-Type": "application/json; charset=utf-8",
|
||||||
}
|
|
||||||
data = {
|
|
||||||
"query":
|
|
||||||
"query searchStoreQuery($allowCountries: String, $category: String, $count: Int, $country: String!, $keywords: String, $locale: String, $namespace: String, $sortBy: String, $sortDir: String, $start: Int, $tag: String, $withPrice: Boolean = false, $withPromotions: Boolean = false) {\n Catalog {\n searchStore(allowCountries: $allowCountries, category: $category, count: $count, country: $country, keywords: $keywords, locale: $locale, namespace: $namespace, sortBy: $sortBy, sortDir: $sortDir, start: $start, tag: $tag) {\n elements {\n title\n id\n namespace\n description\n effectiveDate\n keyImages {\n type\n url\n }\n seller {\n id\n name\n }\n productSlug\n urlSlug\n url\n items {\n id\n namespace\n }\n customAttributes {\n key\n value\n }\n categories {\n path\n }\n price(country: $country) @include(if: $withPrice) {\n totalPrice {\n discountPrice\n originalPrice\n voucherDiscount\n discount\n currencyCode\n currencyInfo {\n decimals\n }\n fmtPrice(locale: $locale) {\n originalPrice\n discountPrice\n intermediatePrice\n }\n }\n lineOffers {\n appliedRules {\n id\n endDate\n discountSetting {\n discountType\n }\n }\n }\n }\n promotions(category: $category) @include(if: $withPromotions) {\n promotionalOffers {\n promotionalOffers {\n startDate\n endDate\n discountSetting {\n discountType\n discountPercentage\n }\n }\n }\n upcomingPromotionalOffers {\n promotionalOffers {\n startDate\n endDate\n discountSetting {\n discountType\n discountPercentage\n }\n }\n }\n }\n }\n paging {\n count\n total\n }\n }\n }\n}\n",
|
|
||||||
"variables": {
|
|
||||||
"allowCountries": "CN",
|
|
||||||
"category": "freegames",
|
|
||||||
"count": 1000,
|
|
||||||
"country": "CN",
|
|
||||||
"locale": "zh-CN",
|
|
||||||
"sortBy": "effectiveDate",
|
|
||||||
"sortDir": "asc",
|
|
||||||
"withPrice": True,
|
|
||||||
"withPromotions": True
|
|
||||||
}
|
}
|
||||||
}
|
data = {
|
||||||
async with AsyncClient(headers=headers) as client:
|
"query": "query searchStoreQuery($allowCountries: String, $category: String, $count: Int, $country: String!, $keywords: String, $locale: String, $namespace: String, $sortBy: String, $sortDir: String, $start: Int, $tag: String, $withPrice: Boolean = false, $withPromotions: Boolean = false) {\n Catalog {\n searchStore(allowCountries: $allowCountries, category: $category, count: $count, country: $country, keywords: $keywords, locale: $locale, namespace: $namespace, sortBy: $sortBy, sortDir: $sortDir, start: $start, tag: $tag) {\n elements {\n title\n id\n namespace\n description\n effectiveDate\n keyImages {\n type\n url\n }\n seller {\n id\n name\n }\n productSlug\n urlSlug\n url\n items {\n id\n namespace\n }\n customAttributes {\n key\n value\n }\n categories {\n path\n }\n price(country: $country) @include(if: $withPrice) {\n totalPrice {\n discountPrice\n originalPrice\n voucherDiscount\n discount\n currencyCode\n currencyInfo {\n decimals\n }\n fmtPrice(locale: $locale) {\n originalPrice\n discountPrice\n intermediatePrice\n }\n }\n lineOffers {\n appliedRules {\n id\n endDate\n discountSetting {\n discountType\n }\n }\n }\n }\n promotions(category: $category) @include(if: $withPromotions) {\n promotionalOffers {\n promotionalOffers {\n startDate\n endDate\n discountSetting {\n discountType\n discountPercentage\n }\n }\n }\n upcomingPromotionalOffers {\n promotionalOffers {\n startDate\n endDate\n discountSetting {\n discountType\n discountPercentage\n }\n }\n }\n }\n }\n paging {\n count\n total\n }\n }\n }\n}\n",
|
||||||
try:
|
"variables": {
|
||||||
res = await client.post(epic_url, json=data, timeout=10.0)
|
"allowCountries": "CN",
|
||||||
resJson = res.json()
|
"category": "freegames",
|
||||||
games = resJson['data']['Catalog']['searchStore']['elements']
|
"count": 1000,
|
||||||
return games
|
"country": "CN",
|
||||||
except Exception as e:
|
"locale": "zh-CN",
|
||||||
logger.error(str(e))
|
"sortBy": "effectiveDate",
|
||||||
return None
|
"sortDir": "asc",
|
||||||
|
"withPrice": True,
|
||||||
|
"withPromotions": True,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
async with AsyncClient(proxies={"all://": None}) as client:
|
||||||
|
try:
|
||||||
|
res = await client.post(epic_url, headers=headers, json=data, timeout=10.0)
|
||||||
|
resJson = res.json()
|
||||||
|
games = resJson["data"]["Catalog"]["searchStore"]["elements"]
|
||||||
|
return games
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(str(e))
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
# 获取 Epic Game Store 免费游戏信息
|
# 获取 Epic Game Store 免费游戏信息
|
||||||
# 处理免费游戏的信息方法借鉴 pip 包 epicstore_api 示例
|
# 处理免费游戏的信息方法借鉴 pip 包 epicstore_api 示例
|
||||||
# https://github.com/SD4RK/epicstore_api/blob/master/examples/free_games_example.py
|
# https://github.com/SD4RK/epicstore_api/blob/master/examples/free_games_example.py
|
||||||
async def get_Epicfree(bot: Bot, event: Event):
|
async def get_epic_free(bot: Bot, event: Event):
|
||||||
games = await get_Epicgame()
|
games = await get_epic_game()
|
||||||
if not games:
|
if not games:
|
||||||
return "Epic 可能又抽风啦，请稍后再试（", 404
|
return "Epic 可能又抽风啦，请稍后再试（", 404
|
||||||
else:
|
else:
|
||||||
msg = ""
|
msg_list = []
|
||||||
msg_list = []
|
for game in games:
|
||||||
for game in games:
|
try:
|
||||||
game_name = game['title']
|
msg = ""
|
||||||
game_corp = game['seller']['name']
|
game_name = game["title"]
|
||||||
game_price = game['price']['totalPrice']['fmtPrice']['originalPrice']
|
game_corp = game["seller"]["name"]
|
||||||
# 赋初值以避免 local variable referenced before assignment
|
game_price = game["price"]["totalPrice"]["fmtPrice"]["originalPrice"]
|
||||||
game_dev, game_pub, game_thumbnail = (None, None, None)
|
game_promotions = game["promotions"]["promotionalOffers"]
|
||||||
try:
|
upcoming_promotions = game["promotions"]["upcomingPromotionalOffers"]
|
||||||
game_promotions = game['promotions']['promotionalOffers']
|
if not game_promotions and upcoming_promotions:
|
||||||
upcoming_promotions = game['promotions']['upcomingPromotionalOffers']
|
continue
|
||||||
if not game_promotions and upcoming_promotions:
|
else:
|
||||||
# 促销暂未上线，但即将上线
|
for image in game["keyImages"]:
|
||||||
promotion_data = upcoming_promotions[0]['promotionalOffers'][0]
|
game_thumbnail = (
|
||||||
start_date_iso, end_date_iso = (promotion_data['startDate'][:-1],
|
image["url"] if image["type"] == "Thumbnail" else None
|
||||||
promotion_data['endDate'][:-1])
|
)
|
||||||
# 删除字符串中最后一个 "Z" 使 Python datetime 可处理此时间
|
for pair in game["customAttributes"]:
|
||||||
start_date = datetime.fromisoformat(start_date_iso).strftime(
|
game_dev = (
|
||||||
'%b.%d %H:%M')
|
pair["value"]
|
||||||
end_date = datetime.fromisoformat(end_date_iso).strftime('%b.%d %H:%M')
|
if pair["key"] == "developerName"
|
||||||
if isinstance(event, GroupMessageEvent):
|
else game_corp
|
||||||
_message = '\n由 {} 公司发行的游戏 {} ({}) 在 UTC 时间 {} 即将推出免费游玩，预计截至 {}。'.format(game_corp, game_name, game_price, start_date, end_date)
|
)
|
||||||
data = {
|
game_pub = (
|
||||||
"type": "node",
|
pair["value"]
|
||||||
"data": {
|
if pair["key"] == "publisherName"
|
||||||
"name": f"这里是{NICKNAME}酱",
|
else game_corp
|
||||||
"uin": f"{bot.self_id}",
|
)
|
||||||
"content": _message,
|
game_desp = game["description"]
|
||||||
},
|
end_date = ""
|
||||||
}
|
if len(game["promotions"]["promotionalOffers"]) != 0:
|
||||||
msg_list.append(data)
|
end_date_iso = game["promotions"]["promotionalOffers"][0][
|
||||||
else:
|
"promotionalOffers"
|
||||||
msg = '\n由 {} 公司发行的游戏 {} ({}) 在 UTC 时间 {} 即将推出免费游玩，预计截至 {}。'.format(
|
][0]["endDate"][:-1]
|
||||||
game_corp, game_name, game_price, start_date, end_date)
|
end_date = datetime.fromisoformat(end_date_iso).strftime(
|
||||||
msg_list.append(msg)
|
"%b.%d %H:%M"
|
||||||
else:
|
)
|
||||||
for image in game['keyImages']:
|
# API 返回不包含游戏商店 URL，此处自行拼接，可能出现少数游戏 404 请反馈
|
||||||
if image['type'] == 'Thumbnail':
|
game_url = f"https://www.epicgames.com/store/zh-CN/p/{game['productSlug'].replace('/home', '')}"
|
||||||
game_thumbnail = image['url']
|
msg = (
|
||||||
for pair in game['customAttributes']:
|
f"[CQ:image,file={game_thumbnail}]\n\n"
|
||||||
if pair['key'] == 'developerName':
|
if game_thumbnail
|
||||||
game_dev = pair['value']
|
else ""
|
||||||
if pair['key'] == 'publisherName':
|
)
|
||||||
game_pub = pair['value']
|
msg += f"FREE now :: {game_name} ({game_price})\n\n{game_desp}\n\n"
|
||||||
# 如 game['customAttributes'] 未找到则均使用 game_corp 值
|
msg += (
|
||||||
game_dev = game_dev if game_dev != None else game_corp
|
f"游戏由 {game_pub} 发售，"
|
||||||
game_pub = game_pub if game_pub != None else game_corp
|
if game_dev == game_pub
|
||||||
game_desp = game['description']
|
else f"游戏由 {game_dev} 开发、{game_pub} 出版，"
|
||||||
end_date_iso = game['promotions']['promotionalOffers'][0]['promotionalOffers'][0]['endDate'][:-1]
|
)
|
||||||
end_date = datetime.fromisoformat(end_date_iso).strftime('%b.%d %H:%M')
|
msg += f"将在 UTC 时间 {end_date} 结束免费游玩，戳链接领取吧~\n{game_url}"
|
||||||
# API 返回不包含游戏商店 URL，此处自行拼接，可能出现少数游戏 404 请反馈
|
_message = msg
|
||||||
game_url_part = (game['productSlug'].replace('/home', '')) if ('/home' in game['productSlug']) else game['productSlug']
|
if isinstance(event, GroupMessageEvent):
|
||||||
game_url = 'https://www.epicgames.com/store/zh-CN/p/{}'.format(
|
data = {
|
||||||
game_url_part)
|
"type": "node",
|
||||||
if isinstance(event, GroupMessageEvent):
|
"data": {
|
||||||
_message = '[CQ:image,file={}]\n\nFREE now :: {} ({})\n{}\n此游戏由 {} 开发、{} 发行，将在 UTC 时间 {} 结束免费游玩，戳链接速度加入你的游戏库吧~\n{}\n'.format(game_thumbnail, game_name, game_price, game_desp, game_dev, game_pub,end_date, game_url)
|
"name": f"{NICKNAME}",
|
||||||
data = {
|
"uin": f"{bot.self_id}",
|
||||||
"type": "node",
|
"content": _message,
|
||||||
"data": {
|
},
|
||||||
"name": f"这里是{NICKNAME}酱",
|
}
|
||||||
"uin": f"{bot.self_id}",
|
msg_list.append(data)
|
||||||
"content": _message,
|
else:
|
||||||
},
|
msg_list.append(msg)
|
||||||
}
|
except TypeError as e:
|
||||||
msg_list.append(data)
|
# logger.info(str(e))
|
||||||
else:
|
pass
|
||||||
msg = '[CQ:image,file={}]\n\nFREE now :: {} ({})\n{}\n此游戏由 {} 开发、{} 发行，将在 UTC 时间 {} 结束免费游玩，戳链接速度加入你的游戏库吧~\n{}\n'.format(
|
return msg_list, 200
|
||||||
game_thumbnail, game_name, game_price, game_desp, game_dev, game_pub,
|
|
||||||
end_date, game_url)
|
|
||||||
msg_list.append(msg)
|
|
||||||
except TypeError as e:
|
|
||||||
# logger.info(str(e))
|
|
||||||
pass
|
|
||||||
return msg_list, 200
|
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user