Python脚本监测提醒

汐语 2023-2-2 1,383 2/2
Python脚本监测提醒

最近突发奇想,想实现一个功能:

用户上传文件资料(图片),python定时检测文件数目,

如有新增,通知管理员审核,并给予“pass,not_pass,未审核”三种状态,

管理员可通过QQ对机器人发送指令进行审批,如审批通过,那么自动发送短信提醒用户

ps:后面感触太多,总结写跑题了,见谅~

实现思路:

监测脚本

用python对某目录下的文件数量进行检测,并定时检测,

我用的是 os库 和 csv库 ,然后用函数封装了一下,方便后续调用我先上代码再具体分析

实现代码如下:

监测脚本,并自动存储在csv文件,状态是 “未审核”

def auto():
​
        csv_keyword = csv.reader(open('Audit.csv', 'r', encoding='UTF8'))
        keywords = []
        for row in csv_keyword:
            keywords.append(row)
        old_num = len(keywords)
        path = r"D:\photos"
        list_add = []
        lists = os.listdir(path)
        lists.sort(key=lambda x: os.path.getmtime((path + "\\" + x)), reverse=True)
        list_new = lists
        num = len(list_new) - old_num
​
        if num != 0:
            for i in range(num):
                list_add.append(list_new[i])
            # for i in range(num):
            #     result = re.findall(r'[0-9]+?[0-9]*', list_new[i])
            #     list_add.append(result[0])
​
            with open(r'./Audit.csv', mode='a', newline='', encoding='utf8') as f:
                wf = csv.writer(f)
                for i in range(len(list_add)):
                    data = [list_add[i], '未审核']
                    a = wf.writerow(data)
            return list_add
        else:
            return 0

Python-CSV

用python操作csv文件实现读取,增加数据,修改数据

创建文件

csv文件将新建于python脚本执行目录下

(第一条数据自己编一条默认数据)

#创建文件:
headers = ['账号', '审核状态']
rows = [('15289991400', 'pass'),]
with open('Audit.csv', 'w+', encoding='utf8', newline='') as f:
writer = csv.writer(f)
writer.writerow(headers)
writer.writerows(rows)

增加数据

可写入多条,以列表嵌套方式写入

#写入
with open(r'./Audit.csv',mode='a',newline='',encoding='utf8') as f:
    wf = csv.writer(f)
    datas = [['1528991400', '未审核'],]
    for i in datas:
        wf.writerow(i)

读取数据

#读取
csv_keyword = csv.reader(open('Audit.csv', 'r',encoding='UTF8'))
keywords = []
for row in csv_keyword:
    keywords.append(row)
print(keywords)
​
for i in range(len(keywords)):
    #此处的id指你要检索的内容,后面的按需更改
    if 'id' in keywords[i][0]:   
        print(keywords[i][-1])

有了上述基础之后,便可实现简单的监测函数

至于修改状态,最后直接看我代码就OK

短信发送

因为我发送短信需求是个人用户的,而非企业,所以市面上的各大云平台大都无法满足要求,但为了实现,我基本上把各种方法都试了,最后还是调用的短信API接口

短信宝 (smsbao.com)

收费制度也相对合理,如果要调用的话,可以联系客服帮你填写模板提高发送速度

(注意:【】发送短信时这个【】里填的叫做签名,中间不能含有空格)

下面是我的实现代码:(只需要填入你的账号密码即可,同时修改发送内容)

def duanxin(phone):
    import urllib.request
​
    def md5(str):
        import hashlib
        m = hashlib.md5()
        m.update(str.encode("utf8"))
        return m.hexdigest()
​
​
    smsapi = "http://api.smsbao.com/"
    # 短信平台账号
    user = '你注册的账号'
    # 短信平台密码
    password = md5('你的密码')
    # 要发送的短信内容
    content = '【汐语】提醒您:尊敬的用户您好,您所提交的图片资料已通过认证,即刻起可享受本平台的活动优惠,祝您生活愉快!'
    # 要发送短信的手机号码
    phone =phone
