"""
|
代码的板块网页接口
|
"""
|
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
|
