""" 代码的板块网页接口 """ import json import logging import urllib import requests from utils import tool __tdx_session_data = {} def get_tdx_blocks(code): """ 通达信板块:https://wenda.tdx.com.cn/site/wenda/stock_index.html?message=%E4%B8%9C%E6%96%B9%E9%80%9A%E4%BF%A1%E6%89%80%E5%B1%9E%E6%A6%82%E5%BF%B5 @param code: @return: """ if not __tdx_session_data.get("session"): __tdx_session_data["session"] = requests.Session() __tdx_session_data["session"].get( "https://wenda.tdx.com.cn/TOUCH?Device=Browser&Ip=0.0.0.0&Mac=00-00-00-00-00-00-00-00&Build=WEB&Type=41&Ver=1.0.0&EP=0") session = __tdx_session_data.get("session") try: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 " "Safari/537.36", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8" # "Cookie": f"LST=10; ASPSessionID=" } session.headers.update(headers) data = [{"message": f"{code}的所属概念", "TDXID": "", "wdbk": "", "RANG": "AG", "forward": "1"}] response = session.post("https://wenda.tdx.com.cn/TQL?Entry=NLPSE.StockSelect&RI=", data=json.dumps(data)) response_headers = response.headers if response.status_code == 200: text = response.text result = json.loads(text) # print(result) data = [ {"nlpse_id": result[3][0], "POS": 0, "COUNT": 30, "order_field": "", "dynamic_order": "", "order_flag": "", "timestamps": 0, "op_flag": 1, "screen_type": 1, "RANG": "AG", "forward": "1"}] response = session.post(f"https://wenda.tdx.com.cn/TQL?Entry=NLPSE.NLPQuery&RI={result[0][3]}", data=json.dumps(data)) response.encoding = 'utf-8' if response.status_code == 200: text = response.text print(text) result = json.loads(text) fset = set() if len(result) > 3 and len(result[3]) > 8: fset |= set([x[2:-2] for x in result[3][8].split(";")]) data = [{"sec_code": code, "RANG": "AG"}] response = session.post(f"https://wenda.tdx.com.cn/TQL?Entry=NLPSE.SSTheme&RI=", data=json.dumps(data)) response.encoding = 'utf-8' if response.status_code == 200: text = response.text arr = json.loads(text) if len(arr) > 3: for i in range(3, len(arr)): fset.add(arr[i][2][1:-1]) return fset else: raise Exception(response.text) else: raise Exception(response.text) finally: pass from playwright.sync_api import sync_playwright class THSBlocksApi: __ths_session_data = {} __ths_hexin_v = "AwY2VNcinP6Nq0ia5LnmR52uV_eIZ0ohHKt-hfAv8ikE86itWPeaMew7zofD" def get_ths_blocks(self, code): """ 同花顺板块 @param code: @return: """ if not self.__ths_session_data.get("session"): self.__ths_session_data["session"] = requests.Session() self.__ths_session_data["session"].headers.update({ "content-type": "application/json", "sec-ch-ua": "\"Chromium\";v=\"128\", \"Not;A=Brand\";v=\"24\", \"Google Chrome\";v=\"128\"", "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": "\"Windows\"", "sec-fetch-dest": "document", "sec-fetch-mode": "navigate", "sec-fetch-site": "none", "sec-fetch-user": "?1", "upgrade-insecure-requests": "1", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/128.0.0.0 Safari/537.36 ", "cookie": "other_uid=Ths_iwencai_Xuangu_lqs6bw1f7wshz8rx2c6zrag86ie1r401; ta_random_userid=vtmc00tyg9; " "cid=f35cdfcfd72e8bb1c8e6dbd66fd5df9d1726938491; v=" + self.__ths_hexin_v + "; " "cid" "=f35cdfcfd72e8bb1c8e6dbd66fd5df9d1726938491; ComputerID=f35cdfcfd72e8bb1c8e6dbd66fd5df9d1726938491; WafStatus=0; PHPSESSID=5766f6a83a4ade56ae767161ee9c2990 " }) session = self.__ths_session_data["session"] params = {"source": "Ths_iwencai_Xuangu", "version": "2.0", "query_area": "", "block_list": "", "add_info": "{\"urp" "\":{" "\"scene\":1,\"company\":1,\"business\":1},\"contentType\":\"json\",\"searchInfo\":true}", "question": f"{code}所属板块", "perpage": 50, "page": 1, "secondary_intent": "stock", "log_info": "{\"input_type\":\"typewrite\"}", "rsh": "null"} url = "https://www.iwencai.com/customized/chart/get-robot-data" print(url) response = session.post(url, data=json.dumps(params)) text = response.text index = text.find("\"所属概念\":") if index >= 0: start = text.rfind("{", 0, index) end = text.find("}", index) text = text[start:end + 1] result = json.loads(text) return set(result["所属概念"].split(";")) raise Exception("没有获取到内容：" + text) @classmethod def load_hexin_v(cls): with sync_playwright() as p: # 启动浏览器 browser = p.chromium.launch(headless=True, args=['--disable-blink-features=AutomationControlled']) context = browser.new_context( bypass_csp=True, # 绕过内容安全策略 accept_downloads=True, # 接受下载 ignore_https_errors=True, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/129.0.0.0 Safari/537.38", viewport={"width": 2000, "height": 720} ) page = context.new_page() page.set_extra_http_headers({ "Accept-Language": "zh-CN,zh;q=0.9", "Accept-Encoding": "gzip, deflate, br, zstd", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng," "*/*;q=0.8,application/signed-exchange;v=b3;q=0.7" }) def intercept(route, request): route.continue_() headers = request.headers cookie = headers.get("cookie") if cookie: params = {v.split("=")[0].strip(): v.split("=")[1].strip() for v in cookie.split(";")} try: cls.__ths_hexin_v = params.get("v") print("获取到hexin_v：", cls.__ths_hexin_v) except Exception as e: logging.exception(e) page.route("**/block-detail*", intercept) # 打开网页 page.goto( "https://www.iwencai.com/unifiedwap/result?w=%E7%BE%8E%E7%9A%84%E9%9B%86%E5%9B%A2%E6%89%80%E5%B1%9E%E6%A6%82%E5%BF%B5&querytype=stock") page.wait_for_timeout(4000) # 关闭浏览器 browser.close() # input() pass def get_eastmoney_block(code): """ 获取东方财富板块 @param code: @return: """ params = { "type": "RPT_F10_CORETHEME_BOARDTYPE", "sty": "BOARD_CODE,BOARD_NAME,IS_PRECISE,BOARD_RANK,BOARD_TYPE", "filter": f"(SECUCODE=\"{code}.{'SH' if tool.is_sh_code(code) else 'SZ'}\")", "p": "1", "ps": "", "sr": "1", "st": "BOARD_RANK", "source": "HSF10", "client": "PC", "v": "0734332486357663" } list = [] for k in params: list.append(f"{k}={urllib.parse.quote(params[k])}") url = "https://datacenter.eastmoney.com/securities/api/data/get?" + "&".join(list) response = requests.get(url) if response.status_code == 200: text = response.text print(text) data = json.loads(text) if data["success"]: blocks = [] for x in data["result"]["data"]: # if x["IS_PRECISE"] == "1": blocks.append(x["BOARD_NAME"]) return blocks return None if __name__ == "__main__": pass