# coding=utf-8 import json import sys import random from Struct.User import User import requests import time from datetime import datetime, timedelta DEBUG = False COMMON_GET_PARAM = ( "&iid=96159232732&device_id=55714661189&channel=xiaomi&aid=32&app_name=video_article&version_code=816" "&version_name=8.1.6&device_platform=android&ab_version=941090,785218,668858,1046292,1073579,830454,956074,929436," "797199,1135476,1179370,994679,959010,900042,1113833,668854,1193963,901277,1043330,1038721,994822,1002058,1230687," "1189797,1143356,1143441,1143501,1143698,1143713,1371009,1243997,1392586,1395695,1395486,1398858,668852,668856," "668853,1186421,668851,668859,999124,668855,1039075&device_type=MI+8+SE&device_brand=Xiaomi&language=zh" "&os_api=28&os_version=9&openudid=70d6668d41512c39&manifest_version_code=412&update_version_code=81606" "&_rticket={TIMESTAMP:.0f}&cdid_ts={TIMESTAMP:.0f}&fp=a_fake_fp&tma_jssdk_version=1290000" "&cdid=ed4295e8-5d9a-4cb9-b2a2-04009a3baa2d&oaid=a625f466e0975d42") SEARCH_USER_API = ( "https://security.snssdk.com/video/app/search/live/?format=json&search_sug=0&forum=0&m_tab=live&is_native_req=0" "&offset=0&from=live&en_qc=1&pd=xigua_live&ssmix=a{COMMON}&keyword={keyword}") USER_INFO_API = "https://is.snssdk.com/video/app/user/home/v7/?to_user_id={userId}{COMMON}" COMMON_HEADERS = { "sdk-version": '1', "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 9) VideoArticle/8.1.6 cronet/TTNetVersion:b97574c0 2019-09-24", "Accept-Encoding": "gzip, deflate" } class XiGuaLiveApi: isValidRoom: bool isLive: bool roomLiver: User or None roomID: int def __init__(self, name=None): """ Api类 Init Function :param name: class:str|User: 主播名 """ if name is None: name = "永恒de草薙" if type(name) == User: self.roomLiver = name self.name = name.name else: self.name = str(name) self.isLive = False self.isValidRoom = False self._rawRoomInfo = {} self.roomID = 0 self.roomLiver = None self.roomPopularity = 0 self._cursor = "0" self.s = requests.session() self.s.headers.update(COMMON_HEADERS) self._updRoomAt = datetime.now() self.updRoomInfo(True) def _updateRoomPopularity(self, _data): """ 更新房间人气的方法 Update Room Popularity :param _data: Received Message """ if "extra" in _data: if "member_count" in _data["extra"] and _data["extra"]["member_count"] > 0: self.roomPopularity = _data["extra"]["member_count"] elif "data" in _data: if "popularity" in _data["data"]: self.roomPopularity = _data["data"]["popularity"] @staticmethod def apiChangedError(msg: str, *args): """ API发生更改时的提示 Warning while Detected Api has Changed :param msg: 提示信息 :param args: DEBUG模式下，显示更多信息 """ print(msg) if DEBUG: print(*args) def _checkUsernameIsMatched(self): """ 验证主播名字是自己想要的那个 Check name matched :return: bool: 是否匹配 """ if self.name is None or self.roomLiver is None: return False return self.name == self.roomLiver.__str__() or self.roomLiver.__str__() in self.name or self.name in self.roomLiver.__str__() def _forceSearchUser(self): """ 搜索主播名 :return: """ _results = self.searchUser(self.name) if len(_results) > 0: self.isValidRoom = True self.roomLiver = _results[0] return self._updateUserOnly() def _updateUserOnly(self): """ 获取用户信息 :return: """ if self.roomLiver is None: return False _formatData = {"COMMON": COMMON_GET_PARAM, "TIMESTAMP": time.time() * 1000, "userId": self.roomLiver.ID} _url = USER_INFO_API.format_map(_formatData).format_map(_formatData) try: p = self.s.get(_url) except Exception as e: self.apiChangedError("更新用户信息接口请求失败", e.__str__()) return False try: d = p.json() except Exception as e: self.apiChangedError("更新房间接口错误", e.__str__()) return False self.isValidRoom = d["status"] == 0 if "user_info" not in d and d["user_info"] is None: self.apiChangedError("Api发生改变，请及时联系我", d) return False self.roomLiver = User(d) if not self._checkUsernameIsMatched(): self.isLive = False return False self.isLive = d["user_info"]["is_living"] self._rawRoomInfo = d["user_info"]['live_info'] if self.isLive: self.roomID = d["user_info"]['live_info']['room_id'] return True def updRoomInfo(self, force=False): """ 更新房间信息 :return: """ if not force and self._updRoomAt > (datetime.now() - timedelta(minutes=3)): return self.isLive self._updRoomAt = datetime.now() if self.isValidRoom: return self._updateUserOnly() else: return self._forceSearchUser() @staticmethod def getUserInfoByUserId(userId): """ 通过UserId查找用户的房间号 :param userId: 用户ID :return: XiGuaLiveApi """ _formatData = {"COMMON": COMMON_GET_PARAM, "TIMESTAMP": time.time() * 1000, "userId": userId} _url = USER_INFO_API.format_map(_formatData).format_map(_formatData) try: p = requests.get(_url, headers=COMMON_HEADERS) except Exception as e: XiGuaLiveApi.apiChangedError("更新用户信息接口请求失败", e.__str__()) return None try: d = p.json() except Exception as e: XiGuaLiveApi.apiChangedError("更新房间接口错误", e.__str__()) return None return XiGuaLiveApi(User(d)) @staticmethod def searchUser(keyword): """ 通过关键词搜索主播 :param keyword: 关键词 :return: array: 搜索结果 """ ret = [] _formatData = {"COMMON": COMMON_GET_PARAM, "TIMESTAMP": time.time() * 1000, "keyword": keyword} _url = SEARCH_USER_API.format_map(_formatData).format_map(_formatData) try: p = requests.get(_url) d = p.json() except json.decoder.JSONDecodeError as e: XiGuaLiveApi.apiChangedError("搜索接口错误", e.__str__()) return ret if "data" in d and d["data"] is not None: for i in d["data"]: if i["block_type"] != 0: continue if "cells" not in i or len(i["cells"]) == 0: break for _j in i["cells"]: ret.append(User(_j)) return ret