​
    data = urllib.parse.urlencode({'u': user, 'p': password, 'm': phone, 'c': content})
    send_url = smsapi + 'sms?' + data
    response = urllib.request.urlopen(send_url)
    the_page = response.read().decode('utf-8')
    return the_page

这个函数返回的是状态码,在官方文档也有说明:(具体看这里)

接口说明(‘0’是发送成功的状态码)

验证号码存在

因为我项目数据存储的问题导致我无法验证用户账号是否为手机号,即是否有效的问题

下面是我写的一个验证函数,也是调用的免费接口

def yanzheng(phone):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.26 Safari/537.36 Core/1.63.6788.400 QQBrowser/10.3.2714.400"}
​
    url = f"https://xiaobai.klizi.cn/API/other/mobile.php?phone={phone}"
    response = requests.post(url, headers=headers).json()
    if response['address']=="号码格式错误":
        return "404"
    else:
        return "200"

说明:404 状态码代表 手机号无效

200 表明手机号有效,再进行后续的发送操作,否则会增添不必要的费用

QQ Bot

对接 go-cqhttp实现自动化,这是我的run.py的全部文件

from receive import rev_msg
import socket
import os,time,datetime
import re,csv
import requests
​
​
qq_robot = "你的机器人QQ号"  # 机器人的QQ号
​
​
def send_msg(resp_dict):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
​
    ip = '127.0.0.1'
    client.connect((ip, 9000))
​
    # print(resp_dict)
    msg_type = resp_dict['msg_type']  # 回复类型(群聊/私聊)
    number = resp_dict['number']  # 回复账号(群号/好友号)
    msg = resp_dict['msg']  # 要回复的消息
​
    # 将字符中的特殊字符进行url编码
    msg = msg.replace(" ", "%20")
    msg = msg.replace("\n", "%0a")
​
    if msg_type == 'group':
        payload = "GET /send_group_msg?group_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":9000\r\nConnection: close\r\n\r\n"
    elif msg_type == 'private':
        payload = "GET /send_private_msg?user_id=" + str(
            number) + "&message=" + msg + " HTTP/1.1\r\nHost:" + ip + ":9000\r\nConnection: close\r\n\r\n"
​
    print("发送" + payload)
    client.send(payload.encode("utf-8"))
    client.close()
    return 0
​
​
​
​
​
def auto():
    csv_keyword = csv.reader(open('Audit.csv', 'r', encoding='UTF8'))
    keywords = []
    for row in csv_keyword:
        keywords.append(row)
    old_num = len(keywords)
    path = r"C:\Users\Administrator\PycharmProjects\Blind_helper\media\photos"
    list_add = []
    lists = os.listdir(path)
    lists.sort(key=lambda x: os.path.getmtime((path + "\\" + x)), reverse=True)
    list_new = lists
    num = len(list_new) - old_num
​
    if num != 0:
        if num != 0:
            for i in range(num):
                list_add.append(list_new[i])
​
        with open(r'./Audit.csv', mode='a', newline='', encoding='utf8') as f:
            wf = csv.writer(f)
            for i in range(len(list_add)):
                data = [list_add[i], '未审核']
                a = wf.writerow(data)
        return list_add
    else:
        return 0
​
def read(id):
    csv_keyword = csv.reader(open('Audit.csv', 'r', encoding='UTF8'))
    keywords = []
    for row in csv_keyword:
        keywords.append(row)
​
    for i in range(len(keywords)):
        if id in keywords[i][0]:
            return keywords[i]
def yanzheng(phone):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.26 Safari/537.36 Core/1.63.6788.400 QQBrowser/10.3.2714.400"}
​
    url = f"https://xiaobai.klizi.cn/API/other/mobile.php?phone={phone}"
    response = requests.post(url, headers=headers).json()
    if response['address']=="号码格式错误":
        return "404"
    else:
        return "200"
def duanxin(phone):
    import urllib.request
