Go to Qiita Advent Calendar 2024 Top
LoginSignup
12
7

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 3 years have passed since last update.

@G-awain株式会社インテック

Qiita Organization の新着記事をチャットに通知して組織の学習効率を高める！

Posted at

社内で Qiita 記事のアウトプット活動を促進しています。G-awa です。アウトプットによる学習方法は学習効率が良く、大変おすすめです。ただその一方で、ひとりでは続けづらいという面もあるのでチームや組織で互いに支え合うことが重要です。ひとりだと心が折れがちですよね。

→→→→

昨日の記事読んだよ、とても参考になった :thumbsup:
などのように、互いに励まし合える環境があると、エンジニアが成長する環境として素晴らしいです。
当社ではそんな素敵な環境を目指して Qiita に投稿する人が参加するチャンネルを作成して運用しています（実際はパブリックチャンネルになっており、社員全員が閲覧、投稿できます）。

Qiita Organization を作って運用していると、以下のような課題に直面します。

  • 日々の皆のアウトプットを見逃したくない
  • モチベーションを維持したい
  • 活発なコミュニケーションを促進したい

そこで、チャットへ記事の更新情報を通知するツールを作成してみました。

ツールについて

Qiita Organization に投稿された新着記事を取得して、Rocket Chat に通知するといったツールです。Qiita API では Organization の情報が取得できなかったため、クローリングによる実装です。AWS の Lambda で cron 実行されます。
※ Qiita 記事をクローリングすることに対して過度なリクエストは送らない想定ですが、運営側からご指摘があれば本記事およびソースコードは取り下げます。

技術スタック 説明
AWS Lambda 実行環境
Serverless Framework デプロイ
Python beautiful soup クローリング
rocket chat api Rocket Chat にメッセージを投稿
CircleCI テスト

ソースコードはこちら
https://github.com/qiita-scraper/qiita-scraper-rocket-chat

クローリング

beautiful soup を使用してクローリングを行います。投稿日も抽出したいので少し面倒なことをしています。いやー、クローリングはしんどい。

image.png

def fetch_recent_user_articles(self, user):
    """
    指定したユーザの投稿した Qiita 記事のうち、最新の記事を複数取得する
    :param user:
    :return:
    """

    qiita_url = 'https://qiita.com/' + user
    response = request.urlopen(qiita_url)
    soup = BeautifulSoup(response, 'html.parser')
    response.close()

    created_ats = []
    created_dates = soup.find_all('div', class_='ItemLink__info')
    for created_date in created_dates:
        div = re.sub('<a.*?>|</a>', '', str(created_date))
        text = re.sub('<div.*?>|</div>', '', div).split()
        month = str(time.strptime(text[3], '%b').tm_mon)
        day = text[4][:-1]
        year = text[5]
        created_at = year + '/' + month + '/' + day
        created_ats.append(created_at)

    articles = []
    a_tags = soup.find_all('a', class_='u-link-no-underline')
    for index, a in enumerate(a_tags):
        href = a.get('href')
        url = 'https://qiita.com' + href
        title = a.string
        articles.append({'title': title, 'url': url, 'created_at': created_ats[index]})

    return articles

Rocket Chat へ投稿する

RocketChat が API 仕様を公開しています。
https://rocket.chat/docs/developer-guides/rest-api/

シンプルに Python の標準ライブラリである urllib を使用します。Python の HTTP クライアントは urllib.request で十分という記事が参考になりました。本当に urllib だけで十分です。

ログインして authToken と userId を取得します。http-header に記述することで認証され、他の API にアクセスします。

def __login_rocket_chat(self, user, password):
    """
    Rocket Chatにログインして auth_token と user_id を取得します。
    :param url:
    :return:
    """
    obj = {
        "user": user,
        "password": password
    }
    json_data = json.dumps(obj).encode("utf-8")
    headers = {"Content-Type": "application/json"}
    req_object = request.Request(self.url + '/api/v1/login', data=json_data, headers=headers, method='POST')
    with request.urlopen(req_object) as response:
        response_body = response.read().decode("utf-8")
        result_objs = json.loads(response_body.split('\n')[0])
        user_id = result_objs["data"]["userId"]
        auth_token = result_objs["data"]["authToken"]
        print(user_id, auth_token)
    return auth_token, user_id

チャットルームの id を名前から検索します。

def fetch_room_id(self, room_name):
    """
    Rocket Chat の room_id を取得します。
    :param room_name:
    :return:
    """

    headers = {
        "Content-Type": "application/json",
        "X-Auth-Token": self.auth_token,
        "X-User-Id": self.user_id
    }
    params = {'roomName': room_name}
    url = '{}?{}'.format(self.url + '/api/v1/channels.info', parse.urlencode(params))
    req_object = request.Request(url, headers=headers, method="GET")
    with request.urlopen(req_object) as response:
        response_body = response.read().decode("utf-8")
        print(response_body)
        result_objs = json.loads(response_body.split('\n')[0])
        channel = result_objs.get('channel')
        return channel.get('_id')

メッセージを投稿します。RocketChat の Web 画面からだとユーザ名とアイコン画像を差し替えて送信できませんが、API からだと差し替えて送信できます。ちょっと裏技のようでいいですね。

