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
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
|
import re
import pytz
import asyncio
from tabulate import tabulate
from datetime import datetime, timedelta
from apscheduler.triggers.base import BaseTrigger
from apscheduler.triggers.combining import AndTrigger
from apscheduler.triggers.interval import IntervalTrigger
from nonebot import get_bot
from nonebot.matcher import Matcher
from nonebot.params import CommandArg, ArgPlainText
from nonebot.permission import Permission, SUPERUSER
from nonebot.adapters.onebot.v11 import Message, MessageSegment, GroupMessageEvent
from ATRI.log import logger as log
from ATRI.utils.apscheduler import scheduler
from ATRI.database import TwitterSubscription
from .data_source import TwitterDynamicSubscriptor
_CONTENT_LIMIT: int = 0
add_sub = TwitterDynamicSubscriptor().cmd_as_group("add", "添加推主订阅")
@add_sub.handle()
async def _td_add_sub(matcher: Matcher, args: Message = CommandArg()):
msg = args.extract_plain_text()
if msg:
matcher.set_arg("td_add_sub_name", args)
@add_sub.got("td_add_sub_name", "推主名呢?速速")
async def _td_deal_add_sub(
event: GroupMessageEvent, _name: str = ArgPlainText("td_add_sub_name")
):
group_id = event.group_id
sub = TwitterDynamicSubscriptor()
t_name, t_screen_name = await sub.get_twitter_username(_name)
if not t_name or not t_screen_name:
await add_sub.finish(f"无法获取名为 {_name} 的推主的信息...操作失败了")
res = await sub.get_twitter_user_info(_name)
tid = res["id"]
query_result = await sub.get_sub_list(tid, group_id)
if len(query_result):
await add_sub.finish(f"该推主 {t_name}@{t_screen_name}\n已在本群订阅列表中啦!")
await sub.add_sub(tid, group_id)
await sub.update_sub(
tid,
group_id,
{
"name": t_name,
"screen_name": t_screen_name,
"last_update": datetime.utcnow(),
},
)
await add_sub.finish(f"成功订阅名为 {t_name}@{t_screen_name} 推主的动态~!")
del_sub = TwitterDynamicSubscriptor().cmd_as_group("del", "删除推主订阅")
@del_sub.handle()
async def _td_del_sub(matcher: Matcher, args: Message = CommandArg()):
msg = args.extract_plain_text()
if msg:
matcher.set_arg("td_del_sub_name", args)
@del_sub.got("td_del_sub_name", "推主名呢?速速")
async def _td_deal_del_sub(
event: GroupMessageEvent, _name: str = ArgPlainText("td_del_sub_name")
):
group_id = event.group_id
sub = TwitterDynamicSubscriptor()
t_name, t_screen_name = await sub.get_twitter_username(_name)
if not t_name or not t_screen_name:
await add_sub.finish(f"无法获取名为 {_name} 的推主的信息...操作失败了")
res = await sub.get_twitter_user_info(_name)
tid = res["id"]
query_result = await sub.get_sub_list(tid=tid, group_id=group_id)
if not query_result:
await del_sub.finish(f"取消订阅失败...该推主 {t_name}@{t_screen_name} 不在本群订阅列表中")
await sub.del_sub(t_screen_name, group_id)
await del_sub.finish(f"成功取消该推主 {t_name}@{t_screen_name} 的订阅~")
get_sub_list = TwitterDynamicSubscriptor().cmd_as_group(
"list", "获取本群推主订阅列表", permission=Permission()
)
@get_sub_list.handle()
async def _td_get_sub_list(event: GroupMessageEvent):
group_id = event.group_id
sub = TwitterDynamicSubscriptor()
query_result = await sub.get_sub_list(group_id=group_id)
if not query_result:
await get_sub_list.finish("本群还未订阅任何推主呢...")
subs = list()
for i in query_result:
tm = i.last_update.replace(tzinfo=pytz.timezone("Asia/Shanghai"))
subs.append([i.name, i.tid, tm + timedelta(hours=8)])
output = "本群订阅的推主列表如下~\n" + tabulate(
subs, headers=["推主", "tid", "最后更新时间"], tablefmt="plain", showindex=True
)
await get_sub_list.finish(output)
limit_content = TwitterDynamicSubscriptor().cmd_as_group(
"limit", "设置订阅内容字数限制", permission=SUPERUSER
)
@limit_content.handle()
async def _td_get_limit(matcher: Matcher, args: Message = CommandArg()):
msg = args.extract_plain_text()
if msg:
matcher.set_arg("td_limit_int", args)
@limit_content.got("td_limit_int", "要限制内容在多少字以内呢?(默认200,0=不限制)")
async def _td_deal_limit(
event: GroupMessageEvent, _limit: str = ArgPlainText("td_limit_int")
):
patt = r"^\d+$"
if not re.match(patt, _limit):
await limit_content.reject("请键入阿拉伯数字:")
global _CONTENT_LIMIT
_CONTENT_LIMIT = int(_limit)
await limit_content.finish(f"成功!订阅内容展示将限制在 {_CONTENT_LIMIT} 以内!")
tq = asyncio.Queue()
class TwitterDynamicChecker(BaseTrigger):
def get_next_fire_time(self, previous_fire_time, now):
sub = TwitterDynamicSubscriptor()
conf = sub.load_service("推特动态订阅")
if conf.get("enabled"):
return now
@scheduler.scheduled_job(
AndTrigger([IntervalTrigger(seconds=30), TwitterDynamicChecker()]),
name="推特动态更新检查",
max_instances=3, # type: ignore
misfire_grace_time=60, # type: ignore
)
async def _check_td():
sub = TwitterDynamicSubscriptor()
try:
all_dy = await sub.get_all_subs()
except Exception:
log.debug("推特订阅列表为空 跳过")
return
if tq.empty():
for i in all_dy:
await tq.put(i)
else:
m: TwitterSubscription = tq.get_nowait()
log.info(f"准备查询推主 {m.name}@{m.screen_name} 的动态,队列剩余 {tq.qsize()}")
ts = m.last_update.timestamp()
info: dict = await sub.get_twitter_user_info(m.screen_name)
if not info.get("status", list()):
log.warning(f"无法获取推主 {m.name}@{m.screen_name} 的动态")
return
tid = info["id"]
t_time = info["status"]["created_at"]
time_patt = "%a %b %d %H:%M:%S +0000 %Y"
raw_t = datetime.strptime(t_time, time_patt) + timedelta(hours=8)
ts_t = raw_t.timestamp()
if ts < ts_t:
raw_media = info["status"]["entities"].get("media", dict())
_pic = raw_media[0]["media_url"] if raw_media else str()
data = {
"name": info["name"],
"content": info["status"]["text"],
}
content = sub.gen_output(data, _CONTENT_LIMIT)
bot = get_bot()
await bot.send_group_msg(group_id=m.group_id, message=content)
if _pic:
pic = Message(MessageSegment.image(_pic))
try:
await bot.send_group_msg(group_id=m.group_id, message=pic)
except Exception:
repo = "图片发送失败了..."
await bot.send_group_msg(group_id=m.group_id, message=repo)
await sub.update_sub(tid, m.group_id, {"last_update": raw_t})
|