​
    def md5(str):
        import hashlib
        m = hashlib.md5()
        m.update(str.encode("utf8"))
        return m.hexdigest()
​
​
    smsapi = "http://api.smsbao.com/"
    # 短信平台账号
    user = '你的账号'
    # 短信平台密码
    password = md5('你的密码')
    # 要发送的短信内容
    content = '【汐语】提醒您:尊敬的用户您好,您所提交的图片资料已通过认证,即刻起可享受本平台的活动优惠,祝您生活愉快!'
    # 要发送短信的手机号码
    phone =phone
​
    data = urllib.parse.urlencode({'u': user, 'p': password, 'm': phone, 'c': content})
    send_url = smsapi + 'sms?' + data
    response = urllib.request.urlopen(send_url)
    the_page = response.read().decode('utf-8')
    return the_page
​
​
​
def test():
    id_list = []
    now = datetime.datetime.now()
    group_id = 'xxxxx'#此处是默认发送的群号
    if (now.minute ==00 or now.minute == 30):
        end = auto()
        if end == 0:
            send_msg({'msg_type': 'private', 'number':'1528991400', 'msg':""})
        else:
            num = len(end)
            send_msg(
                {'msg_type': 'group', 'number': group_id, 'msg': f"本次检测新增用户截图{num}条,具体如下:"})
            time.sleep(0.3)
            for i in range(num):
                result = re.findall(r'[0-9]+?[0-9]*', end[i])
                send_msg({'msg_type': 'group', 'number': group_id,
                          'msg': f"第{i + 1}个id:{result[0]}\n"f'[CQ:image,file=http://127.0.0.1:8000/media/photos/{end[i]}]'})
        time.sleep(30)
    while True:
        try:
            rev = rev_msg()
            id = rev['message_id']
            if (len(id_list) >= 50):
                id_list = []
                print(id_list)
​
            if id not in id_list:
                id_list.append(id)
                print(rev)
            else:
                continue
        except:
            continue
​
        if rev["post_type"] == "message":
​
            if rev["message_type"] == "private":  # 私聊
​
                message = rev['raw_message']
                if "检测" in message:
                    end = auto()
                    if end == 0:
                        send_msg({'msg_type': 'private', 'number': '1528991400', 'msg': ""})
                    else:
                        group_id = 'XXXXX'#要发送的群聊号
                        num = len(end)
                        send_msg(
                            {'msg_type': 'group', 'number': group_id, 'msg': f"本次检测新增用户截图{num}条,具体如下:"})
                        time.sleep(0.3)
                        for i in range(num):
                            result = re.findall(r'[0-9]+?[0-9]*', end[i])
                            send_msg({'msg_type': 'group', 'number': group_id, 'msg': f"第{i + 1}个id:{result[0]}\n"f'[CQ:image,file=http://127.0.0.1:8000/media/photos/{end[i]}]'})
                            time.sleep(0.5)
                elif "测试" in message:
                    send_msg({'msg_type': 'private', 'number': '1528991400', 'msg': "test~"})
                else:
                    continue
​
            elif rev["message_type"] == "group":  # 群聊
​
                group = rev['group_id']
​
                if '查询' in rev['raw_message']:
​
                    try:
                        uid=rev['raw_message'].split(' ')[-1]
                        result = read(uid)
                        send_msg(
                            {'msg_type': 'group', 'number': group_id, 'msg':f"id:{uid}\nstatus:{result[-1]}"})
                    except:
                        send_msg(
                            {'msg_type': 'group', 'number': group_id, 'msg':"用户不存在或查询格式有误!"})
​
                elif '通过' in rev['raw_message'] or 'pass' in rev['raw_message']:
                    try:
                        uid=rev['raw_message'].split(' ')[-1]
                        r = csv.reader(open('Audit.csv', 'r', encoding='UTF8'))
                        lines = list(r)
                        for i in lines:
                            if i == []:
                                lines.remove([])
                        for i in lines:
                            if uid in i[0]:
                                i[-1] = 'pass'