def send_message_to_rocket_chat(self, msg, room_name):
    """
    Rocket Chatにメッセージを送信する
    :param msg:
    :param room_name
    :return:
    """
    headers = {
        "Content-Type": "application/json",
        "X-Auth-Token": self.auth_token,
        "X-User-Id": self.user_id
    }
    print(headers)
    body = {
        "message": {
            "rid": self.fetch_room_id(room_name),
            "msg": msg,
            "alias": 'Qiita Bot',
            "avatar": 'https://haskell.jp/antenna/image/logo/qiita.png'
        }
    }
    print(body)
    req_object = request.Request(self.url + '/api/v1/chat.sendMessage', data=json.dumps(body).encode("utf-8"), headers=headers, method="POST")
    with request.urlopen(req_object) as response:

こんな感じ。

image.png

テスト

docker で RocketChat と mongoDB を起動して、実際の RocketChat アプリケーションに対してリクエストを送ることでテストを実行します。Qiita はごめんなさい、本物にアクセスしてテストします。

RocketChat を docker-compose で立ち上げます。
https://rocket.chat/docs/installation/docker-containers/docker-compose/

OVERWRITE_SETTING_Show_Setup_Wizard=completed を環境変数に指定することで RocketChat 起動時のうざいウィザード画面をスキップできるようです。参考：https://github.com/RocketChat/Rocket.Chat/issues/2233

docker-compose.yml
version: "2"

services:
  rocketchat:
    image: rocketchat/rocket.chat:latest
    command: >
      bash -c
        "for i in `seq 1 30`; do
          node main.js &&
          s=$$? && break || s=$$?;
          echo \"Tried $$i times. Waiting 5 secs...\";
          sleep 5;
        done; (exit $$s)"
    restart: unless-stopped
    volumes:
      - ./uploads:/app/uploads
    environment:
      - PORT=3000
      - ROOT_URL=http://localhost:3000
      - MONGO_URL=mongodb://mongo:27017/rocketchat
      - MONGO_OPLOG_URL=mongodb://mongo:27017/local
      - MAIL_URL=smtp://smtp.email
      - ADMIN_USERNAME=admin
      - ADMIN_PASS=supersecret
      - ADMIN_EMAIL=admin@example.com
      # https://github.com/RocketChat/Rocket.Chat/issues/2233
      - OVERWRITE_SETTING_Show_Setup_Wizard=completed
    depends_on:
      - mongo
    ports:
      - 3000:3000
    labels:
      - "traefik.backend=rocketchat"
      - "traefik.frontend.rule=Host: your.domain.tld"

  mongo:
    image: mongo:4.0
    restart: unless-stopped
    volumes:
      - ./data/db:/data/db
    command: mongod --smallfiles --oplogSize 128 --replSet rs0 --storageEngine=mmapv1
    labels:
      - "traefik.enable=false"

  # this container's job is just run the command to initialize the replica set.
  # it will run the command and remove himself (it will not stay running)
  mongo-init-replica:
    image: mongo:4.0
    command: >
      bash -c
        "for i in `seq 1 30`; do
          mongo mongo/rocketchat --eval \"
            rs.initiate({
              _id: 'rs0',
              members: [ { _id: 0, host: 'localhost:27017' } ]})\" &&
          s=$$? && break || s=$$?;
          echo \"Tried $$i times. Waiting 5 secs...\";
          sleep 5;
        done; (exit $$s)"
    depends_on:
      - mongo

あとは Python の unittest で localhost:3000 に起動した RocketChat を使用してテストを実行します。簡単ですね。

import unittest
from rocket_chat.rocket_chat import RocketChat
import urllib
from qiita.qiita import Qiita
import freezegun
import json

class TestQiitaScraper(unittest.TestCase):
    def setUp(self):
        # rocket chat admin user set in docker-compoose.yml rocketchat service environment value.
        self.aurhorized_user = 'admin'
        self.aurhorized_password = 'supersecret'
        self.rocket_chat_url = 'http://localhost:3000'

    def test_login_success(self):
        rocket_chat = RocketChat(self.rocket_chat_url, self.aurhorized_user, self.aurhorized_password)
        self.assertNotEqual(len(rocket_chat.auth_token), 0)
        self.assertNotEqual(len(rocket_chat.user_id), 0)

    def test_login_failed(self):
        with self.assertRaises(urllib.error.HTTPError):
            unauthorized_user = 'mbvdr678ijhvbjiutrdvbhjutrdfyuijhgf'
            unauthorized_pass = 'gfr67865tghjgfr567uijhfrt67ujhgthhh'
            RocketChat(self.rocket_chat_url, unauthorized_user, unauthorized_pass)

CircleCI でテストを実行する

circleci の実行環境で docker-compose を使用するために executer には imagae ではなくて machine を指定しましょう。※説明の簡略化のためにキャッシュの設定はのぞいています。

.circleci/config.yml
version: 2
jobs:
  build:
    machine:
      image: circleci/classic:201808-01
    steps:
      - checkout
      - run:
          name: "Switch to Python v3.7"
          command: |
            pyenv versions
            pyenv global 3.7.0
      - run:
          name: docker-compose up
          command: sh dcup.sh
      - run:
          name: install dependencies and test
          command: |
            python3 -m venv venv
            . venv/bin/activate
            pip install -r requirements.txt
            pip install -r requirements-dev.txt
            python -m unittest test.py

まとめ

本質的なところ（社内組織のアウトプット活動を促進する）から話がそれてしまいましたが、クローリングや docker-compose を使用して継続的にテストする方法など、自分が勉強になってしまいました。これで組織の学習効率がすこしでも高くなれば幸いです。

12
7
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
12
7

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?