<meta property=

完全自動情報収集X投稿AIツールの概要

日本の企業のさまざまな最新情報が集まるメディアといったら「PR TIMES」はその一つとして挙げられると思います。
今回のツールでは、PR TIMESから生成AIに関連するニュースを探し出し、わかりやすく一言で概要を説明し、Xに投稿するまでをすべて自動で30分ごとに5件ずつ遂行するAIツールの開発方法について公開します。
AWS Lambda、DynamoDB、OpenAI API、Twitter APIで連携させて動かします。

機能概要

機能としてはシンプルで、以下のようになります。

PR TIMESから生成AI関連の最新記事を取得
GPTを使用して記事の要約とコメントを生成
Xに投稿
DynamoDBを使用して処理済み記事を管理

システム構成

インフラストラクチャ

AWS
- Lambda
  - コンテナイメージベースの実行環境
  - Python 3.9ランタイム
  - タイムアウト: 900秒（15分）
- ECR (Elastic Container Registry)
  - Lambdaコンテナイメージの保存
  - pr_times_botイメージを管理
- DynamoDB
  - テーブル名: LastProcessedItems
  - 処理済み記事の管理
  - PAY_PER_REQUESTモード
- EventBridge
  - 定期実行スケジューラー
  - cron式: cron(0/30 * * * ? *)（30分ごとに実行）

アプリケーション機能

外部API/サービス連携
- OpenAI API
  - GPT-4oモデルを使用
  - 記事要約・コメント生成
- Twitter API
  - tweepyライブラリ使用
  - 記事の自動投稿
- PR TIMES
  - Webスクレイピング
  - BeautifulSoup4を使用

デプロイメント

Serverless Framework
- AWSリソースの管理
- インフラのコード化（IaC）
- コンテナイメージのデプロイ

主要なPythonライブラリ

- requests        # HTTP通信
- beautifulsoup4  # Webスクレイピング
- tweepy         # Twitter API
- openai         # OpenAI API
- boto3          # AWS SDK

監視/ロギング

CloudWatch
- Lambdaのログ管理
- 実行スケジュール管理

環境変数

- OPENAI_API_KEY
- TWITTER_ACCESS_TOKEN
- TWITTER_ACCESS_TOKEN_SECRET
- TWITTER_API_KEY
- TWITTER_API_SECRET
- TWITTER_BEARER_TOKEN

構成図

デプロイするための前提条件

開発環境の構築は説明しません
Dockerがローカルにインストールされていること
- コンテナベースのLambda関数を使用しているため
AWS認証情報の設定
- AWS Access Key IDとSecret Access Keyの設定
- ~/.aws/credentialsの設定または環境変数での設定

AWS_ACCESS_KEY_ID=xxx
AWS_SECRET_ACCESS_KEY=xxx
AWS_DEFAULT_REGION=ap-northeast-1 # 使用するリージョン

Serverless Framework
- Node.jsがローカルにインストールされていること
- Serverless Frameworkがローカルにインストールされていること

npm install -g serverless

手順

TwitterAPIの準備

https://developer.twitter.com/en/portal/petition/essential/basic-info にアクセスする。
【Sign up for Free Account】をクリックする。
ここでXにログインできてない場合はログインを促されるので、自動投稿をしたいアカウントにログインする。

250字以上でAPIの申請理由を記載する。(GPTなどに書いてもらえれば良い)
チェックボックスを入れ【Submit】をクリックする。

以下のような画面に遷移する。
【Projects & Apps】内の【Default project-xxxxxx】内のxxxxxをクリックする。

上タブの【Keys and tokens】をクリックする。

この画面で以下を取得し、メモに保存しておく。

API Key
API Key Secret
Bearer Token
Access Token
Access Token Secret

必要なファイルを作成

ディレクトリ構成

.
├── Dockerfile
├── README.md
├── app
│   └── lambda_function.py
├── docker-compose.yaml
├── requirements.txt
└── serverless.yml

プロジェクトのルートディレクトリ配下に以下を作成する。

Dockerfile
serverless.yml
requirements.txt
app
- lambda_function.py

touch Dockerfile serverless.yml

