Update data_source.py

plugins/epic/data_source.py

@@ -1,124 +1,159 @@
 from httpx import AsyncClient
 from datetime import datetime
 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
 
 
 # 获取所有 Epic Game Store 促销游戏
 # 方法参考：RSSHub /epicgames 路由
 # 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(headers=headers) as client:
+        try:
+            res = await client.post(epic_url, json=data, timeout=10.0)
+            res_json = res.json()
+            games = res_json["data"]["Catalog"]["searchStore"]["elements"]
+            return games
+        except Exception as e:
+            logger.error(str(e))
+            return None
 
 
 # 获取 Epic Game Store 免费游戏信息
-# 处理免费游戏的信息方法借鉴 pip 包 epicstore_api 示例	
+# 处理免费游戏的信息方法借鉴 pip 包 epicstore_api 示例
 # 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:
+            game_name = game["title"]
-      game_name = game['title']
+            game_corp = game["seller"]["name"]
-      game_corp = game['seller']['name']
+            game_price = game["price"]["totalPrice"]["fmtPrice"]["originalPrice"]
-      game_price = game['price']['totalPrice']['fmtPrice']['originalPrice']
+            # 赋初值以避免 local variable referenced before assignment
-      # 赋初值以避免 local variable referenced before assignment
+            game_dev, game_pub, game_thumbnail = (None, None, None)
-      game_dev, game_pub, game_thumbnail = (None, None, None)
+            try:
-      try:
+                game_promotions = game["promotions"]["promotionalOffers"]
-        game_promotions = game['promotions']['promotionalOffers']
+                upcoming_promotions = game["promotions"]["upcomingPromotionalOffers"]
-        upcoming_promotions = game['promotions']['upcomingPromotionalOffers']
+                if not game_promotions and upcoming_promotions:
-        if not game_promotions and upcoming_promotions:
+                    # 促销暂未上线，但即将上线
-          # 促销暂未上线，但即将上线
+                    promotion_data = upcoming_promotions[0]["promotionalOffers"][0]
-          promotion_data = upcoming_promotions[0]['promotionalOffers'][0]
+                    start_date_iso, end_date_iso = (
-          start_date_iso, end_date_iso = (promotion_data['startDate'][:-1],
+                        promotion_data["startDate"][:-1],
-                                          promotion_data['endDate'][:-1])
+                        promotion_data["endDate"][:-1],
-          # 删除字符串中最后一个 "Z" 使 Python datetime 可处理此时间
+                    )
-          start_date = datetime.fromisoformat(start_date_iso).strftime(
+                    # 删除字符串中最后一个 "Z" 使 Python datetime 可处理此时间
-            '%b.%d %H:%M')
+                    start_date = datetime.fromisoformat(start_date_iso).strftime(
-          end_date = datetime.fromisoformat(end_date_iso).strftime('%b.%d %H:%M')
+                        "%b.%d %H:%M"
-          if isinstance(event, GroupMessageEvent):
+                    )
-            _message = '\n由 {} 公司发行的游戏 {} ({}) 在 UTC 时间 {} 即将推出免费游玩，预计截至 {}。'.format(game_corp, game_name, game_price, start_date, end_date)
+                    end_date = datetime.fromisoformat(end_date_iso).strftime(
-            data = {
+                        "%b.%d %H:%M"
-                "type": "node",
+                    )
-                "data": {
+                    if isinstance(event, GroupMessageEvent):
-                    "name": f"这里是{NICKNAME}酱",
+                        _message = "\n由 {} 公司发行的游戏 {} ({}) 在 UTC 时间 {} 即将推出免费游玩，预计截至 {}。".format(
-                    "uin": f"{bot.self_id}",
+                            game_corp, game_name, game_price, start_date, end_date
-                    "content": _message,
+                        )
-                },
+                        data = {
-            }
+                            "type": "node",
-            msg_list.append(data)
+                            "data": {
-          else:
+                                "name": f"这里是{NICKNAME}酱",
-            msg = '\n由 {} 公司发行的游戏 {} ({}) 在 UTC 时间 {} 即将推出免费游玩，预计截至 {}。'.format(
+                                "uin": f"{bot.self_id}",
-              game_corp, game_name, game_price, start_date, end_date)
+                                "content": _message,
-            msg_list.append(msg)
+                            },
-        else:
+                        }
-          for image in game['keyImages']:
+                        msg_list.append(data)
-            if image['type'] == 'Thumbnail':
+                    else:
-              game_thumbnail = image['url']
+                        msg = "\n由 {} 公司发行的游戏 {} ({}) 在 UTC 时间 {} 即将推出免费游玩，预计截至 {}。".format(
-          for pair in game['customAttributes']:
+                            game_corp, game_name, game_price, start_date, end_date
-            if pair['key'] == 'developerName':
+                        )
-              game_dev = pair['value']
+                        msg_list.append(msg)
-            if pair['key'] == 'publisherName':
+                else:
-              game_pub = pair['value']
+                    for image in game["keyImages"]:
-          # 如 game['customAttributes'] 未找到则均使用 game_corp 值
+                        if image["type"] == "Thumbnail":
-          game_dev = game_dev if game_dev != None else game_corp
+                            game_thumbnail = image["url"]
-          game_pub = game_pub if game_pub != None else game_corp
+                    for pair in game["customAttributes"]:
-          game_desp = game['description']
+                        if pair["key"] == "developerName":
-          end_date_iso = game['promotions']['promotionalOffers'][0]['promotionalOffers'][0]['endDate'][:-1]
+                            game_dev = pair["value"]
-          end_date = datetime.fromisoformat(end_date_iso).strftime('%b.%d %H:%M')
+                        if pair["key"] == "publisherName":
-          # API 返回不包含游戏商店 URL，此处自行拼接，可能出现少数游戏 404 请反馈
+                            game_pub = pair["value"]
-          game_url_part = (game['productSlug'].replace('/home', '')) if ('/home' in game['productSlug']) else game['productSlug']
+                    # 如 game['customAttributes'] 未找到则均使用 game_corp 值
-          game_url = 'https://www.epicgames.com/store/zh-CN/p/{}'.format(
+                    game_dev = game_dev if game_dev is not None else game_corp
-            game_url_part)
+                    game_pub = game_pub if game_pub is not None else game_corp
-          if isinstance(event, GroupMessageEvent):
+                    game_desp = game["description"]
-            _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)
+                    end_date_iso = game["promotions"]["promotionalOffers"][0][
-            data = {
+                        "promotionalOffers"
-                "type": "node",
+                    ][0]["endDate"][:-1]
-                "data": {
+                    end_date = datetime.fromisoformat(end_date_iso).strftime(
-                    "name": f"这里是{NICKNAME}酱",
+                        "%b.%d %H:%M"
-                    "uin": f"{bot.self_id}",
+                    )
-                    "content": _message,
+                    # API 返回不包含游戏商店 URL，此处自行拼接，可能出现少数游戏 404 请反馈
-                },
+                    game_url_part = (
-            }
+                        (game["productSlug"].replace("/home", ""))
-            msg_list.append(data)
+                        if ("/home" in game["productSlug"])
-          else:
+                        else game["productSlug"]
-            msg = '[CQ:image,file={}]\n\nFREE now :: {} ({})\n{}\n此游戏由 {} 开发、{} 发行，将在 UTC 时间 {} 结束免费游玩，戳链接速度加入你的游戏库吧~\n{}\n'.format(
+                    )
-              game_thumbnail, game_name, game_price, game_desp, game_dev, game_pub,
+                    game_url = "https://www.epicgames.com/store/zh-CN/p/{}".format(
-              end_date, game_url)
+                        game_url_part
-            msg_list.append(msg)
+                    )
-      except TypeError as e:
+                    if isinstance(event, GroupMessageEvent):
-        # logger.info(str(e))
+                        _message = "[CQ:image,file={}]\n\nFREE now :: {} ({})\n{}\n此游戏由 {} 开发、{} 发行，将在 UTC 时间 {} 结束免费游玩，戳链接速度加入你的游戏库吧~\n{}\n".format(
-        pass
+                            game_thumbnail,
-    return msg_list, 200
+                            game_name,
+                            game_price,
+                            game_desp,
+                            game_dev,
+                            game_pub,
+                            end_date,
+                            game_url,
+                        )
+                        data = {
+                            "type": "node",
+                            "data": {
+                                "name": f"这里是{NICKNAME}酱",
+                                "uin": f"{bot.self_id}",
+                                "content": _message,
+                            },
+                        }
+                        msg_list.append(data)
+                    else:
+                        msg = "[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,
+                        )
+                        msg_list.append(msg)
+            except TypeError as e:
+                # logger.info(str(e))
+                pass
+        return msg_list, 200