J'ai créé un robot LINE qui envoie des images recommandées tous les jours à l'heure

introduction

Je suis fan de Mayu Matsuoka (communément appelée Mayura) J'ai créé un LINE Bot juste pour moi, juste pour moi.

Après environ 5 ans en tant que fan

→ Rechercher des tweets liés à Mayu Matsuoka sur la chronologie Twitter → Enregistrer des images et des vidéos de Mayu Matsuoka

Je n'apprécie plus ce travail autant qu'avant, et je deviens moi-même membre de la société, donc je n'ai pas le temps de surfer sur Twitter tous les jours.

** Automatisons tout! !! ** **

Donc, comme première étape, recherchez des tweets avec des images et des vidéos qui sont récemment populaires sur Twitter, obtenez l'URL de l'image et envoyez-la à l'heure chaque jour en utilisant LINE Bot! !! !! !!

Veuillez ajouter des amis si vous le souhaitez.

L'explication de l'utilisation de l'API Twitter et de l'enregistrement des développeurs LINE est omise ici.

Recherche de tweets, acquisition d'images / vidéos avec tweepy

Page des développeurs Twitter https://developer.twitter.com/en/portal/dashboard

documentation officielle tweepy http://docs.tweepy.org/en/latest/

Définition des variables d'environnement lors de l'utilisation de tweepy

Les variables d'environnement sont décrites dans le fichier Yaml dans le dossier env.

search_tweets.py

import os
import tweepy
from datetime import datetime, date, timedelta
from dateutil.relativedelta import relativedelta

#Ensemble de variables d'environnement
consumer_key = os.getenv('TWITTER_CONSUMER_KEY')
consumer_secret = os.getenv('TWITTER_CONSUMER_SECRET')
access_token = os.getenv('TWITTER_ACCESS_TOKEN')
access_token_secret = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')
bearer_token = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')


#sur Twitter#Fonction pour rechercher l'image de Mayu Matsuoka et obtenir l'URL
def search_tweets():
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth)

env/stg.yml

STAGE: stg
LINE_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
LINE_CHANNEL_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
USER_ID: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_KEY: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_BEAR_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TZ: Asia/Tokyo

Recherche de tweets et acquisition d'images / vidéos

search_tweets.py

    #Obtenez la date d'hier
    yesterday = datetime.strftime(datetime.today() - relativedelta(days=1), f"%Y-%m-%d")
    #Mot de recherche Twitter
    q = f'#Mayu Matsuoka OU Mayu Matsuoka-Filtre de type Matsuoka Mayu:media exclude:retweets min_faves:10 since:{yesterday} min_retweets:2'
    #Chercher
    cric_tweet = tweepy.Cursor(
        api.search, q=q, 
        tweet_mode='extended', #Obtenez tous les tweets omis
        include_entities=True).items(20) #Récupérer tous les liens omis

    #image
    contents = []

    for tweet in cric_tweet:
        print(tweet.full_text)
        try:
            #Un tableau qui stocke les URL des images et des vidéos dans des tweets
            media = tweet.extended_entities['media']
            print(media)
            for m in media:
                print(m)
                #URL d'image de l'écran de prévisualisation requise lors de l'envoi avec LINE Bot
                preview = m['media_url_https']
                #Pour la vidéo
                if m['type'] == 'video':
                    #Lors de l'obtention de l'URL de la vidéo, du contenu_le type est la vidéo/Si ce n'est pas mp4, la vidéo ne sera pas lue lorsqu'elle est envoyée avec LINE Bot.
                    #De plus, je voulais l'écrire en une seule ligne, alors j'ai utilisé de force la notation d'inclusion et finalement[0]J'ai l'URL.
                    origin = [variant['url'] for variant in m['video_info']
                              ['variants'] if variant['content_type'] == 'video/mp4'][0]
                #Pour les images
                else:
                    #URL de l'image de contenu affichée lorsque vous cliquez sur l'image d'aperçu
                    origin = m['media_url_https']

                #Organisez-le sous la forme requise lors de l'envoi avec l'API LINE Bot Messaging.
                content = {'preview': preview,
                           'origin': origin, 'type': m['type']}
                contents.append(content)

            print('--------------------------------------------')
        except:
            print('noUrl')
            print('--------------------------------------------')
    return contents