requirements.txt mkdir app

touch app/lambda_function.py

Dockerfile
環境変数にメモしておいたtokenやkeyを記載する。

Dockerfile

FROM --platform=linux/amd64 public.ecr.aws/lambda/python:3.9

# 必要なパッケージのインストール
COPY requirements.txt ./
RUN pip install -r requirements.txt

# Lambdaハンドラーコードのコピー
COPY app/lambda_function.py ${LAMBDA_TASK_ROOT}

# 環境変数を設定
ENV OPENAI_API_KEY=
ENV EXEC_TIME=:30
ENV TWITTER_ACCESS_TOKEN=
ENV TWITTER_ACCESS_TOKEN_SECRET=
ENV TWITTER_API_KEY=
ENV TWITTER_API_SECRET=
ENV TWITTER_BEARER_TOKEN=

# Lambdaハンドラーの設定
CMD ["lambda_function.lambda_handler"]

serverless.yml
スケジュールを変更したい場合は、このファイルのschedule: 'cron(0/30 * * * ? *)'の部分を変更する。

YAML

service: pr-times-bot

frameworkVersion: '3'

provider:
  name: aws
  runtime: python3.9
  region: ap-northeast-1
  environment:
    ENV: production
  ecr:
    images:
      pr_times_bot:
        path: .
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - dynamodb:DescribeTable
            - dynamodb:CreateTable
            - dynamodb:PutItem
            - dynamodb:GetItem
            - dynamodb:UpdateItem
            - dynamodb:DeleteItem
            - dynamodb:Scan
            - dynamodb:Query
          Resource:
            - Fn::GetAtt: [LastProcessedItemsTable, Arn]

functions:
  app:
    image:
      name: pr_times_bot
      command: ["lambda_function.lambda_handler"]
    timeout: 900
    events:
      - eventBridge:
          name: 'appSchedule'
          description: 'Schedule for app function'
          schedule: 'cron(0/30 * * * ? *)'

resources:
  Resources:
    AppLambdaFunctionUrl:
      Type: AWS::Lambda::Url
      Properties:
        AuthType: NONE
        TargetFunctionArn:
          Fn::GetAtt:
            - AppLambdaFunction
            - Arn
    LastProcessedItemsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: LastProcessedItems
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        BillingMode: PAY_PER_REQUEST

requirements.txt

requests==2.32.3
beautifulsoup4==4.12.3
pandas==2.2.2
python-dateutil==2.9.0.post0
pytz==2024.1
openai==1.31.0
tweepy==4.14.0
boto3==1.34.34

lambda_function.py

Python

import json
import os
import requests
from bs4 import BeautifulSoup
from dateutil import parser
import pytz
from io import BytesIO
import re
import tweepy
from openai import OpenAI
import boto3
from botocore.exceptions import ClientError

# DynamoDBクライアントの初期化
dynamodb = boto3.resource('dynamodb')
table_name = 'LastProcessedItems'

# Twitterの認証
twitter_client = tweepy.Client(
    bearer_token=os.environ["TWITTER_BEARER_TOKEN"],
    consumer_key=os.environ["TWITTER_API_KEY"],
    consumer_secret=os.environ["TWITTER_API_SECRET"],
    access_token=os.environ["TWITTER_ACCESS_TOKEN"],
    access_token_secret=os.environ["TWITTER_ACCESS_TOKEN_SECRET"]
)

def ensure_dynamodb_table():
    table = dynamodb.Table(table_name)
    try:
        table.load()
        print(f"Table {table_name} exists.")
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceNotFoundException':
            print(f"Table {table_name} does not exist. It should be created by Serverless Framework.")
            raise
        else:
            print(f"An error occurred: {e}")
            raise
    return table

def _load_latest_content(table, key):
    try:
        response = table.get_item(Key={'id': key})
        return response.get('Item', {}).get('content')
    except ClientError as e:
        print(f"Error reading from DynamoDB: {e}")
    return None

def _save_latest_content(table, key, content):
    try:
        table.put_item(Item={'id': key, 'content': content})
    except ClientError as e:
        print(f"Error writing to DynamoDB: {e}")

