monitor-twitter-bio-update.py

from datetime import datetime
import json
import os
import twint
import uuid
import firebase_admin
from firebase_admin import firestore
import tempfile
from werkzeug.utils import secure_filename
import requests


def get_file_path(filename):
    file_name = secure_filename(filename)
    return os.path.join(tempfile.gettempdir(), file_name)


def did_update(a, b):
    r = True
    if a["id"] != b["id"]:
        return r
    if a["name"] != b["name"]:
        return r
    if a["username"] != b["username"]:
        return r
    if a["bio"] != b["bio"]:
        return r
    if a["location"] != b["location"]:
        return r
    if a["url"] != b["url"]:
        return r
    if a["private"] != b["private"]:
        return r
    if a["profile_image_url"] != b["profile_image_url"]:
        return r
    if a["background_image"] != b["background_image"]:
        return r
    return False


def main(request):
    tmp_json = get_file_path("o.json")

    try:
        os.remove(tmp_json)
    except OSError:
        pass

    # 現在のbioを取得
    c = twint.Config()
    c.Username = ""
    c.Store_json = True
    c.Output = tmp_json
    twint.run.Lookup(c)

    # 直前と比較
    with open(tmp_json) as f:
        current_bio = json.load(f)

        if len(firebase_admin._apps) == 0:
            default_app = firebase_admin.initialize_app()
        db = firestore.client()

        # 直前のbioを取得
        latest_ref = db.collection("histories").order_by(
            "last_update", direction=firestore.Query.DESCENDING).limit(1).stream()
        for l in latest_ref:
            latest_bio = l.to_dict()

        # 差異があれば通知を投げる
        if did_update(latest_bio, current_bio):
            payload = {
                "id": current_bio["id"],
                "name": current_bio["name"],
                "username": current_bio["username"],
                "bio": current_bio["bio"],
                "location": current_bio["location"],
                "url": current_bio["url"],
                "private": current_bio["private"],
                "profile_image_url": current_bio["profile_image_url"],
                "background_image": current_bio["background_image"],
                "last_update": datetime.now()
            }

            # 履歴を更新
            doc_ref = db.collection("histories").document(str(uuid.uuid4()))
            doc_ref.set(payload)

            # Discordに通知
            webhook_url = ""
            payload.pop("last_update")
            x = {
                "content": json.dumps(payload, ensure_ascii=False, indent=4)
            }
            requests.post(webhook_url, headers={
                          'Content-Type': 'application/json'}, data=json.dumps(x))