import json import random import requests from peewee import fn from models.ips import Ips from models.xstart import Xstart from models.xtoken import XToken url = "http://127.0.0.1:54345" headers = {'Content-Type': 'application/json'} tg_url = "https://web.telegram.org/a/" def createBrowser( groupId=None, host=None, port=None, proxyUserName=None, proxyPassword=None, name='google', proxyType="socks5" ): # 创建或者更新窗口，指纹参数 browserFingerPrint 如没有特定需求，只需要指定下内核即可，如果需要更详细的参数，请参考文档 json_data = { "groupId": groupId, # 分组id 'name': name, # 窗口名称 'remark': '', # 备注 'proxyMethod': 1, # 代理方式 2自定义 3 提取IP # 代理类型 ['noproxy', 'http', 'https', 'socks5', 'ssh'] 'proxyType': proxyType, 'host': host, # 代理主机EE 'port': port, # 代理端口 'proxyUserName': proxyUserName, # 代理账号 'proxyPassword': proxyPassword, # 代理账号 "browserFingerPrint": { # 指纹对象 'coreVersion': '138' # 内核版本，注意，win7/win8/winserver 2012 已经不支持112及以上内核了，无法打开 } } res = requests.post(f"{url}/browser/update", data=json.dumps(json_data), headers=headers).json() print(res) browserId = res['data']['id'] return browserId def updateBrowser(): # 更新窗口，支持批量更新和按需更新，ids 传入数组，单独更新只传一个id即可，只传入需要修改的字段即可，比如修改备注，具体字段请参考文档，browserFingerPrint指纹对象不修改，则无需传入 json_data = {'ids': ['93672cf112a044f08b653cab691216f0'], 'remark': '我是一个备注', 'browserFingerPrint': {}} res = requests.post(f"{url}/browser/update/partial", data=json.dumps(json_data), headers=headers).json() print(res) def openBrowser(id): # 直接指定ID打开窗口，也可以使用 createBrowser 方法返回的ID json_data = {"id": f'{id}', "args": [ # "--disable-application-cache", # "--disable-cache", # "--disable-gpu-shader-disk-cache", # "--media-cache-size=1", # "--disk-cache-size=1", # "--incognito" ]} res = requests.post(f"{url}/browser/open", data=json.dumps(json_data), headers=headers).json() print(res) return res["data"]["http"].split(":")[1] def closeBrowser(id): # 关闭窗口 json_data = {'id': f'{id}'} res = requests.post(f"{url}/browser/close", data=json.dumps(json_data), headers=headers) return res.json() def deleteBrowser(id): # 删除窗口 json_data = {'id': f'{id}'} print(requests.post(f"{url}/browser/delete", data=json.dumps(json_data), headers=headers).json()) def query_bit_browser(page, page_size): data = {"page": page, "pageSize": page_size, 'sort': 'asc'} res = requests.post(f'{url}/browser/list', data=json.dumps(data), headers=headers) return res.json()["data"]["list"] def update_proxy_Browser( id, host, port, proxyType="socks5", proxyUserName="", proxyPassword="" ): json_data = { "ids": [id], # "ipCheckService": "ip123in", "proxyMethod": 2, "proxyType": proxyType, "host": host, "port": port, "proxyUserName": proxyUserName, "proxyPassword": proxyPassword } res = requests.post(f'{url}/browser/proxy/update', data=json.dumps(json_data), headers=headers) print(res.json()) return res.json() def get_group_lists_Browser(): json_data = { "page": 0, "pageSize": 100, "all": True } res = requests.post(f'{url}/group/list', data=json.dumps(json_data), headers=headers) return res.json()["data"]["list"] def get_browser_lists_Browser(id, page=0): json_data = { "groupId": id, "page": page, "pageSize": 100 } res = requests.post(f'{url}/browser/list', data=json.dumps(json_data), headers=headers) return res.json()["data"]["list"] def get_group_lists(): # 获取全部分组的信息 # url = "/group/list" json_data = { "page": 0, "pageSize": 100, "all": True } res = requests.post(f'{url}/group/list', data=json.dumps(json_data), headers=headers) data = {} for i in res.json()["data"]["list"]: data[i["groupName"]] = i["id"] return data def group_add(groupName): json_data = { "groupName": groupName, "sortNum": 0 } res = requests.post(f'{url}/group/add', data=json.dumps(json_data), headers=headers) return res.json() def browser_detail(id): json_data = { "id": id } res = requests.post(f'{url}/browser/detail', data=json.dumps(json_data), headers=headers) return res.json() def group_update(groupId, browserIds): # json_data = { # "groupId": "41notc1202sr8gu5o6emb9ihaqbzbkic", # "browserIds": ["af25e626167f4870b8f257e697bb4f05", "3baa6e990fee4e839c72722c8dc18019"] # } json_data = { "groupId": groupId, "browserIds": browserIds } res = requests.post(f'{url}/browser/group/update', data=json.dumps(json_data), headers=headers) return res.json() if __name__ == '__main__': # for i in Xstart.select().where( # Xstart.x_id.is_null() # ): # ips_info = Ips.select().where(Ips.start == 1, Ips.country == "法国").order_by(fn.Rand()).get() # # update_proxy_Browser( # id=i.bit_id, # host=ips_info.host, # port=ips_info.port, # proxyUserName=ips_info.username, # proxyPassword=ips_info.password # ) # fz_datas = get_group_lists() # # fz_datas['推特'] # # for i in range(10): # for i in get_browser_lists_Browser(id=fz_datas['推特'], page=i): # x_start_info = Xstart.get_or_none( # Xstart.bit_id == i["id"] # ) # # if not x_start_info: # deleteBrowser(id=i["id"]) # # continue # # if x_start_info.start: # continue # # deleteBrowser(id=i["id"]) # # x_start_info.bit_id = None # x_start_info.save() # for i in Xstart.select(): # res = browser_detail(id=i.bit_id) # print(res) # # if not res["success"]: # i.bit_id = None # i.save() # print(browser_detail(id="532651f5330e4caa917e644f9b676b")) # 批量修改代理 for i in Xstart.select().where(Xstart.start == 1): update_proxy_Browser(id=i.bit_id, proxyType="http", host="127.0.0.1", port=random.randint(42000, 42089), ) # fz_datas = get_group_lists() # print(fz_datas) # bit_id_list = [] # for i in XToken.select().where(XToken.account_start == 2): # sql_info = Xstart.get_or_none( # Xstart.x_id == i.id # ) # # bit_id_list.append(sql_info.bit_id) # # print(len(bit_id_list)) # print(bit_id_list) # # print(group_update(fz_datas["西班牙语"], bit_id_list))