J'ai écrit dans l'article sur le moment où j'étais accro à ce processus, alors ne le manquez pas ↓

Une histoire à laquelle j'étais accro à essayer d'obtenir une URL de vidéo avec tweepy https://qiita.com/soma_sekimoto/items/65c664f00573284b0b74

Envoyer un message depuis LINE Bot

Référence officielle du bot LINE https://developers.line.me/ja/docs/messaging-api/getting-started/

LINE Bot SDK Github https://github.com/line/line-bot-sdk-python

send_media.py

# -*- coding:utf-8 -*-
import logging
from linebot import LineBotApi

from linebot.exceptions import LineBotApiError
from linebot.models import (
    TextSendMessage, ImageSendMessage, VideoSendMessage
)

import requests
import os


import search_tweets

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def send_media(event, context):

    logger.info("Authentication OK.")
    #Créer un objet API LineBot
    token = os.getenv('LINE_ACCESS_TOKEN')

    line_bot_api = LineBotApi(token)

    try:
        #Obtenez un tableau d'objets URL d'image et de vidéo avec la recherche Twitter
        all_media_list = search_tweets.search_tweets()
        print('all_media_list')
        print(all_media_list)

        messages = []

        #Transformez davantage le tableau obtenu par tweepy pour LINE Bot
        for media in all_media_list:
            item = VideoSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview']) if media['type'] == 'video' else ImageSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview'])

            messages.append(item)

        print('messages')
        print(messages)
        #Assurez-vous de n'envoyer des messages qu'à vous-même, sauf dans un environnement de production.
        if os.getenv('STAGE') == 'prod':
            #Pour une raison quelconque, je ne peux envoyer que 4 images à la fois, donc je les enverrai en deux parties. (Pourtant, c'est toujours un mystère que seulement 7 feuilles peuvent être envoyées au total)
            line_bot_api.broadcast(
                messages[0:3]
            )
            line_bot_api.broadcast(
                messages[4:8]
            )
        else:
            user_id = os.getenv('USER_ID')
            line_bot_api.push_message(
                user_id,
                messages[0:3]
            )
            line_bot_api.push_message(
                user_id,
                messages[4:8]
            )

    except LineBotApiError as e:
        print(e.status_code)
        print(e.error.message)
        print(e.error.details)

    return {"stautsCode": 200, "body": "OK"}


if __name__ == '__main__':
    send_media(None, None)

Déployer avec une infrastructure sans serveur

requirements.py

import os
import sys


requirements = os.path.join(
    os.path.split(__file__)[0],
    '.requirements',
)

if requirements not in sys.path:
    sys.path.append(requirements)

requirements.txt

requests 
print_function
line-bot-sdk #Module pouvant utiliser le SDK LINE Bot en Python
tweepy #Module pouvant utiliser l'API Twitter en Python

serverless.yml

service: mayu-delivery

provider:
  name: aws
  runtime: python3.7
  region: ap-northeast-1
  stage: stg
  deploymentBucket: sls-deps #Spécification d'un bucket de déploiement
  environment: ${file(./env/${opt:stage, self:provider.stage}.yml)}

plugins:
  - serverless-python-requirements

custom:
  scheduleEnabled:
    prod: true
    stg: false
    local: false

functions:
  send_media:
    handler: media_deliver.send_media
    timeout: 300
    events:
      - http:
          path: linebot/send_media
          method: post
      - schedule:
          rate: cron(30 3 * * ? *)
          enabled: ${self:custom.scheduleEnabled.${opt:stage, self:provider.stage}}

sls deploy

endpoints:
  POST - https://xxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/linebot/send_media
functions:
  send_media: mayu-delivery-prod-send_media

N'oubliez pas de copier l'URL des points de terminaison ci-dessus du côté du bot LINE.

Code d'achèvement

search_tweets.py

import os
import tweepy
from datetime import datetime, date, timedelta
from dateutil.relativedelta import relativedelta

consumer_key = os.getenv('TWITTER_CONSUMER_KEY')
consumer_secret = os.getenv('TWITTER_CONSUMER_SECRET')
access_token = os.getenv('TWITTER_ACCESS_TOKEN')
access_token_secret = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')
bearer_token = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')


