[PYTHON] Notifiez les nouveaux articles de l'organisation Qiita pour discuter et améliorer l'efficacité d'apprentissage de l'organisation!

Nous faisons la promotion des activités de production d'articles Qiita au sein de l'entreprise. C'est G-awa. La méthode d'apprentissage de sortie a une bonne efficacité d'apprentissage et est fortement recommandée. Cependant, d'un autre côté, il est difficile de continuer seul, il est donc important que les équipes et les organisations se soutiennent mutuellement. Il est facile de vous briser le cœur lorsque vous êtes seul.

→→→→

J'ai lu l'article d'hier, c'était très utile: thumbsup: Avoir un environnement dans lequel vous pouvez vous encourager mutuellement, par exemple, est un excellent environnement pour que les ingénieurs se développent. Dans le but de créer un environnement aussi merveilleux, nous créons et exploitons un canal auquel participent les personnes qui publient sur Qiita (en fait, il s'agit d'un canal public et tous les employés peuvent voir et publier).

Lors de la création et de l'exploitation de l'organisation Qiita, nous sommes confrontés aux défis suivants.

――Je ne veux pas manquer la sortie quotidienne de tout le monde —— Je veux rester motivé ――Je souhaite promouvoir une communication active

J'ai donc créé un outil pour informer le chat des informations de mise à jour de l'article.

À propos des outils

C'est un outil qui récupère les nouveaux articles publiés sur l'organisation Qiita et informe Rocket Chat. Étant donné que l'API Qiita n'a pas pu obtenir les informations de l'organisation, elle est mise en œuvre par exploration. Il exécute cron sur Lambda sur AWS.

Pile technologique La description
AWS Lambda Environnement d'exécution
Serverless Framework Déployer
Python beautiful soup Rampant
rocket chat api Publier un message sur Rocket Chat
CircleCI tester

Cliquez ici pour le code source https://github.com/qiita-scraper/qiita-scraper-rocket-chat

Rampant

Utilisez beautiful soup pour l'exploration. Je veux aussi extraire la date de publication, donc je fais une petite chose gênante. Non, ramper est difficile.

image.png

def fetch_recent_user_articles(self, user):
    """
Obtenez plusieurs derniers articles Qiita publiés par l'utilisateur spécifié
    :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

Publier sur Rocket Chat

RocketChat publie les spécifications de l'API. https://rocket.chat/docs/developer-guides/rest-api/

Utilisez simplement urllib, la bibliothèque Python standard. L'article urllib.request est suffisant pour le client HTTP Python a été utile. Vraiment urllib suffit.

Connectez-vous pour obtenir authToken et userId. Il est authentifié en écrivant dans l'en-tête http et accède à d'autres API.

def __login_rocket_chat(self, user, password):
    """
Connectez-vous à Rocket Chat et authentifiez-vous_jeton et utilisateur_Obtenez l'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

Recherchez l'ID de la salle de discussion par nom.

def fetch_room_id(self, room_name):
    """
Salle de chat Rocket_Obtenez l'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')

Envoyer un message. Vous ne pouvez pas envoyer en remplaçant le nom d'utilisateur et l'image de l'icône de l'écran Web RocketChat, mais vous pouvez envoyer en le remplaçant à partir de l'API. C'est un peu délicat, n'est-ce pas?

def send_message_to_rocket_chat(self, msg, room_name):
    """
Envoyer un message à 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:

Comme ça.

image.png

tester

Exécutez le test en lançant RocketChat et mongoDB dans docker et en envoyant une requête à l'application RocketChat réelle. Je suis désolé Qiita, j'accède et teste la vraie chose.

Lancez RocketChat avec docker-compose. https://rocket.chat/docs/installation/docker-containers/docker-compose/

Il semble que vous puissiez ignorer l'écran ennuyeux de l'assistant lors du démarrage de RocketChat en spécifiant OVERWRITE_SETTING_Show_Setup_Wizard = complété comme variable d'environnement. Référence: 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

Ensuite, exécutez le test en utilisant RocketChat lancé sur localhost: 3000 avec unittest de Python. C'est facile.

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)

Exécuter des tests sur CircleCI

Pour ʻexecuter, spécifiez machine au lieu de ʻimagae pour utiliser docker-compose dans l'environnement d'exécution circleci. * Les paramètres de cache sont exclus pour des raisons de simplicité.

.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

Résumé

Je suis décroché de l'essentiel (promotion des activités de sortie de l'organisation interne), mais j'ai appris à tester en permanence à l'aide de l'exploration et de docker-compose. C'était. J'espère que cela améliorera encore un peu l'efficacité d'apprentissage de l'organisation.

Recommended Posts

Notifiez les nouveaux articles de l'organisation Qiita pour discuter et améliorer l'efficacité d'apprentissage de l'organisation!
Obtenez des visites d'articles et des likes avec l'API Qiita + Python
J'ai considéré la méthode d'apprentissage automatique et son langage d'implémentation à partir des informations de balise de Qiita
J'ai affiché le chat de YouTube Live et essayé de jouer
[API Qiita] [Statistiques • Apprentissage automatique] J'ai essayé de résumer et d'analyser les articles publiés jusqu'à présent.
Obtenez le nombre de PV d'articles Qiita que vous avez publiés avec l'API
Notifier le contenu de la tâche avant et après l'exécution de la tâche avec Fabric
Pour améliorer la réutilisabilité et la maintenabilité des flux de travail créés avec Luigi
J'ai essayé d'améliorer l'efficacité du travail quotidien avec Python