别忘记充电费呀

由于夏天到了，我们宿舍的电费总是会很快就用完，有时候半夜的时候突然停电，真是令人痛苦一整夜（主要是因为我们学校的微信接口在夜间会关闭，所以夜里我们是没法通过微信的接口去充电费的）所以为了能够有一个方法来提醒我们什么时候需要去充电费就显的格外的重要。于是这个项目便诞生了。

目录结构

.
├── __pycache__
│   └── eleFee.cpython-39.pyc
├── count.json
├── eleFee.py
└── email_send.py

主要模块讲述

这个小 demo 其实很简单的，主要是上述的三个文件，分别是 count.json,eleFee.py, 以及 email_send.py，这几个文件的主要作用分别是：

下面就可以贴代码了：

• count,json

  {"email_15": 0, "email_30": 0}
  

• eleFee.py

  import requests
  import re
  
  # 参考：https://blog.nanshaobit.top/106
  
  s = requests.Session()def query():
      '''
      查询电费的函数
      return: 电费字符串和电费值
      '''
      url = "http://wxxy.cczu.edu.cn/wechat/callinterface/queryElecRoomInfo.html"
      request_data = {
          "aid": "*****************",
          "account": "******",
          "area": '{"area":"***校区 ","areaname":"****校区 "}',
          "building": '{"building":"**","buildingid":"**"}',
          "floor": '{"floorid":"","floor":""}',
          "room":'{"room":"","roomid":"******"}'
      }
      resp = s.post(url, data=request_data)
      # print(resp.json())
      fee = resp.json()["errmsg"]
      print(fee)
      fee_value = float(re.findall(r"\d+\.?\d*", fee)[0])
      print(fee_value)
      return {"resp": resp, "fee": fee, "fee_value": fee_value}
  

• email_send.py

  # smtplib 用于邮件的发信动作
  import smtplib
  # email 用于构建邮件内容
  from email.mime.text import MIMEText
  # 构建邮件头
  from email.header import Header
  import eleFee
  import time
  import json
  from apscheduler.schedulers.blocking import BlockingScheduler
  from datetime import datetime
  
  alert_fee_value = 30
  alarm_fee_value = 15
  cur_time = time.localtime()fee_info = eleFee.query()
  
  
  def sendmail(email_msg: str)->str:
      type = 'nan'
      # 发信方的信息：发信邮箱，邮箱授权码
      from_addr = '*********************'
      password = '***********************'
      # 收信方邮箱
      to_addr = '*****************'
      # 发信服务器
      smtp_server = '**************'
  
      # 邮箱正文内容，第一个参数为内容，第二个参数为格式 (plain 为纯文本)，第三个参数为编码
      msg = MIMEText(email_msg, 'plain', 'utf-8')
      # 邮件头信息
      msg['From'] = Header('知不道')  # 发送者
      # msg['To'] = Header('李四')  # 接收者
      subject = '电费余量提醒'
      msg['Subject'] = Header(subject, 'utf-8')  # 邮件主题
  
      try:
          smtpobj = smtplib.SMTP_SSL(smtp_server)
          # 建立连接 --qq 邮箱服务和端口号（可百度查询）
          smtpobj.connect(smtp_server, 465)
          # 登录 -- 发送者账号和口令
          smtpobj.login(from_addr, password)
          # 发送邮件
          smtpobj.sendmail(from_addr, to_addr, msg.as_string()
          print("邮件发送成功")
          type = 'ok'
      except smtplib.SMTPException:
          print("无法发送邮件")
          type = 'no'
      finally:
          # 关闭服务器
          smtpobj.quit()
      return type
  
  
  def start():
      '''
      入口函数，加入了定时任务的功能，监测是否需要重复发送信息的功能。
      '''
      if fee_info["fee_value"] <= alarm_fee_value:  # 警告值 15
          cur_time_str = time.strftime("%Y 年 %m 月 %d 日 %H:%M:%S", cur_time)
          alarm_str = "【警告】截至" + cur_time_str + "，当前 302 电费余量为" + str(fee_info["fee_value"]) + "，小于预警值" + str(alarm_fee_value) + "，请立即充值!"
          print(alarm_str)
          with open("count.json", 'r') as counter:
              count = json.load(counter)
          if count["email_15"] == 0:
              sendmail(alarm_str)
              count["email_15"] = 1
              count["email_30"] = 0
              with open("count.json", 'w') as counters:
                  json.dump(count, counters)
          elif count["email_15"] == 1:
              pass
  
      elif fee_info["fee_value"] <= alert_fee_value:  # 预警值 50
          cur_time_str = time.strftime("%Y 年 %m 月 %d 日 %H:%M:%S", cur_time)
          alarm_str = "【注意】截至" + cur_time_str + "，当前 302 电费余量为" + str(fee_info["fee_value"]) + "，小于提醒值" + str(alert_fee_value) + "，请准备充值!"
          print(alarm_str)
          with open("count.json", 'r') as counter:
              count = json.load(counter)
          if count["email_30"] == 0:
              sendmail(alarm_str)
              count["email_30"] = 1
  
              # count["email_15"] = 0
              with open("count.json", 'w') as counters:
                  json.dump(count, counters)
          elif count["email_30"] == 1:
              print("已经发送过邮件")
              pass
      else:
          with open("count.json", 'w') as counters:
              count = {"email_15": 0, "email_30": 0}
              json.dump(count, counters)
              print("已低于警戒值，已重置邮件计数")
      '''
      简单解释一下函数的运行的内容，主要是判断是否需要发送邮件，但是我们不能够一碰到这种情况就发送邮件，这样会导致接受消息的用户被邮件轰炸，
      所以判断是否需要发送邮件应该是状态的变化，比如我们第一次发现电量低于 30，则我们立刻发送邮件但在下一次中电量也会超过 30，怎么处理呢，有一种办法
      就是引入数据库对数据库进行操作，比如把当前时间和电量值存入数据库，然后每次发送邮件的时候，查询数据库，如果当前时间和电量值变化小于一个范围
      ，则不发送邮件，但是这是一个小项目，而且项目本身也不是一直在运行，所以这种方法不是很好，所以我们可以使用一个定时任务，每隔一段时间请求一下数据，然后
      根据是否发送过邮件来判断是否需要发送邮件，这样就可以解决这个问题。
      '''
  
  
  
  if __name__ == "__main__":
      sched = BlockingScheduler()sched.add_job(start, 'interval', seconds=30)
      sched.start()
  

  只需要将代码中 ********* 部分填入你的参数，就可以定时给你发送电量预警邮件，确保我们能够及时的充电费，不会有断电的风险。
  
  上图展示的是运行后的结果，只需要将发送方设为全宿舍的所有的邮箱地址就可以啦！
  
  将代码部署在云服务器上运行，更加的高效。另外这个代码也可以在云函数上进行部署，但是云函数上 count.json 无法使用（因为云函数是只读的，不能写入），只能链接数据库进行操作，或者将代码执行的间隔放大到一天，否则自己的邮箱会被轰炸

  标签：fee,py,充电,json,别忘记,import,email,邮件
  来源： https://www.cnblogs.com/oziasly/p/16558921.html