def load_keywords(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        data = json.load(file)
    return data['process_words']

def __get_press_release(table):
    url = '<https://prtimes.jp/topics/keywords/%E7%94%9F%E6%88%90AI>'
    latest_title = _load_latest_content(table, 'prtimes_last_title')
    response = requests.get(url)
    html_content = response.content
    soup = BeautifulSoup(html_content, 'html.parser')
    press_releases = []
    articles = soup.find_all('article', class_='item-ordinary')
    new_latest_title = None
    for article in articles[:5]:
        title_element = article.find('a', class_='link-title-item-ordinary')
        title = title_element.get_text(strip=True) if title_element else 'タイトルなし'
        if title == latest_title:
            break
        date_element = article.find('time')
        date_str = date_element.get('datetime') if date_element else None
        company_element = article.find('a', class_='thumbnail-name-company')
        company = company_element.get_text(strip=True) if company_element else '会社名なし'
        link_element = article.find('a', class_='link-title-item-ordinary')
        link = link_element['href'] if link_element else 'リンクなし'
        if date_str:
            date = parser.isoparse(date_str)
            date = date.replace(tzinfo=None)
            press_release = {
                'title': title,
                'date': date,
                'company': company,
                'link': link
            }
            press_releases.append(press_release)
        if new_latest_title is None:
            new_latest_title = title
    if new_latest_title:
        _save_latest_content(table, 'prtimes_last_title', new_latest_title)
    return press_releases

def post_to_x(url, comment, ogp):
    title = ogp['title']
    if title is None:
        return
    content = f"""{comment}
    {url}
    """
    twitter_client.create_tweet(text=content)

def create_comments(combined_text):
    MODEL = "gpt-4o"
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "hoge"))
    system_prompt = """
    入力されたニュース記事の内容から一言コメントをお願いします。
    コメントは語尾に「コレ」とつくような口調で。
    コメントは経済評論家の視点で確信をつくようなコメントをお願いします。
    2文が好ましい。

    <news>
    ###NEWS
    </news>
    """
    NEWS_REP = "###NEWS"
    messages = [
        {"role": "system", "content": system_prompt.replace(NEWS_REP, combined_text)},
        {"role": "user", "content": "一言コメントをお願いします{"}
    ]
    completion = client.chat.completions.create(
        model=MODEL,
        messages=messages
    )
    comment = completion.choices[0].message.content
    print(comment)
    return comment

def get_ogp_data(url):
    try:
        headers = {
            'User-Agent': (
                'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
                ' AppleWebKit/537.36 (KHTML, like Gecko)'
                ' Chrome/91.0.4472.124 Safari/537.36'
            )
        }
        response = requests.get(url, headers=headers)
        soup = BeautifulSoup(response.text, 'html.parser')
        ogp_data = {
            'title': None,
            'description': None,
            'image': None,
            'url': None,
            'type': None
        }
        for prop in ogp_data.keys():
            meta = soup.find('meta', property=f'og:{prop}')
            if meta:
                ogp_data[prop] = meta['content']
        if ogp_data['title']:
            ogp_data['title'] = ogp_data['title'].replace('\\r', '')
        return ogp_data
    except Exception as e:
        print(f"Error fetching OGP data: {e}")
    return None

def __get_contens_str(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.content, 'html.parser')
    paragraphs = soup.find_all('p')
    combined_text = ' '.join(p.get_text(strip=True) for p in paragraphs)
    return combined_text

def process_article(url):
    contents = __get_contens_str(url)
    ogp = get_ogp_data(url)
    comment = create_comments(contents)
    post_to_x(url, comment, ogp)

def lambda_handler(event, context):
    try:
        table = ensure_dynamodb_table()
        print("### PRTimesからの記事取得 処理開始 ###")
        press_releases = __get_press_release(table)
        for press_release in press_releases:
            url = "<https://prtimes.jp>" + press_release['link']
            process_article(url)
        print("### PRTimesからの記事取得 処理終了 ###")
    except Exception as e:
        print(f"An error occurred: {e}")
        raise

    return {
        'statusCode': 200,
        'body': json.dumps('Process completed successfully')
    }

