はじめに

社内用のもくもく会サーバーを作ったのでQiitaのOrganizationの記事をポストするBOTを作りました

使うもの

Python
Qiita API
discord.py

参考にした記事

Pythonで実用Discord Bot(discordpy解説)

前準備

discord BOTアカウントの作成

Qiitaのアクセストークンはなくてもいいです

方針

Organizationのユーザー取得してくる
1で取得したユーザーの記事を取得してくる
discordにポストする

実装


import requests
from bs4 import BeautifulSoup
import json
import datetime
from dataclasses import dataclass
import discord
import asyncio
import platform

URL = "https://qiita.com/organizations/{organization名}/members?page="
API_URL = "https://qiita.com/api/v2/items?page="
QIITA_ACCESS_TOKEN = "{Qiitaのアクセストークン}"
DISCORD_ACCESS_TOKEN = "{discordのアクセストークン}"
CHANNEL_ID = {ポストしたいチャンネルのID}


def main():
    today = datetime.date.today()
    yesterday = today - datetime.timedelta(days=1)
    today_str = today.strftime('%Y-%m-%d')
    yesterday_str = yesterday.strftime('%Y-%m-%d')
    query_dict = get_query(today_str, yesterday_str)
    if len(query_dict) == 0:
        return
    topic_list = get_topic(query_dict)
    if len(topic_list) == 0:
        return
    post_discord(topic_list)


def get_query(today_str, yesterday_str):
    count = 0
    user_dict = {}
    while True:
        count += 1
        response = requests.get(URL+str(count))
        soup = BeautifulSoup(response.text, 'html.parser')
        get_user_list = soup.find_all(
            'span', class_='od-MemberCardHeaderIdentities_userid')
        if get_user_list == None or len(get_user_list) == 0:
            return user_dict
        for item in get_user_list:
            user_id = item.get_text()[1:]
            user_dict[user_id] = QueryBean(
                user_id, yesterday_str, today_str)


def get_topic(query_dict):
    query = "query="
    count = 0
    list = []
    for bean in query_dict.values():
        count += 1
        query += "created:>="+bean.start_date+"+created:<" + \
            bean.end_date+"+user:"+bean.user+" OR "
        if count >= 5:
            list += request(query)
            count = 0
            query = "query="
    list += request(query)
    return list


def request(query):
    count = 0
    list = []
    while True:
        count += 1
        url = (
            API_URL + str(count)+"&per_page=100&" + query
        )
        headers = {"Authorization": 'Bearer {}'.format(QIITA_ACCESS_TOKEN)}
        response = requests.get(url, headers)

        if response.status_code > 200:
            print("うまくいかなかった")
            print(response)
            return list
        text = json.loads(response.text)

        if len(text) == 0:
            print("Nothing")
            return list

        for v2 in text:
            list.append(PostContents(v2["title"], v2["url"]))


def post_discord(topic_list):
    if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

    client = discord.Client()

    @client.event
    async def on_ready():
        print('ログインしました')
        await greet()
        await client.close()

    async def greet():
        channel = client.get_channel(CHANNEL_ID)
        for topic in topic_list:
            await channel.send(topic.title+"\n"+topic.url)

    client.run(DISCORD_ACCESS_TOKEN)


@dataclass
class QueryBean:
    user: str = ""
    start_date: str = ""
    end_date: str = ""


@dataclass
class PostContents():
    title: str = ""
    url: str = ""


if __name__ == "__main__":
    main()

コードの解説

get_query()

organizationメンバーを取得して、記事の検索クエリに必要な情報を返却します。
Qiitaにはorganization系のAPIが用意されてないので、スクレイピングをして取得します。
記事の検索はこのAPIでできるのでこれに必要な情報をClassに入れて返却してます。

get_topic()

get_queryで取得した情報をもとにクエリを作成し、request()に渡します。

request()

QiitaApiにリクエストを投げ、返却された値から必要な情報を取得し、Classに格納して返却します。

post_discord()

Discordにポストします

if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

これは、RuntimeError: Event loop is closedが発生するのを回避するために入れてます。

ここらへんを読むと、開発環境がWindowsだから発生してるみたいでした。
FedoraとUbuntuではエラーは発生しませんでした。

まとめ

みんなもくもく会サーバーに入ってほしいです