Vertex AI Vision + Google Custom Search API + ChatGPT で作るドラクエウォーク おすすめの宝の地図を教えてくれるチャットボット
ドラゴンクエストウォーク4周年おめでとうございます。リリース当初から楽しんでいます。
最近は娘もすっかりハマってしまい家にはスライムのぬぐるみがいくつも。冒頭の写真は娘が作った真鍮板のレリーフ。
今回は娘に負けじと僕もドラクエで工作してみます。
このページで利用している株式会社スクウェア・エニックスを代表とする共同著作者が権利を所有する画像の転載・配布は禁止いたします。
(C) ARMOR PROJECT/BIRD STUDIO/SQUARE ENIX All Rights Reserved.
4周年で宝の地図登場
このCMはドキドキしました。
地図選ぶのが面倒
宝の地図、とっても楽しいんですけれど、地図を選ぶのがむずかしい。ランクやシェアされた数で大体の評価はわかるけれど中身の詳細はわからないようになっている。適当に選んでもいいのだけれど、持てる地図の数や交換に必要なポイントに限りがあるのでより良い地図を選びたい。
そうすると良さそうな地図名をネットで検索することになるんだけどこれがいちいち面倒。自動化できないか?
作業の流れ
スクリーンショットを手動で撮ったあとは自動化できそう。
ゲーム内のうけとり地図一覧のスクリーンショットを撮る(ここは手作業)
スクリーンショットからテキストを抽出(Vertex AI Vision 使用)
テキストから地図名を見つける
地図名でグーグル検索して地図情報ページを見つける(game8.jpで探す)
ページをスクレイピングして地図の詳細情報を取得
自分の要望と地図情報をChatGPTに渡し返事を得る
実装開始
今回もグーグルさんにはいっぱいお世話になるのでColabで作ります。
最初に必要なライブラリをまとめてインストールしてしまいます。ランタイムの再起動を要求されるので最初にまとめました。
# for Vison API
!pip install --upgrade google-cloud-vision
# for google search API
!pip install --upgrade google-api-python-client
# for scraping
!pip install httpx
!pip install openai
スクリーンショットはグーグルドライブの特定フォルダに置いてあるとします。
# グーグルドライブと接続
from google.colab import drive
drive.mount('/content/drive')
スクリーンショット一覧取得。
import glob
src_files = glob.glob('/content/drive/MyDrive/dqw_maps/*.png')
スクリーンショットから地図名を抽出
Vision API利用のため、認証ファイルを準備して環境変数に読み込ませる。
from googleapiclient import discovery
from oauth2client.client import GoogleCredentials
import os
cred_path = '/content/drive/MyDrive/dqw_maps/credentials.json'
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = cred_path
スクリーンショットからテキストを抽出します。画像1枚につき処理に数秒かかるので非同期処理しています。画像10枚で15秒ぐらいかかっていた処理が非同期処理で3秒ほどになりました。
from google.cloud import vision
# Colabそのものがイベントループ内で動いているのでasyncioをimportする前に以下の2行が必要
import nest_asyncio
nest_asyncio.apply()
import asyncio
# 抽出したテキストを地図ごとに格納
text_annotation = {}
async def process_image(src_file):
client = vision.ImageAnnotatorClient()
with io.open(src_file, 'rb') as image_file:
content = image_file.read()
image = vision.Image(content=content)
res = await asyncio.to_thread(client.text_detection, image=image)
texts = res.text_annotations
text_annotation[src_file] = texts[0]
async def main():
tasks = [process_image(src_file) for src_file in src_files]
await asyncio.gather(*tasks)
asyncio.run(main())
APIの詳細は以下を見てください。
テキストから地図名を取得
この部分は少し考えました。
抽出データの全文を観察すると地図名の前には必ずスペースが入っていたのでスペースで始まり一つ以上の任意の文字が続き「地図」で終わる部分を地図名としました。
import re
#スペースで始まり一つ以上の任意の文字が続き「地図」で終わる正規表現
pattern = r" .+地図"
maps = []
map_names = []
for src_file in src_files:
description = text_annotation[src_file].description
if isinstance(description, str):
maps = re.finditer(pattern, description)
for map in maps:
# スペース削除
map_name = map.group().strip()
map_names.append(map_name)
# 重複する地図名を削除
map_names = list(set(map_names))
地図名でグーグル検索
検索する前に一工夫。
地図情報は保存して次回以降は再度検索しないようにします。
地図名一覧から情報取得済みの地図を省きます。
import os
map_data_file_path = '/content/drive/MyDrive/dqw_maps/map_data/'
def exists_map_data(map_name):
map_file = f"{map_data_file_path}{map_name}.json"
return os.path.exists(map_file)
map_names_need_search = [x for x in map_names if not exists_map_data(x)]
プログラム可能な検索エンジンを作ります。詳細は以前書いた記事を見てください。
検索対象のサイトとして「game8.jp/dqwalk/*」としました。
Game8さんは地図名ごとにページを作ってくれていてとても検索しやすかったです。
地図名での検索結果からタイトルに地図名が含まれているページのURLを取得します。
from googleapiclient.discovery import build
# API クライアントを構築
service = build("customsearch", "v1", developerKey=API_KEY)
# 地図名がタイトルとして含まれるページのURLを取得
search_results = []
if len(map_names_need_search ) > 0
for search_query in map_names_need_search:
# API リクエストを送信します
response = service.cse().list(q=search_query, cx=SEARCH_ENGINE_ID).execute()
if "items" in response:
for item in response["items"]:
title = item["title"]
link = item["link"]
# タイトルに地図名が含まれるページ
if search_query in title:
search_results.append([search_query, link])
break
ここも本当は非同期処理したほうが良いと思う。あるいはクエリを工夫すればAPIを叩く回数を減らせるかも。intitleという演算子もあるようなのでこれ
当初はグーグル検索じゃなくてX(旧ツイッター)を検索しようと思ったのですが、XのAPIが高くなってしまったのでやめました。地図のスクリーンショット大量にポストされているので話が早いと思ったのですが。
地図の詳細情報取得
地図ページのURL先からHTMLを取得して情報を取得します。
ページのHTMLの構造から必要部分だけ取り出します。
時間がかかる処理なので非同期で処理。
import json
from bs4 import BeautifulSoup
import nest_asyncio
nest_asyncio.apply()
import asyncio
import httpx
async def scraping_doc(map_name,url):
async with httpx.AsyncClient() as client:
try:
# URLからHTMLを取得
response = await client.get(url)
response.raise_for_status() # エラーチェック
# ファイルにテキストを保存
output_file = f"{map_data_file_path}{map_name}.json"
# BeautifulSoupを使ってHTMLを解析
soup = BeautifulSoup(response.content, 'html.parser')
# h3要素のリストを取得
h3_elements = soup.find_all('h3')
# 結果を格納するためのリスト
result = []
# h3要素をループで処理
for h3 in h3_elements:
if h3.get_text() == "出現モンスターの詳細情報":
# "出現モンスターの詳細情報"を持つh3要素が見つかった場合
next_table = h3.find_next_sibling('table')
if next_table:
tr_elements = next_table.find_all('tr')
result.extend(tr_elements)
json_data = {'mapName': map_name, 'monster': []}
# 結果を表示
for tr in result:
th = tr.find('th')
if th:
th_text = th.get_text()
if th_text:
td_elements = tr.find_all('td')
monster = {'category': th_text,'name': '','quantity': 1,'location': ''}
if td_elements[0]:
monster_text = td_elements[0].get_text(strip=True)
if monster_text:
name_quantity = monster_text.split('×')
monster['name'] = name_quantity[0]
if len(name_quantity) == 2:
monster['quantity'] = int(name_quantity[1])
if td_elements[1]:
location_text = td_elements[1].get_text(strip=True)
if location_text:
monster['location'] = location_text
json_data['monster'].append(monster)
with open(output_file, 'w', encoding='utf-8') as file:
json.dump(json_data, file, ensure_ascii=False, indent=4)
except client.exceptions.RequestException as e:
print("エラーが発生しました:", e)
async def main():
tasks = []
for result in search_results:
map_name = result[0]
url = result[1]
tasks.append(scraping_doc(map_name,url))
#タスク実行
await asyncio.gather(*tasks)
if len(search_results) > 0:
asyncio.run(main())
取得した情報はjsonファイルとして保存しました。
一例「福井のグリーンティのねむれる地図」。
{
"mapName": "福井のグリーンティのねむれる地図",
"monster": [
{
"category": "お宝1",
"name": "マーブルン",
"quantity": 2,
"location": "地域限定2章8話など(ときどき見かける)"
},
{
"category": "お宝2",
"name": "マーブルン",
"quantity": 2,
"location": "地域限定2章8話など(ときどき見かける)"
},
{
"category": "お宝3",
"name": "マーブルン",
"quantity": 2,
"location": "地域限定2章8話など(ときどき見かける)"
},
{
"category": "ボス",
"name": "エリミネーター",
"quantity": 1,
"location": "9章7話など(ときどき見かける)"
}
]
}
ChatGPTにおすすめを尋ねる
目当てのモンスターがあるのならChatGPTに聞くまでもないのですが、最近の流行に乗っていきます。
!pip install openai
import openai
import glob
import io
import json
openai.api_key = OPENAI_API
openai.organization = OPENAI_ORGANIZATION
GPT_MODEL = 'gpt-4'
mad_data_files = glob.glob('/content/drive/MyDrive/dqw_maps/map_data/*.json')
map_datas = ''
#jsonファイルを読み込んでテキストにして単純に連結
for map_data_file in mad_data_files:
with io.open(map_data_file, 'r', encoding='utf-8') as f:
map_data = json.dumps(json.load(f), ensure_ascii=False, indent=0)
map_datas += map_data
query = 'おすすめ地図教えて'
messages = [
{"role": "system", "content": f'A treasure map is a map where you can always get こころ from ボス monsters and お宝 monsters. Follow the steps from the map data to answer the questions.If the question does not include the name of the monster, please suggest a map with more 地域限定 monsters. Step 1. Display the contents of the map (map name | ボス name | お宝1 monster name x number of pieces | お宝2 monster name x number of pieces | お宝3 monster name x number of pieces) in a table. Step 2. Bullet point the name of the recommended map and the reason for the recommendation. Include the total number of pieces per monster per map. Below is the map data.```{map_datas}```'},
{"role": "user", "content": query},
]
response = openai.ChatCompletion.create(
model=GPT_MODEL,
messages=messages,
temperature=0
)
response_message = response["choices"][0]["message"]["content"]
print(response_message)
これで完成です。今回はプロンプトも多少工夫しているので解説します。
プロンプト解説
最初プロンプトは日本語のみで書いていましたが認識がいまいちだったので英語にしました。ゲーム内用語はそのまま日本語にしました。
最初に宝の地図の説明を入れました。これはなくても返答の質には関係ないのですが、ゲームの仕組みに即した返答になるので読んで自然です。
次に、ステップバイステップで返答するように指示じています。
これは、得られる心の数の総数を計算させるためです。
(ChatGPTに計算させる必要ないんだけどね)
質問に特定のモンスター名が含まれない時の評価基準を与えています。今回は地域限定モンスターがたくさん含まれるものを選ぶように指示しています。ここは工夫のしどころですね。すでに所有しているこころ一覧を予め与えたり、有用なこころの情報を与えたりすれば良いものができますね。
次に、地図情報を表にまとめ、おすすめを箇条書きで表示し、手に入る心の数の合計を示すように指示しています。
返答例
けっこういい感じです。「ときどき見かける」などの出現頻度の意味を知らないのでちょっと違和感ありますが、これは教えてないので仕方がないです。
特定のモンスターを含めて質問した場合。
ちゃんと合計数を計算して答えてくれています。
今後の展望
地図のランクを含めたい。
自分がすでに持っているこころ一覧をまとめて与えたい。
こころ一覧画面のスクショさえとればできるな。
でも結局スクショが面倒だな。
今週は娘と富士急ハイランド。楽しみ。
この記事が気に入ったらサポートをしてみませんか?