AWSにデプロイする

プロジェクトのルートディレクトリで以下のコマンドを実行する。

docker build -t pr_times_bot .

Docker イメージのビルド

docker build コマンドは、カレントディレクトリ（最後の .）にある Dockerfile を元に、コンテナイメージをビルドするコマンドです。
t pr_times_bot はビルドされるイメージに付与する“タグ”を指定していて、ここではpr_times_bot という名前でイメージを作成します。
結果として、Dockerfile の手順に従った環境（Python ランタイムやライブラリを含む）が「pr_times_bot」という名前のイメージとしてローカルに作成されます。

sls deploy

Serverless Framework を使用したデプロイ

sls は Serverless Framework の CLI コマンドです。
sls deploy はプロジェクトに含まれる serverless.yml（または同等の設定ファイル）に従い、AWS Lambda などのクラウド環境へコードをアップロードし、必要なリソースを設定・更新します。
具体的には、Lambda 関数としてアップロードされるコードや IAM 権限、DynamoDB などのサービス設定が一括で自動反映されます。

AWSコンソールでデプロイが成功しているか確認する

AWSコンソールにログインし、Lambdaを開く。
pr-times-bot-dev-appが作成されているか確認する。

コンソール上から【テスト】をクリックし、関数を実行してみる。

以下のように、コメントやOGPが正しく投稿されていたら成功です。

lambda_function.pyの解説

PRTimes からの記事取得部分

Python

def __get_press_release(table):
    url = '<https://prtimes.jp/topics/keywords/%E7%94%9F%E6%88%90AI>'
    ...
    for article in articles[:5]:
        ...
        if title == latest_title:
            break
        ...
    if new_latest_title:
        _save_latest_content(table, 'prtimes_last_title', new_latest_title)
    return press_releases

URL を変える: PRTimes の特定キーワードページをスクレイピングしています。今回は「AI」ですが、別のキーワードや他のニュースサイトに変える場合はここを切り替えてください。
最新タイトル管理: latest_title を DynamoDB に保存し、前回処理した記事以降のみを取得します。ここをカスタムすることで、「何件前まで遡って取得するか」などの制御が可能です。
取得件数の変更: articles[:5] を変更すれば、より多く or 少ない数のニュースを取得できます。

ニュース記事本文の抽出

Python

def __get_contens_str(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.content, 'html.parser')
    paragraphs = soup.find_all('p')
    combined_text = ' '.join(p.get_text(strip=True) for p in paragraphs)
    return combined_text

本文の抽出方法: 今はシンプルに <p> タグをまとめて取得しています。別の構造のサイトをスクレイピングする場合は、タグやクラス名を変えるなどして柔軟に処理を変えてください。
事前処理の追加: ここで余計な文字や広告を除去したり、Markdown／HTMLタグを削除したりするロジックを自分用に追加するのがおすすめです。

OpenAI でのコメント生成

Python

def create_comments(combined_text):
    MODEL = "gpt-4o"
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "hoge"))
    system_prompt = """
    入力されたニュース記事の内容から一言コメントをお願いします。
    コメントは語尾に「コレ」とつくような口調で。
    コメントは経済評論家の視点で確信をつくようなコメントをお願いします。
    2文が好ましい。

    <news>
    ###NEWS
    </news>
    """
    ...
    completion = client.chat.completions.create(...)
    comment = completion.choices[0].message.content
    return comment

プロンプトの調整: “語尾をどうするか” や “何文で書くか” など、生成されるコメントの個性はここのプロンプト (system_prompt) を変えるだけで大きく変わります。
モデル指定: 今回は "gpt-4o" となっていますが、実際の用途に合わせて他のLLMに変更可能です。

OGP データ取得

Python

def get_ogp_data(url):
    ...
    ogp_data = {
        'title': None,
        'description': None,
        'image': None,
        'url': None,
        'type': None
    }
    for prop in ogp_data.keys():
        meta = soup.find('meta', property=f'og:{prop}')
        if meta:
            ogp_data[prop] = meta['content']
    ...
    return ogp_data

