diff --git a/plugins/dns_query/dns_query.py b/plugins/dns_query/dns_query.py index 756e17d446e11192e6ef852c102db9dd715cebee..95dda936caa50145d89e9320e8663ce2675fbe04 100644 --- a/plugins/dns_query/dns_query.py +++ b/plugins/dns_query/dns_query.py @@ -1,22 +1,23 @@ from register import command from global_vars import CONFIG import threading -import json from dns import resolver from util import print_log from io import StringIO + def plugin(): return { - "author":"Antares", - "version":1.0, - "description":"DNS查询" + "author": "Antares", + "version": 1.0, + "description": "DNS查询" } -def A_query(domain:str)->list: + +def A_query(domain: str) -> list: result = [] try: - items = resolver.query(domain,'A') + items = resolver.query(domain, 'A') for i in items.response.answer: for j in i.items: result.append(j.address) @@ -25,21 +26,24 @@ def A_query(domain:str)->list: return [] return result -def MX_query(domain:str)->list: + +def MX_query(domain: str) -> list: result = [] try: - MX_items = resolver.query(domain,'MX') + MX_items = resolver.query(domain, 'MX') for i in MX_items: - result.append(f"MX preference = {i.preference}, main exchanger = {i.exchange}") + result.append( + f"MX preference = {i.preference}, main exchanger = {i.exchange}") except Exception as err: print_log(err) return [] return result -def BASE_query(domain:str,query_mode:str)->list: + +def BASE_query(domain: str, query_mode: str) -> list: result = [] try: - items = resolver.query(domain,query_mode) + items = resolver.query(domain, query_mode) for i in items.response.answer: for j in i.items: result.append(j.to_text()) @@ -48,49 +52,51 @@ def BASE_query(domain:str,query_mode:str)->list: return [] return result -def ALL_query(domain:str,query_mode:str)->list: + +def ALL_query(domain: str, query_mode: str) -> list: if query_mode == 'A': return A_query(domain) elif query_mode == 'MX': return MX_query(domain) else: - return BASE_query(domain,query_mode) + return BASE_query(domain, query_mode) + -@command(name="dns",help ="DNS查询") -def dns_query(bot,context,args): +@command(name="dns", help="DNS查询") +def dns_query(bot, context, args): while args[-1] == "": del args[-1] - + if len(args) == 1: - bot.send(context,"请输入正确的域名") + bot.send(context, "请输入正确的域名") return def handle(): - mode_list = ["A","MX","NS","CNAME"] + mode_list = ["A", "MX", "NS", "CNAME"] if args[1] == "help": - bot.send(context,f"dns 域名 查询类型(可省)\n{mode_list}") + bot.send(context, f"dns 域名 查询类型(可省)\n{mode_list}") return if len(args) > 2: if not args[2] in mode_list: - bot.send(context,f"请输入正确的查询模式:{mode_list}") + bot.send(context, f"请输入正确的查询模式:{mode_list}") return buf = StringIO() buf.write(f"查询域名:{args[1]}\n") buf.write(f"查询模式:{args[2]}\n") buf.write("查询结果:\n") - result = ALL_query(args[1],args[2]) + result = ALL_query(args[1], args[2]) for item in result: buf.write(f"{item}\n") - bot.send(context,buf.getvalue()) - else : + bot.send(context, buf.getvalue()) + else: buf = StringIO() buf.write(f"查询域名:{args[1]}\n查询结果:\n") for opt in mode_list: buf.write(f"{opt}:\n") - result = ALL_query(args[1],opt) + result = ALL_query(args[1], opt) for item in result: buf.write(f"{item}\n") buf.write("\n") - bot.send(context,buf.getvalue()) - threading.Thread(target=handle).start() \ No newline at end of file + bot.send(context, buf.getvalue()) + threading.Thread(target=handle).start() diff --git a/plugins/history/config_default.py b/plugins/history/config_default.py new file mode 100644 index 0000000000000000000000000000000000000000..a390440d6d81102e854376687582bb88351c9889 --- /dev/null +++ b/plugins/history/config_default.py @@ -0,0 +1 @@ +APP_KEY = "" \ No newline at end of file diff --git a/plugins/history/history.py b/plugins/history/history.py new file mode 100644 index 0000000000000000000000000000000000000000..4d4b175ec72a54446245bd36ec9f31ee86b66a3f --- /dev/null +++ b/plugins/history/history.py @@ -0,0 +1,71 @@ +from io import StringIO +from util import print_log +from register import command +from global_vars import CONFIG +import threading +config = CONFIG[__name__] + + +def plugin(): + return { + "author": "Antares", + "version": 1.0, + "description": "历史上的今天" + } + + +def get_histroy(date: str, key: str) -> list: + url = f"http://v.juhe.cn/todayOnhistory/queryEvent.php?key={key}&date={date}" + import json + import urllib.request + try: + with urllib.request.urlopen(url) as urlf: + data = json.JSONDecoder().decode(urlf.read().decode("utf-8")) + return data['result'] + except Exception as err: + print_log(err) + return [] + + +@command(name="histroy", help="历史上的今天，参数为月和日") +def history(bot, context, args): + while args[-1] == "": + del args[-1] + + def handle(): + import datetime + today = datetime.date.today() + date = f"{today.month}/{today.day}" + if len(args) == 3: + try: + month = int(args[1]) + day = int(args[2]) + except ValueError as err: + print_log(err) + bot.send(context, "请输入合法的日期") + return + + if month < 1 or month > 12: + bot.send(context, "月份不合法") + return + + if day < 1 or day > 31: + bot.send(context, "日期不合法") + return + + date = f"{month}/{day}" + + result = get_histroy(date, config.APP_KEY)[::-1] + if len(result) == 0: + bot.send(context, "无结果") + return + + buf = StringIO() + buf.write(f"查询日期: {date}\n\n") + count = 0 + for item in result: + buf.write(f"{item['date']} {item['title']}\n") + + bot.send(context,buf.getvalue()) + + threading.Thread(target=headle).start()