zoukankan      html  css  js  c++  java
  • python爬虫加定时任务,制作微信提醒备忘录

    一、任务的记录与提取

    1.1 制作每日任务

      为了便于爬取,推荐使用网页版的在线记事本,现在这种工具很多,我选择“石墨文档”进行操作演示。记录内容的 格式可以根据自己的需求和爬虫自行确定,例如我在11月20日之前记录了一份以下样式的任务清单:
      01月08日 6:30 起床
      01月08日 8:00 上班
      01月08日 11:30 吃饭
      01月08日 4:30 下班
      01月08日 18:00 学习python

    1.2 爬取任务信息

    .  模拟登陆使用爬虫来实现待办事项的提取,发现在线记事本是需要登录的,这里用selenuim进行模拟登陆
    然后从代办事项的文档中获取之前编辑好的代办事项(注意编辑待办事项的格式方便爬取)

    二、信息推送给微信

    2.1 获取server酱SCKEY

      说到推送信息到微信,不得不介绍一个很方便的工具——server酱。说的简单一点,server酱就是一款能向目标微信账号发送信息的工具,她最大的有点就是功能简单、使用方便、轻量,使用server酱只需下面几个步骤:
      登入: 用GitHub账号登入网站,获得自己的SCKEY。
      绑定: 通过微信扫码关注完成绑定。
      发消息: 往 http://sc.ftqq.com/SCKEY.send 发GET请求,就可以在微信里收到消息。
      另外,server酱还提供了网页版的在线发送工具,可以进行体验和测试。值得一提的是,server酱的初始目的是通过手机监测服务器报警和日志,因此为避免程序出错,限制了每人每天发 送上限500条,相同内容5分钟内不能重复发送,不同内容一分钟只能发送30条。对于本系统来说,这个数据上限 绝对绰绰有余了。

    2.2 定时任务与检测

      现在我们需要设置一个定时任务,这里使用APScheduler来实现。APScheduler是基于Quartz的一个python定时任务框架,提供了若干种类的调度器,由于我们的定时任务进在当前进程中执行,所以选择BlockingScheduler调度器,然后通过add_job来动态增加待执行的任务。
    关于APScheduler,其实并不需要了解太多,因为它的基本用法非常简单
    现在还有一个问题,就是要保证我们的程序始终处于正常运行的状态,上面说过,server酱通过向目标URL发送get请求的方式推送信息,而这个get请求是有返回值的,其格式为“{'errno': 0, 'errmsg': 'success', 'dataset': 'done'}”,其中若errno为0则表示发送成功,否则会在errmsg中给出错误信息。基于此,我们可以依托logging模块记录程序每天的运行状态

    具体代码如下:

    import requests
    import logging
    from selenium import  webdriver
    from lxml import etree
    from selenium.webdriver.common.action_chains import ActionChains
    from apscheduler.schedulers.blocking import BlockingScheduler
    
    class Memorandum():
        # 第一步获取石墨文档代办事项的文档
        def get_docs(self):
            user = '******'  #这里是石墨文档的账号
            pwd = '******'  #这里是石墨文档的密码
            browser = webdriver.Chrome()
            browser.implicitly_wait(4)  #隐试延迟4秒,是全局作用只需写一次
            browser.get("https://shimo.im/login")
    
            #  清空输入框的内容然后输入自己的账号密码
            browser.find_element_by_xpath('//*[@id="root"]/div/div[1]/div[1]/div/div/div[2]/div[1]/div[1]/div/input').clear()
            browser.find_element_by_xpath('//*[@id="root"]/div/div[1]/div[1]/div/div/div[2]/div[1]/div[1]/div/input').send_keys(user)
            browser.find_element_by_xpath('//*[@id="root"]/div/div[1]/div[1]/div/div/div[2]/div[1]/div[2]/div/input').clear()
            browser.find_element_by_xpath('//*[@id="root"]/div/div[1]/div[1]/div/div/div[2]/div[1]/div[2]/div/input').send_keys(pwd)
            browser.find_element_by_xpath('//*[@id="root"]/div/div[1]/div[1]/div/div/div[2]/div[1]/button').click()
            # 点击我的桌面
            a = browser.find_element_by_xpath('//*[@id="root"]/div/div[2]/div[1]/div[1]/a[2]')
            ActionChains(browser).double_click(a).perform()
            # 点击待办事项
            b = browser.find_element_by_xpath('/html/body/div[1]/div/div[2]/div[2]/div/div[1]/div[2]/div[2]/div[1]/div/div/div/div[2]/div/div/a/div[1]')
            ActionChains(browser).double_click(b).perform()
            html = etree.HTML(browser.page_source)
            content = html.xpath('//div[@class="ql-editor"]/p/span/text()')
            browser.close()
            return content
    
        def judge_date(self,content):
            # 获取当天的时间 格式为  (02月08日)
            today = datetime.date.today().strftime("%m-%d").replace("-","")+""
            print(today)
            today_task = []
            for i in content:
                # print(i.split(" ")[0])
                if i.split()[0]== today:
                    today_task.append(''.join(i.split()[1:]))
            print(today_task)
            return today,'   
    '.join(today_task)
    
        def send_mssg(self,today,messg):
            api = "https://sc.ftqq.com/SCU38773T7f2756aae48b772e8366c458fa535ff15c301c5bf3189.send"
            title = today +"代办事项"
            content = messg
            data = {
                "text": title,
                "desp": content
            }
            req = requests.post(api, data=data)
    
            logging.basicConfig(filename='log.log', format='%(asctime)s:%(message)s', level=logging.INFO)
            if req.json()['errno'] == 0:
                logging.info('发送成功')
            else:
                logging.warning('发送失败 %s' % req.json()['errmsg'])
        #  定时任务
        def timing_task(self):
            content = self.get_docs()
            today , messg = self.judge_date(content)
            scheduler = BlockingScheduler()
            scheduler.add_job(func=self.send_mssg,args=(today,messg,),trigger='cron', day_of_week='0-6', hour=16, minute=25, end_date='2019-10-02')
            scheduler.start()
    
    
    if __name__=="__main__":
        m = Memorandum()
        m.timing_task()
        # c = m.get_docs()
        # print(c)
        # m.judge_date(c)

     下面是效果图:

  • 相关阅读:
    提升ASP.NET性能
    人性的弱点
    墨菲定律
    沟通
    网站
    程序员思维模式
    CSS
    HTML
    路由和数据传递(04)
    Sql Server中查看所有数据库,表名,字段名以及字段类型
  • 原文地址:https://www.cnblogs.com/wangchengpei/p/10951835.html
Copyright © 2011-2022 走看看