#sur Twitter#Rechercher l'image de Mayu Matsuoka et obtenir l'URL
def search_tweets():
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth)

    yesterday = datetime.strftime(
        datetime.today() - relativedelta(days=1), f"%Y-%m-%d")

    q = f'#Mayu Matsuoka OU Mayu Matsuoka-Filtre de type Matsuoka Mayu:media exclude:retweets min_faves:10 since:{yesterday} min_retweets:2'

    cric_tweet = tweepy.Cursor(
        api.search, q=q, tweet_mode='extended', include_entities=True).items(20)

    contents = []
    for tweet in cric_tweet:
        print(tweet.full_text)
        try:
            media = tweet.extended_entities['media']
            print(media)
            for m in media:
                print(m)
                preview = m['media_url_https']
                if m['type'] == 'video':
                    origin = [variant['url'] for variant in m['video_info']
                              ['variants'] if variant['content_type'] == 'video/mp4'][0]
                else:
                    origin = m['media_url_https']

                content = {'preview': preview,
                           'origin': origin, 'type': m['type']}
                contents.append(content)

            print('--------------------------------------------')
        except:
            print('noUrl')
            print('--------------------------------------------')
    return contents


if __name__ == "__main__":
    search_tweets()

media_deliver.py

# -*- coding:utf-8 -*-
import logging
from linebot import LineBotApi

from linebot.exceptions import LineBotApiError
from linebot.models import (
    TextSendMessage, ImageSendMessage, VideoSendMessage
)

import requests
import os


import search_tweets

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def send_media(event, context):

    logger.info("Authentication OK.")
    #Créer un objet API LineBot
    token = os.getenv('LINE_ACCESS_TOKEN')

    line_bot_api = LineBotApi(token)

    try:
        #Obtenez un tableau d'objets URL d'image et de vidéo avec la recherche Twitter
        all_media_list = search_tweets.search_tweets()
        print('all_media_list')
        print(all_media_list)

        messages = []

        for media in all_media_list:
            item = VideoSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview']) if media['type'] == 'video' else ImageSendMessage(
                original_content_url=media['origin'], preview_image_url=media['preview'])

            messages.append(item)

        print('messages')
        print(messages)
        if os.getenv('STAGE') == 'prod':
            line_bot_api.broadcast(
                messages[0:3]
            )
            line_bot_api.broadcast(
                messages[4:8]
            )
        else:
            user_id = os.getenv('USER_ID')
            line_bot_api.push_message(
                user_id,
                messages[0:3]
            )
            line_bot_api.push_message(
                user_id,
                messages[4:8]
            )

    except LineBotApiError as e:
        print(e.status_code)
        print(e.error.message)
        print(e.error.details)

    return {"stautsCode": 200, "body": "OK"}


if __name__ == '__main__':
    send_media(None, None)

requirements.py

import os
import sys


requirements = os.path.join(
    os.path.split(__file__)[0],
    '.requirements',
)

if requirements not in sys.path:
    sys.path.append(requirements)

requirements.txt

requests
line-bot-sdk
print_function
tweepy

serverless.yml

service: mayu-delivery

provider:
  name: aws
  runtime: python3.7
  region: ap-northeast-1
  stage: stg
  deploymentBucket: sls-deps
  environment: ${file(./env/${opt:stage, self:provider.stage}.yml)}

plugins:
  - serverless-python-requirements

custom:
  scheduleEnabled:
    prod: true
    stg: false
    local: false

functions:
  send_media:
    handler: media_deliver.send_media
    timeout: 300
    events:
      - http:
          path: linebot/send_media
          method: post
      - schedule:
          rate: cron(30 3 * * ? *)
          enabled: ${self:custom.scheduleEnabled.${opt:stage, self:provider.stage}}

env/stg.yml

STAGE: stg
LINE_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
LINE_CHANNEL_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
USER_ID: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_KEY: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_CONSUMER_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_BEAR_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TWITTER_ACCESS_TOKEN_SECRET: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
TZ: Asia/Tokyo

en conclusion

LINE Bot n'est toujours pas satisfaisant pour moi, mais je vais continuer à m'améliorer et à ajouter des fonctions.

Si vous trouvez quelque chose de mal ou quelque chose de plus comme ça, laissez un commentaire! !!

Et veuillez ajouter des amis! !! !! !!

Article de référence

https://blog.serverworks.co.jp/sls-line-beginner