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机器人其实是需要一定基础但是网上也有很多无脑教程，跟着搭建一定可以！

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

题外话

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

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

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

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

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

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

最后，提一句与君共勉

——

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

共有 10 条评论

点击这里取消回复。

emoji表情

😀 😃 😄 😁 😆 😅 😂 🤣 ☺️ 😇 🙂 🙃 😉 😌 😍 😘 😗 😙 😚 😋 😜 😝 😛 🤑 🤓 😎 🤡 🤠 😏 😒 🤗 😞 😔 😟 😕 🙁 ☹️ 😣 😖 😫 😩 😤 😠 😡 😶 😐 😑 😯 😦 😧 😮 😲 😵 😳 😱 😨 😰 😢 😥 🤤 😭 😓 😪 😴 🙄 🤔 🤥 😬 🤐 🤢 🤧 😷 🤒 🤕 😣 😖 😫 😩 😤 😠 😡 😶 😐 😑 😯 😦 😧 😮 😲 😵 😳 😱 😨 😰 😢 😥 🤤 😭 😓 😪 😴 🙄 🤔 🤥 😬 🤐 🤢 🤧 😷 🤒 🤕 😈 👿 👹 👺 💩 👻 💀 ☠️ 👽 👾 🤖 🎃 😺 😸 😹 😻 😼 😽 🙀 😿 😾 👐🏻 🙌🏻 👏🏻 🙏🏻 🤝 👍 👎🏻 👊🏻 ✊🏻 🤛🏻 🤜🏻 🤞🏻 ✌🏻 🤘🏻 👌 👈🏻 👉🏻 👆🏻 👇🏻 ☝🏻 ✋🏻 🤚🏻 🖐🏻 🖖🏻 👋🏻 🤙🏻 💪🏻 🖕🏻 ✍🏻 🤳🏻 💅🏻 💍 💄 💋 👄 👅 👂🏻 👃🏻 👣 👀 👤 👥 👶🏻 👦🏻 👧🏻 👨🏻 👩🏻 👱🏻‍♀️ 👱🏻 👴🏻 👵🏻 👲🏻 👳🏻‍♀️ 👳🏻 👮🏻‍♀️ 👮🏻 👷🏻‍♀️ 👷🏻 💂🏻‍♀️ 💂🏻 🕵🏻‍♀️ 🕵🏻 👩🏻‍⚕️ 👨🏻‍⚕️ 👩🏻‍🌾 👩🏻‍🍳 👨🏻‍🍳 👩🏻‍🎓

汐语博主
评论显示ip归属地测试~

2023-02-03 21:56 Edge 109.0.1518.70 Windows 10/11 回复
1. 夏恋
  @汐语：博主可以分享一下怎么添加的吗
  
  2023-02-07 21:52 Chrome 107.0.0.0 Android 11 | V2046A 回复
  1. 汐语博主
    @夏恋：有个叫easy location的插件，你如果找不到的话我发给你，本地上传
    
    2023-02-07 21:56 Chrome 87.0.4280.141 Android 11 | V2046A 回复
  2. 夏恋
    @夏恋：你好能给我邮箱发一下吗
    
    2023-02-07 22:15 Chrome 107.0.0.0 Android 11 | V2046A 回复
    1. 汐语博主
      @夏恋：已发送，请查收
      
      2023-02-07 22:23 Chrome 87.0.4280.141 Android 11 | V2046A 回复
嘉然
博主，cq遇到异地登录，需要在同一网络下登录解决方法是啥呢

2023-02-03 09:28 Chrome 87.0.4280.141 Android 11 | V2046A 回复
1. 汐语博主
  @嘉然：可以尝试用密码和滑条方式登录，如果还是不行的话，就下载爱加速APP，将vpn的位置改到服务器位置，再扫码就可以了
  
  2023-02-03 09:33 Chrome 87.0.4280.141 Android 11 | V2046A 回复
2. 汐语博主
  @嘉然：最好使用爱加速这个软件，相当于把你的地理位置和服务器放到一起了，虚拟定位我试过不太行，估计它是通过网络检测的IP而不是GPS
  
  2023-02-04 20:22 Chrome 87.0.4280.141 Android 11 | V2046A 回复
  1. 夏恋
    @汐语：好嘞，谢谢啦
    
    2023-02-07 21:49 Chrome 107.0.0.0 Android 11 | V2046A 回复
汐语博主
2022再见，迎接更好的2023！ :flower-flower:

2023-02-02 21:17 Edge 109.0.1518.70 Windows 10/11 回复

用Python脚本实现文件监测及提醒，搭配网站使用更佳，基于go-cqhttp的QQBot~

Python脚本监测提醒

实现思路：

监测脚本

Python-CSV

创建文件

增加数据

读取数据

短信发送

验证号码存在

QQ Bot

题外话

相关推荐

MDI Jade-3.15

Matlab安装教程

Alist

Windows的使用技巧

共有 10 条评论

点击这里取消回复。

全站搜索

归档

You-say

2024 年 5 月
一	二	三	四	五	六	日
		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