OGP 情報をどう活用するか: OGP は SNS などでシェアした際のタイトル・画像・説明文を取得できる便利な仕組みです。不要ならこの部分を削除・縮小できますし、逆にここを拡張して追加のメタ情報を収集することも可能です。

X(Twitter) 投稿

Python

def post_to_x(url, comment, ogp):
    title = ogp['title']
    if title is None:
        return
    content = f"""{comment}
    {url}
    """
    twitter_client.create_tweet(text=content)

投稿フォーマット: 現状はコメントと URL を改行しながら一緒に投稿しています。content を組み立てるところで文言やハッシュタグを追加したり、添付メディアを扱う場合は Tweepy の他のメソッドを利用したりできます。

Lambda のエントリーポイント

Python

def lambda_handler(event, context):
    table = ensure_dynamodb_table()
    press_releases = __get_press_release(table)
    for press_release in press_releases:
        url = "<https://prtimes.jp>" + press_release['link']
        process_article(url)
    ...
    return {
        'statusCode': 200,
        'body': json.dumps('Process completed successfully')
    }

定期実行: AWS Lambda を Amazon EventBridge (旧 CloudWatch イベント) でスケジュール呼び出しすることで、定期的にスクレイピング＋コメント生成＋Twitter 投稿が自動化されます。
複数サイト対応: PRTimes 以外にも別のサイトをスクレイピングしたい場合、__get_press_release のような関数を増やして、ループ処理で process_article を複数回呼び出す設計にするとよいでしょう。

自社のビジネスに関連するコア情報を自動で収集しましょう

今回は、PR TIMESにて発表されるAI情報を収集してXに投稿するまでの自動ツールについて解説しました。
あなたのビジネスの業界で毎日キャッチアップしたい情報などあれば、ぜひ応用していただければと思います。
他の情報カテゴリ、他のメディア、Xではなくて自社のSlackやMicrosoft Teamsにて投稿させたり、さまざまなカスタマイズができます。

30万円ほどで開発支援可能

もし実現したいけど、開発をすることに躓く場合などや、自社内にエンジニアがいない場合など、コーレで開発支援をすることもできますので、その際はお気軽にお問い合わせください。
内容によりますが、30万円ほどで開発できるはずです。

コーレではAIシステムの要求整理〜要件定義〜デザイン〜開発まですべてサポート

コーレができること

コーレでは、AI戦略などの上流工程から、AI導入・AI開発・AI開発におけるアノテーションなどの下請け工程まで、一気通貫支援も、スポット支援も、幅広く柔軟に対応しています。

お問い合わせお待ちしています

コーレは、戦略コンサルタント、デザイナー、エンジニアが中心となり、AIとビジネスをつなぐAIコネクティブカンパニーです。戦略・企画から制作や開発、マーケティング支援や営業代行まで、一気通貫で上流から末端まで担うパートナーとして伴走します。お客様の要望に沿ったオーダーメイドなサポートをします。
お気軽にご連絡ください：https://co-r-e.net/contact/
コーレについてもっと知りたい方は、こちらから会社パンフレットをご覧ください：コーレ株式会社パンフレット

メソッド

最新情報を効率よく取得できるかが命取り

完全自動情報収集X投稿AIツールの概要

機能概要

システム構成

インフラストラクチャ

アプリケーション機能

デプロイメント

主要なPythonライブラリ

監視/ロギング

環境変数

構成図

デプロイするための前提条件

手順

TwitterAPIの準備

必要なファイルを作成

AWSにデプロイする

Docker イメージのビルド

Serverless Framework を使用したデプロイ

AWSコンソールでデプロイが成功しているか確認する

lambda_function.pyの解説

PRTimes からの記事取得部分

ニュース記事本文の抽出

OpenAI でのコメント生成

OGP データ取得

X(Twitter) 投稿

Lambda のエントリーポイント

自社のビジネスに関連するコア情報を自動で収集しましょう

30万円ほどで開発支援可能

コーレではAIシステムの要求整理〜要件定義〜デザイン〜開発まですべてサポート

コーレができること

お問い合わせお待ちしています

記事カテゴリー

人気記事

資料請求

ご相談/お問い合わせ