Slack + Python + CircleCI 做一个免费的自动脚本工具

如果你听过 RSS 或 IFTTT,那你一定很熟悉我要做的东西用途大致是什么了。

Demo 地址:https://github.com/DianQK/my_slack_bot_demo

标题三个关键词对应功能如下:

  • Slack:可以理解为类似钉钉、企业微信即时通信工具(颜值最高的应该就是 Slack 了,同时接入 Bot 也比较方便)
  • Python:一门编程语言,你可以替换成喜欢的语言实现后面的文章内容抓取等功能(Python 环境配置请选择 https://conda.io/
  • CircleCI:类似 Jenkins,我们用来执行定时任务(把集成部署的脚本换成我们的抓取脚本)

完整的流程:CircleCI 定时执行脚本,将获取的内容推送到 Slack。

效果如下:

完成的功能如下:

  • 少数派新文章定时推送
  • TESTV 新视频定时推送(抓取 Bilibili,通过 URL Scheme 直接打开 App)
  • 我最爱的歌手有新的演唱会啦
  • 自动签到

一起来做个推送 Bot 吧。

创建配置 Slack Bot

Slack 地址: https://slack.com/ 注册完毕后就可以开始创建我们的个人 Bot 啦。

地址如下:https://api.slack.com/apps

在 features and functionality 中建议使用 Bots 功能,使用 Incoming Webhooks 会遇到发送多个 channel 无法自定义头像名称问题。

一起先来试试 Bot 的推送功能:

引入依赖 slackclient

from slackclient import SlackClient

slack_token = "xoxb-xxx-xxx-xxx"
slack = SlackClient(slack_token)

slack_token 获取地址在 Installed App Settings -> Bot User OAuth Access Token。

测试一下 slack_token 是否配置正确:

slack 推送格式文档见:https://api.slack.com/messaging/composing

slack.api_call(
  "chat.postMessage",
  channel="#general",
  username="my_bot",
  icon_url="https://blog.dianqk.org/favicon.png",
  text='Hello World!'
)

这一步 Repo 参见:https://github.com/DianQK/my_slack_bot_demo/tree/a627a39594e53082fce6a2fcbd3e5981110638e0

完成抓取逻辑

哔哩哔哩有可以直接调用的 JSON 接口,代码如下:

import requests
from slack import slack

def fetchLatestTestVideos():
  response = requests.get(
    url="https://space.bilibili.com/ajax/member/getSubmitVideos",
    params={
      "mid": "11336264",
      "pagesize": "1",
      "tid": "0",
      "page": "1",
      "keyword": "值不值得买",
      "order": "pubdate",
    }
  )

  videos = response.json()["data"]["vlist"]
  videos = [
    (
    video['aid'], video["title"], video["description"], f"https:{video['pic']}")
    for video in videos
  ]

  for (aid, title, summary, coverUrl) in videos:
    appUrl = f"bilibili://video/{aid}"
    webUrl = f"https://www.bilibili.com/video/av{aid}"
    slack.api_call(
      "chat.postMessage",
      channel="#review",
      username="TESTV",
      icon_url="https://tva1.sinaimg.cn/crop.0.0.640.640.180/005QGjbqjw8f884tmcirlj30hs0hs0sw.jpg",
      attachments=[
        {
          "fallback": title,
          "title": title,
          "title_link": webUrl,
          "image_url": coverUrl,
          "text": summary,
          "actions": [
            {
              "type": "button",
              "text": "本期视频",
              "url": appUrl
            },
            {
              "type": "button",
              "text": "TESTV 主页",
              "url": "bilibili://space/11336264"
            }
          ]
        }
      ]
    )

def main():
  fetchLatestTestVideos()

if __name__ == '__main__':
  main()

执行 python -m bilibili.testv 来体验一下推送最新一条视频的效果:

这一步 Repo 参见:https://github.com/DianQK/my_slack_bot_demo/tree/936ef6504ad10017f4d63f27676e4968c1f0b83e

CircleCI 配置定时推送

CircleCI 文档见:https://circleci.com/docs/2.0/language-python/

在 CircleCI 中有 job 和 workflow 两个概念:

  • job 指执行的工作内容
  • workflow 指工作流程,具体描述执行哪些 job 以及触发 job 逻辑
executors:
  my-executor:
    docker:
      - image: circleci/python:3.6.1
    working_directory: ~/repo

jobs:
  my-job:
    environment:
      RUN_TASKS: testv,sspai
    executor: my-executor
    steps:
      - checkout
      - restore_cache:
          keys:
          - v1-dependencies-{{ checksum "requirements.txt" }}
          - v1-dependencies-
      - run:
          name: install dependencies
          command: |
            python3 -m venv venv
            . venv/bin/activate
            pip install -r requirements.txt
      - save_cache:
          paths:
            - ./venv
          key: v1-dependencies-{{ checksum "requirements.txt" }}
      - run:
          name: run all
          command: |
            . venv/bin/activate
            python main.py # 执行我们的抓取任务

workflows:
  version: 2
  morning-scheduled-workflow:
    triggers:
      - schedule:
          cron: "0 0 * * *" # 我们是 +8 时区,当前配置为每天早上 8 点
          filters:
            branches:
              only: master
    jobs:
      - my-job

根据 RUN_TASKS 环境变量定制需要执行哪些任务:

import os
from bilibili.testv import fetchLatestTestVideos
import sys

RUN_TASKS = os.environ["RUN_TASKS"]

tasks = RUN_TASKS.split(',')

if 'testv' in tasks:
  fetchLatestTestVideos()

if 'sspai' in tasks:
  print('TODO')

Next

基本的功能都完成了,剩下的我们还需要考虑:

  • 脚本异常处理,挂了推送到 Slack
  • 文章内容重复推送,CircleCI 似乎提供了一个可以读写的 MySQL,我使用的阿里的 OSS,除去做一些基本的记录还可以存储抓取的图片的,Web 归档
  • 主动抓取文章交互,Hey Siri,来篇文章。我好像不明白你在说什么。
上次更新: 1 个月前