|
| 1 | +import datetime |
| 2 | +import json |
| 3 | +import logging |
| 4 | +import threading |
| 5 | +from urllib import request |
| 6 | + |
| 7 | + |
| 8 | +class serverstatus: |
| 9 | + # サーバーステータスAPIのURL |
| 10 | + api_url = "https://game-status-api.ubisoft.com/v1/instances?spaceIds=57e580a1-6383-4506-9509-10a390b7e2f1,05bfb3f7-6c21-4c42-be1f-97a33fb5cf66,96c1d424-057e-4ff7-860b-6b9c9222bdbf,98a601e5-ca91-4440-b1c5-753f601a2c90,631d8095-c443-4e21-b301-4af1a0929c27" |
| 11 | + pc_api_url = "https://game-status-api.ubisoft.com/v1/instances?appIds=e3d5ea9e-50bd-43b7-88bf-39794f4e3d40" |
| 12 | + |
| 13 | + # サーバーステータス辞書を初期化 |
| 14 | + data = {} |
| 15 | + |
| 16 | + # サーバーステータスを取得して整えて返す |
| 17 | + def get(): |
| 18 | + logging.info("サーバーステータスを取得") |
| 19 | + # サーバーステータスを取得する |
| 20 | + res = request.urlopen(request.Request(serverstatus.api_url)) |
| 21 | + #logging.info(str(res.read())) |
| 22 | + res_pc = request.urlopen(request.Request(serverstatus.pc_api_url)) |
| 23 | + #logging.info(str(res_pc.read())) |
| 24 | + |
| 25 | + # ステータスコードが200ではない場合は不明というステータスを返す |
| 26 | + if res.status != 200 or res_pc.status != 200: |
| 27 | + logging.error(f"サーバーステータスの取得に失敗 - ステータスコード: {res.status} / {res_pc.status}") |
| 28 | + status = {"Unknown": {"Status": {"Connectivity": "Unknown", "Authentication": "Unknown", "Leaderboard": "Unknown", "Matchmaking": "Unknown", "Purchase": "Unknown"}, "Maintenance": False}, "_Update_At": datetime.datetime.utcnow().timestamp()} |
| 29 | + return status |
| 30 | + |
| 31 | + raw_status = json.loads(res_pc.read()) |
| 32 | + raw_status.extend(json.loads(res.read())) |
| 33 | + logging.info("取得完了") |
| 34 | + |
| 35 | + # 各プラットフォームのステータスをループ、ステータス辞書に加える |
| 36 | + status = {} |
| 37 | + for s in raw_status: |
| 38 | + p = s["Platform"] |
| 39 | + status[p] = {"Status": {"Connectivity": None, "Authentication": "Operational", "Leaderboard": "Operational", "Matchmaking": "Operational", "Purchase": "Operational"}, "Maintenance": None} |
| 40 | + |
| 41 | + status[p]["Status"]["Connectivity"] = s["Status"] |
| 42 | + if status[p]["Status"]["Connectivity"] == "Online": status[p]["Status"]["Connectivity"] = "Operational" |
| 43 | + |
| 44 | + # ImpactedFeatures をループする リストに含まれる場合は停止中なので、該当するステータスをOutageにする |
| 45 | + for f in s["ImpactedFeatures"]: |
| 46 | + status[p]["Status"][f] = "Outage" |
| 47 | + |
| 48 | + # Maintenance が null の場合はステータスの Maintenance に False をセットする |
| 49 | + if s["Maintenance"] == None: |
| 50 | + status[p]["Maintenance"] = False |
| 51 | + else: |
| 52 | + status[p]["Maintenance"] = s["Maintenance"] |
| 53 | + |
| 54 | + status[p]["_Update_At"] = datetime.datetime.utcnow().timestamp() |
| 55 | + |
| 56 | + #logging.info(str(status)) |
| 57 | + |
| 58 | + logging.info("サーバーステータスの整形完了") |
| 59 | + logging.info(str(status)) |
| 60 | + return status |
| 61 | + |
| 62 | + def update_status(): |
| 63 | + serverstatus.data = serverstatus.get() |
| 64 | + |
| 65 | +def update_serverstatus(): |
| 66 | + serverstatus.update_status() |
| 67 | + logging.info("Server Status Updated") |
| 68 | + |
| 69 | +async def get_serverstatus(platform: list = None): |
| 70 | + status = {} |
| 71 | + |
| 72 | + # パラメーターが指定されていない場合は全てのプラットフォームのステータスを返す |
| 73 | + if platform == None: |
| 74 | + status = serverstatus.data |
| 75 | + if "_last_update" in status.keys(): del status["_last_update"] |
| 76 | + else: |
| 77 | + # 指定されたプラットフォームのステータスだけ返す |
| 78 | + for p in platform: |
| 79 | + d = serverstatus.data.get(p) |
| 80 | + if d != None: |
| 81 | + status[p] = d |
| 82 | + |
| 83 | + # JSONでサーバーステータスのデータを返す |
| 84 | + return status |
| 85 | + |
| 86 | +# サーバーステータスの初回更新 |
| 87 | +serverstatus.update_status() |
0 commit comments