We are promoting Qiita article output activities within the company. This is G-awa. The learning method using output is highly recommended because it has good learning efficiency. However, on the other hand, it is difficult to continue alone, so it is important for teams and organizations to support each other. If you are alone, your heart tends to break.

→→→→

I read yesterday's article, it was very helpful: thumbsup: Having an environment where you can encourage each other, such as, is a great environment for engineers to grow. We aim to create such a wonderful environment by creating and operating a channel in which people who post to Qiita participate (actually, it is a public channel, and all employees can view and post it).

When creating and operating a Qiita Organization, you will face the following challenges.

――I don't want to miss the daily output of everyone ――I want to stay motivated ――I want to promote active communication

So, I created a tool to notify the chat of the update information of the article.

About tools

It is a tool that retrieves new articles posted to the Qiita Organization and notifies Rocket Chat. Since the Qiita API could not get the information of Organization, it is implemented by crawling. It runs cron on AWS Lambda.

It is assumed that we will not send excessive requests for crawling Qiita articles, but if the management side points out, we will withdraw this article and the source code.

Tech stack	Description
AWS Lambda	Execution environment
Serverless Framework	Deploy
Python beautiful soup	Crawling
rocket chat api	Post a message to Rocket Chat
CircleCI	test

Click here for source code https://github.com/qiita-scraper/qiita-scraper-rocket-chat

Crawling

Use beautiful soup for crawling. I also want to extract the posting date, so I am doing a little troublesome thing. No, crawling is tough.

def fetch_recent_user_articles(self, user):
    """
Get multiple latest Qiita articles posted by the specified user
    :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

Post to Rocket Chat

RocketChat publishes an API specification. https://rocket.chat/docs/developer-guides/rest-api/

Simply use urllib, Python's standard library. The article urllib.request is sufficient for Python HTTP client was helpful. Really urllib is enough.

def __login_rocket_chat(self, user, password):
    """
Log in to Rocket Chat and auth_token and user_Get the 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

Search for the chat room id by name.

def fetch_room_id(self, room_name):
    """
Rocket Chat room_Get the 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')

Post a message. You cannot send by replacing the user name and icon image from the RocketChat web screen, but you can send by replacing it from the API. It's a little tricky, isn't it?

def send_message_to_rocket_chat(self, msg, room_name):
    """
Send a message to 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:

Like this.

test

Run the test by launching RocketChat and mongoDB with docker and sending a request to the actual RocketChat application. I'm sorry Qiita, access the real thing and test it.

Launch RocketChat with docker-compose. https://rocket.chat/docs/installation/docker-containers/docker-compose/

It seems that you can skip the annoying wizard screen when starting RocketChat by specifying OVERWRITE_SETTING_Show_Setup_Wizard = completed as an environment variable. Reference: 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
      - [email protected]
      # 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

Then run the test using RocketChat launched on localhost: 3000 with a Python unittest. It's easy.

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)

Run tests on CircleCI

To use docker-compose in the circleci runtime environment, specify machine for ʻexecuter instead of ʻimagae. * The cache settings are excluded for the sake of simplicity.

`.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

Summary

I'm off the hook from the essence (promoting the output activity of the internal organization), but I've learned how to continuously test using crawling and docker-compose. It was. I hope this will improve the learning efficiency of the organization even a little.

Notify the chat of new articles of Qiita Organization and improve the learning efficiency of the organization!