​
                        with open(r'./Audit.csv', mode='w', newline='', encoding='utf8') as f:
                            wf = csv.writer(f)
                            for i in lines:
                                a = wf.writerow(i)
                        send_msg({'msg_type': 'group', 'number': group_id, 'msg': f"审批成功!\nid:{uid}\nstatus:pass"})
                        time.sleep(1)
​
                        jg=yanzheng(uid)
                        if jg=='200':
                            result_code=duanxin(uid)
​
                            if result_code=='0':
                                send_msg(
                                    {'msg_type': 'group', 'number': group_id, 'msg': f"id:{uid}\n已发送短信通知用户啦!"})
                            elif result_code=='41':
                                send_msg(
                                    {'msg_type': 'group', 'number': group_id, 'msg': f"id:{uid}\n短信发送失败!\n失败原因:余额不足"})
                            elif result_code=='-2':
                                send_msg(
                                    {'msg_type': 'group', 'number': group_id, 'msg': f"id:{uid}\n短信发送失败!\n失败原因:服务器空间不支持"})
                            elif result_code=='50':
                                send_msg(
                                    {'msg_type': 'group', 'number': group_id, 'msg': f"id:{uid}\n短信发送失败!\n失败原因:内容含有敏感词"})
                            elif result_code=='43':
                                send_msg(
                                    {'msg_type': 'group', 'number': group_id, 'msg': f"id:{uid}\n短信发送失败!\n失败原因:IP地址限制"})
                            else:
                                send_msg({'msg_type': 'group', 'number': group_id, 'msg': f"id:{uid}\n短信发送失败"})
                        else:
                            send_msg(
                                {'msg_type': 'group', 'number': group_id, 'msg': f"id:{uid}\n用户注册时使用的号码不存在,发送短信失败!请管理员予以检测"})
                    except:
                        send_msg({'msg_type': 'group', 'number': group_id, 'msg':"审批失败,请稍后再试~"})
                elif '驳回' in rev['raw_message'] or 'fail' in rev['raw_message']or 'notpass' in rev['raw_message']:
                    try:
                        uid=rev['raw_message'].split(' ')[-1]
                        r = csv.reader(open('Audit.csv', 'r', encoding='UTF8'))
                        lines = list(r)
                        for i in lines:
                            if i == []:
                                lines.remove([])
                        for i in lines:
                            if uid in i[0]:
                                i[-1] = 'fail'
​
                        with open(r'./Audit.csv', mode='w', newline='', encoding='utf8') as f:
                            wf = csv.writer(f)
                            for i in lines:
                                a = wf.writerow(i)
                        send_msg({'msg_type': 'group', 'number': group_id, 'msg': f"修改成功!\nid:{uid}\nstatus:fail"})
​
                    except:
                        send_msg({'msg_type': 'group', 'number': group_id, 'msg': "修改失败,请稍后再试~"})
                elif rev['raw_message']=="检测":
                    end = auto()
                    if end == 0:
                        send_msg({'msg_type': 'group', 'number': group_id, 'msg':"暂无新增哦~"})
                    else:
                        num = len(end)
                        send_msg(
                            {'msg_type': 'group', 'number': group_id, 'msg': f"本次检测新增用户截图{num}条,具体如下:"})
                        time.sleep(0.3)
                        for i in range(num):
                            result = re.findall(r'[0-9]+?[0-9]*', end[i])
                            send_msg({'msg_type': 'group', 'number': group_id,'msg': f"第{i + 1}个id:{result[0]}\n"f'[CQ:image,file=http://47.113.190.152:8000/media/photos/{end[i]}]'})
                            time.sleep(0.5)
                else:
                    continue
            else:
                continue
        else:
            continue
​
​
try:
    send_msg({'msg_type': 'private', 'number': '1528991400', 'msg': '开机成功!'})
    test()
