[PYTHON] J'ai créé et publié une image Docker qui lit RSS et tweete automatiquement régulièrement.

introduction

J'implémentais des tweets RSS réguliers avec ifttt, mais je l'ai fait moi-même car les restrictions étaient strictes.

Produit fini

https://hub.docker.com/r/hashito/tweetrss

docker run -it --rm \
    -e RSS_URL=https://news.google.com/rss/search?hl=ja&gl=JP&ceid=JP:ja&q=twitter \
    -e CONSUMER_KEY=x \
    -e CONSUMER_SECRET=x \
    -e ACCESS_TOKEN=x \
    -e ACCESS_TOKEN_SECRET=x \
    -e 'ADD_TEXT="#rss2tweet"' \
    hashito/tweetrss

D'en haut

• URL RSS
• Twitter CONSUMER_KEY
• Twitter CONSUMER_SECRET
• Twitter ACCESS_TOKEN
• Twitter ACCESS_TOKEN_SECRET
• Texte à ajouter

Vous pouvez le faire en insérant. Si vous voulez changer le cycle de Tweet, jouez avec SCAN_SPAN et TWEET_DELAY. (Les deux sont des secondes) Soyez prudent lorsque vous vous fâchez si vous tweetez trop ...

De plus, si vous avez une valeur monétaire en interne et que vous la reconstruisez, vous oublierez jusqu'où vous avez lu RSS la dernière fois.

Créons un fichier comme celui ci-dessous et montons-le.

Contenu

{}

-v {you chash file}:/root/cash

(C'est gênant ici ... je vais le réparer un jour)

Code source

La source est simple. Lisez et tweetez régulièrement des flux RSS. RSS conserve l'ID et ne cible pas le même ID. Si le même ID n'est pas reçu pendant un certain temps, l'ID sera ignoré.

import feedparser
import json
import os
import time
import datetime
import json
from requests_oauthlib import OAuth1Session

RSS_URL=os.environ["RSS_URL"]
CONSUMER_KEY=os.environ["CONSUMER_KEY"]
CONSUMER_SECRET=os.environ["CONSUMER_SECRET"]
ACCESS_TOKEN=os.environ["ACCESS_TOKEN"]
ACCESS_TOKEN_SECRET=os.environ["ACCESS_TOKEN_SECRET"]

CASH_FILE=os.environ["CASH_FILE"]
ADD_TEXT=os.environ["ADD_TEXT"]
SCAN_SPAN=int(os.environ["SCAN_SPAN"])
TWEET_DELAY=int(os.environ["TWEET_DELAY"])


def cash_read():
    with open(CASH_FILE) as f:
        return json.loads(f.read())

def cash_write(data):
    with open(CASH_FILE,mode="w") as f:
        f.write(json.dumps(data,ensure_ascii=False))

def read_rss():
    return feedparser.parse(RSS_URL)
    
def send_tweet(text):
    twitter = OAuth1Session(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET) #Processus d'authentification
    url = "https://api.twitter.com/1.1/statuses/update.json" #Point de terminaison du message du Tweet
    params = {"status" : text}
    res = twitter.post(url, params = params) #post envoyer
    return res.status_code == 200

if(__name__ == '__main__'):
    cash=cash_read()
    print("cash:",cash)
    while(1):
        tm=datetime.datetime.now().timestamp()
        rss=read_rss()
        for i in rss["entries"]:
            if(not i["id"] in cash):
                print("tweet:",f"{i['title']} {i['link']} {ADD_TEXT}")
                send_tweet(f"{i['title']} {i['link']} {ADD_TEXT}")
                time.sleep(TWEET_DELAY)
            cash[i["id"]]=tm
            cash_write(cash)
        for k in list(cash.keys()):
            if((cash[k]+60)<tm):
                print("delete:",k,cash[k],tm)
                cash.pop(k)
        time.sleep(SCAN_SPAN)

docker file C'est aussi simple. Installez requirements.txt en utilisant l'image de python3. C'est comme préparer une variable d'environnement et la démarrer.

  
FROM python:3
COPY . /root/

RUN pip install -r /root/requirements.txt

ENV RSS_URL=https://news.google.com/rss/search?hl=ja&gl=JP&ceid=JP:ja&q=twitter \
    CONSUMER_KEY=x \
    CONSUMER_SECRET=x \
    ACCESS_TOKEN=x \
    ACCESS_TOKEN_SECRET=x \
    CASH_FILE=/root/cash \
    ADD_TEXT="#rss2tweet" \
    SCAN_SPAN=600 \
    TWEET_DELAY=150

CMD ["python","/root/main.py"]