-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathsteam_crawler_botV2.py
198 lines (183 loc) · 8.97 KB
/
steam_crawler_botV2.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
import base64
import io
import json
import os
import re
from bs4 import BeautifulSoup as bs
from hoshino import Service
from PIL import Image
from .config import *
from .take_my_money import pic_creater
sv = Service("stbot-steam")
with open(os.path.join(FILE_PATH, "data/tag.json"), "r", encoding="utf-8") as f:
tagdata = json.loads(f.read())
# steam爬虫
def steam_crawler(url_choose: str):
result = []
get_url = other_request(url_choose).text
soup = bs(get_url.replace(r"\n", "").replace(r"\t", "").replace(r"\r", "").replace("\\", ""), "lxml")
row_list = soup.find_all(name="a", class_="search_result_row")
for row in row_list:
appid = row.get("data-ds-appid")
gameinfo = {
"标题": row.find(name="span", class_="title").text,
"链接": row.get("href"),
"appid": appid,
"高分辨率图片": f"https://media.st.dl.pinyuncloud.com/steam/apps/{appid}/capsule_231x87.jpg",
"低分辨率图片": row.find(name="img").get("src"),
}
if not row.find(name="div", class_="discount_pct"):
try:
price = (
row.find(name="div", class_="discount_final_price")
.text.replace("\r", "")
.replace("\n", "")
.replace(" ", "")
)
gameinfo["折扣价"] = " "
if price != "" and "免费" not in price and "Free" not in price:
gameinfo["原价"] = price
elif "免费" in price or "Free" in price:
gameinfo["原价"] = "免费开玩"
else:
gameinfo["原价"] = "无价格信息"
except Exception:
gameinfo["原价"] = "无价格信息"
gameinfo["折扣价"] = " "
else:
discount_price = row.find(name="div", class_="discount_final_price").text.strip().replace(" ", "")
discount_percent = row.find(name="div", class_="discount_pct").text.replace("\n", "").strip()
gameinfo["原价"] = row.find(name="div", class_="discount_original_price").text.strip().replace(" ", "")
gameinfo["折扣价"] = f'{str(discount_price).strip().replace(" ", "")}({discount_percent})'
try:
rate = row.find(name="span", class_="search_review_summary").get("data-tooltip-html")
gameinfo["评测"] = rate.replace("<br>", ",").replace(" ", "")
except Exception:
gameinfo["评测"] = "暂无评测"
try:
tag = row.get("data-ds-tagids").strip("[]").split(",")
tagk = "".join(tagdata["tag_dict"].get(i) + "," for i in tag)
gameinfo["标签"] = tagk.strip(",")
except Exception:
gameinfo["标签"] = "无用户标签"
result.append(gameinfo)
return result
# 根据传入的tag创建tag搜索链接,返回tag搜索链接以及传入的tag中有效的tag(有效tag具体参考data文件夹中的tag.json)
def tagurl_creater(tag: list, page: int):
tag_search_num = "&tags="
tag_name = ""
tag_list = tag
count = f"&start={(page-1)*50}&count=50"
for i in tag_list:
if tagdata["tag_dict"].get(i, "") != "":
tag_search_num += tagdata["tag_dict"][i] + ","
tag_name += f"{i},"
tag_search_url = (
"https://store.steampowered.com/search/results/?l=schinese&query&force_infinite=1&filter=topsellers&category1=998&infinite=1"
+ tag_search_num.strip(",")
+ count
)
return tag_search_url, tag_name.strip(",")
def mes_creater(result: dict):
mes_list = []
for i in range(len(result)):
if result[i]["原价"] in ["免费开玩", "无价格信息"]:
mes = f"[CQ:image,file={result[i]['低分辨率图片']}]\n{result[i]['标题']}\n原价:{result[i]['原价']}\
\n链接:{result[i]['链接']}\n{result[i]['评测']}\n用户标签:{result[i]['标签']}\nappid:{result[i]['appid']}"
else:
mes = f"[CQ:image,file={result[i]['低分辨率图片']}]\n{result[i]['标题']}\n原价:{result[i]['原价']} 折扣价:{result[i]['折扣价']}\
\n链接:{result[i]['链接']}\n{result[i]['评测']}\n用户标签:{result[i]['标签']}\nappid:{result[i]['appid']}"
data = {"type": "node", "data": {"name": "sbeam机器人", "uin": "2854196310", "content": mes}}
mes_list.append(data)
return mes_list
# 匹配关键词发送相关信息,例:今日特惠,发送今日特惠信息,今日新品则发送新品信息
@sv.on_prefix("今日")
async def Gameinfo(bot, ev):
model = ev.message.extract_plain_text().strip()
try:
if model == "新品":
data = steam_crawler(url_new)
elif model == "特惠":
data = steam_crawler(url_specials)
else:
return
except Exception as e:
sv.logger.error(f"Error:{traceback.format_exc()}")
await bot.send(ev, f"哦吼,出错了,报错内容为:{e},请检查运行日志!")
return
await bot.send(ev, "正在生成消息,请稍等片刻!", at_sender=True)
try:
if send_pic_mes:
await bot.send(ev, f"[CQ:image,file={pic_creater(data, is_steam=True)}]")
return
await bot.send_group_forward_msg(group_id=ev["group_id"], messages=mes_creater(data))
except Exception as err:
if "retcode=100" in str(err):
await bot.send(ev, "消息可能被风控,正在转为其他形式发送!")
try:
if send_pic_mes:
await bot.send_group_forward_msg(group_id=ev["group_id"], messages=mes_creater(data))
return
await bot.send(ev, f"[CQ:image,file={pic_creater(data, is_steam=True)}]")
except Exception as err:
if "retcode=100" in str(err):
await bot.send(ev, "消息可能依旧被风控,无法完成发送!")
else:
sv.logger.error(f"Error:{traceback.format_exc()}")
await bot.send(ev, f"发生了其他错误,报错内容为{err},请检查运行日志!")
# 后接格式:页数(阿拉伯数字) 标签1 标签2,例:st搜标签 动作 射击
@sv.on_prefix(("st搜", "St搜", "ST搜", "sT搜"))
async def search_tag(bot, ev):
mes = ev.message.extract_plain_text().strip()
try:
if "标签" in mes:
tags = mes[2:].split(" ")
tagurl = tagurl_creater(tags, 1)
if tagurl[1] == "":
await bot.send(ev, "没有匹配到有效标签")
return
data = steam_crawler(tagurl[0])
elif "游戏" in mes:
gamename = mes[2:]
search_url = f"https://store.steampowered.com/search/results/?l=schinese&query&start=0&count=50&dynamic_data=&sort_by=_ASC&snr=1_7_7_151_7&infinite=1&term={gamename}"
data = steam_crawler(search_url)
if len(data) == 0:
await bot.send(ev, "无搜索结果")
return
except Exception as e:
sv.logger.error(f"Error:{traceback.format_exc()}")
await bot.send(ev, f"哦吼,出错了,报错内容为{e},请检查运行日志!")
return
try:
await bot.send(ev, "正在搜索并生成合并消息中,请稍等片刻!", at_sender=True)
if "标签" in mes:
await bot.send(ev, f"标签{tagurl[1]}搜索结果如下:")
elif "游戏" in mes:
await bot.send(ev, f"游戏{gamename}搜索结果如下:")
else:
return
if send_pic_mes:
await bot.send(ev, f"[CQ:image,file={pic_creater(data, is_steam=True)}]")
return
await bot.send_group_forward_msg(group_id=ev["group_id"], messages=mes_creater(data))
except Exception as err:
if "retcode=100" in str(err):
await bot.send(ev, "消息可能被风控,正在转为其他形式发送!")
try:
if send_pic_mes:
await bot.send_group_forward_msg(group_id=ev["group_id"], messages=mes_creater(data))
return
await bot.send(ev, f"[CQ:image,file={pic_creater(data, is_steam=True)}]")
except Exception as err:
if "retcode=100" in str(err):
await bot.send(ev, "消息可能依旧被风控,无法完成发送!")
else:
sv.logger.error(f"Error:{traceback.format_exc()}")
await bot.send(ev, f"发生了其他错误,报错内容为{err},请检查运行日志!")
@sv.on_fullmatch(("st帮助", "St帮助", "ST帮助", "sT帮助"))
async def help(bot, ev):
helpimg = Image.open(os.path.join(FILE_PATH, "data/help.png"))
b_io = io.BytesIO()
helpimg.save(b_io, format="png")
base64_str = f"base64://{base64.b64encode(b_io.getvalue()).decode()}"
await bot.send(ev, f"[CQ:image,file={base64_str}]")