[PYTHON] Ich habe ein Docker-Image erstellt, das RSS liest und automatisch regelmäßig twittert, und es veröffentlicht.

Einführung

Ich habe reguläre RSS-Tweets mit ifttt implementiert, aber ich habe es selbst gemacht, weil die Einschränkungen streng waren.

Fertiges Produkt

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

Von oben

• RSS-URL
• Twitter CONSUMER_KEY
• Twitter CONSUMER_SECRET
• Twitter ACCESS_TOKEN
• Twitter ACCESS_TOKEN_SECRET
• Hinzufügender Text

Sie können dies durch Einfügen tun. Wenn Sie den Tweet-Zyklus ändern möchten, spielen Sie mit SCAN_SPAN und TWEET_DELAY. (Beide sind Sekunden) Sei vorsichtig, wenn du wütend wirst, wenn du zu viel twitterst ...

Wenn Sie intern einen Barwert haben und diesen neu erstellen, werden Sie vergessen, wie weit Sie das letzte Mal RSS gelesen haben.

Lassen Sie uns eine Datei wie die folgende erstellen und sie mounten.

Inhalt

{}

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

(Es ist unangenehm hier ... ich werde es eines Tages reparieren)

Quellcode

Die Quelle ist einfach. Lesen und twittern Sie regelmäßig RSS. RSS behält die ID bei und zielt nicht auf dieselbe ID ab. Wenn dieselbe ID für einen bestimmten Zeitraum nicht empfangen wird, wird die ID verworfen.

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) #Authentifizierungsprozess
    url = "https://api.twitter.com/1.1/statuses/update.json" #Tweet Post Endpunkt
    params = {"status" : text}
    res = twitter.post(url, params = params) #post senden
    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 Das ist auch einfach. Installieren Sie require.txt mit dem Image von python3. Es ist, als würde man eine Umgebungsvariable vorbereiten und starten.

  
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"]