Twitterのいいね欄を取得してリーディングリストを作る

- .env.yaml
- .gcloudignore
- main.py
- requirements.txt

import requests
import pandas as pd
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
import functions_framework
import os

bearer_token = os.environ.get("bearer_token")

def create_url():
  """
  API エンドポイントになるurlを生成
  """
  tweet_fields = "tweet.fields=created_at"
  # user_id
  id = "1104275583171624960"
  # API request
  url = "https://api.twitter.com/2/users/{}/liked_tweets".format(id)
  return url, tweet_fields

def bearer_oauth(r):
  """
  oauthトークンの生成
  """
  r.headers["Authorization"] = f"Bearer {bearer_token}"
  r.headers["User-Agent"] = "v2LikedTweetsPython"
  return r

def connect_to_endpoint(url, tweet_fields):
  """
  エンドポイントに接続
  """
  response = requests.request(
      "GET", url, auth=bearer_oauth, params=tweet_fields)
  print(response.status_code)
  if response.status_code != 200:
      raise Exception(
          "Request returned an error: {} {}".format(
              response.status_code, response.text
          )
      )
  return response.json()

def df_to_body_text(df):
  """
  データフレームから文面を作成
  """
  body_text = ""
  for d in df["text"]:
    body_text = body_text + d + "\n\n\n"
  return body_text

def send_mail(body_text):
  """
  メールを送信
  """
  # SMTPサーバー名とポート番号指定
  smtp = smtplib.SMTP("smtp.gmail.com",587)

  # サーバーと接続を確立
  smtp.ehlo()
  smtp.starttls()

  my_mail =  os.environ.get("mail_from")
  app_password = os.environ.get("mail_pw")
  smtp.login(my_mail,app_password)

  charset = "iso-2022-jp"
  From = my_mail
  To = os.environ.get("mail_to")
  Kenmei = "オラッ！今週のリーディングリストだぞ！"
  Body = f'最新のお気に入りを送るよ！ちゃんと読めよ！！\n\n{body_text}'

  #メール本文を読み込み
  msg = MIMEMultipart()
  msg.attach(MIMEText(Body))
  msg["Subject"] = Header(Kenmei.encode(charset),charset)

  smtp.sendmail(From,To,msg.as_string())
  smtp.quit()

@functions_framework.http
def trigger(request):
  url, tweet_fields = create_url()
  json_response = connect_to_endpoint(url, tweet_fields)
  df_latest =  pd.DataFrame(json_response["data"])
  df_latest_with_article = df_latest[df_latest["text"].str.contains("https://")][0:11]
  send_mail(df_to_body_text(df_latest_with_article))

  return "SUCCESS"

gcloud functions deploy likes-to-newsletter \
--gen2 \
--region=asia-northeast1         \
--runtime=python310 \
--entry-point=trigger \
--trigger-http \
--env-vars-file=.env.yaml \
--allow-unauthenticated

url, tweet_fields = create_url()
json_response = connect_to_endpoint(url, tweet_fields)
# サンプルのコードにensure_ascii=Falseを追加
print(json.dumps(json_response, indent=4, sort_keys=True, ensure_ascii=False))