except:
    send_msg({'msg_type': 'private', 'number': '1528991400', 'msg': '坏了,正在重启...'})
    test()
finally:
    send_msg({'msg_type': 'private', 'number': '1528991400', 'msg': '手动重启吧'})

注意:其中

[CQ:image,file=http://127.0.0.1:8000/media/photos/{end[i]}]

类似这种的,其实需要将ip换成服务器的ip,这个相当于发送图片的链接

其中的receive其实是receive.py文件(基于socket库)

import socket
import json
​
ListenSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ListenSocket.bind(('127.0.0.1', 9001))
ListenSocket.listen(100)
​
HttpResponseHeader = '''HTTP/1.1 200 OK
Content-Type: text/html
'''
​
def request_to_json(msg):
    for i in range(len(msg)):
        if msg[i]=="{" and msg[-1]=="\n":
            return json.loads(msg[i:])
    return None
​
#需要循环执行,返回值为json格式
def rev_msg():# json or None
    Client, Address = ListenSocket.accept()
    Request = Client.recv(1024).decode(encoding='utf-8')
    rev_json=request_to_json(Request)
    Client.sendall((HttpResponseHeader).encode(encoding='utf-8'))
    Client.close()
    return rev_json

最后一部分的QQ机器人其实是需要一定基础但是网上也有很多无脑教程,跟着搭建一定可以!

比如我参考的是大佬 皮小孩lsQQ机器人栏目,但是它的全部源码可能需要付费,如果有需要的话,可以在评论区问我要源码~

题外话

如果是在2022,我会迫不及待出个推文如何搭建自己的QQ机器人,但是现在我只想用它来实现一些最实用简单的功能,也无心再去维护了

提到QQ机器人,可能我去年的热情都放在上面了吧,我基本上研究了个彻底,任何报错问题我都遇到过,只不过那时没有记录的习惯,错误来回犯,因此萌生了2023年初搭建博客记录的想法

我去年利用QQ机器人实现了好多自己的想法和功能,比如

爬学校课程表定时提醒上课,每日好友早晚提醒,翻译,互动系统,防撤回功能,对接原神发送语音消息以及各种活跃群的功能等等,

可以用一句话说,我将我的2022献给了QQ机器人,一有时间就研究,各种尝试摸索,同时感觉对自己的影响很大其实,熟悉我的都知道,我甚至搞出了测试群,测试功能,听取建议,现在看来好像挺好笑的,哈哈,学业或多或少受到一定影响吧。

这一年间我也尝试了不同的框架,造过也使用过各种轮子,感触颇多,跑题了,那么就到这里吧

最后,提一句与君共勉

——

其实有想法还是要实现的,趁年轻,敢拼敢闯!

- THE END -

汐语

3月23日20:45

最后修改:2023年3月23日
0

非特殊说明,本博所有文章均为博主原创。

共有 10 条评论

  1. 汐语

    汐语博主

    评论显示ip归属地测试~

    1. 夏恋

      @汐语: 博主可以分享一下怎么添加的吗

      1. 汐语

        汐语博主

        @夏恋: 有个叫easy location的插件,你如果找不到的话我发给你,本地上传

      2. 夏恋

        @夏恋: 你好能给我邮箱发一下吗

        1. 汐语

          汐语博主

          @夏恋: 已发送,请查收

  2. 嘉然

    博主,cq遇到异地登录,需要在同一网络下登录解决方法是啥呢

    1. 汐语

      汐语博主

      @嘉然: 可以尝试用密码和滑条方式登录,如果还是不行的话,就下载爱加速APP,将vpn的位置改到服务器位置,再扫码就可以了

    2. 汐语

      汐语博主

      @嘉然: 最好使用爱加速这个软件,相当于把你的地理位置和服务器放到一起了,虚拟定位我试过不太行,估计它是通过网络检测的IP而不是GPS

      1. 夏恋

        @汐语: 好嘞,谢谢啦

  3. 汐语

    汐语博主

    2022再见,迎接更好的2023! :flower-flower: