1分に1回、タスクスケジューラ等で呼ぶこと前提。
import asyncio
from datetime import datetime, timedelta
import pytz
import requests
from bs4 import BeautifulSoup
import discord
import aiohttp
# Discordボットのトークン
DISCORD_TOKEN = 'Discordボットのトークン'
CHANNEL_ID = '投稿したいチャンネルのID'
def parse_time(date_str, time_str):
try:
# 日付と時間を結合
datetime_str = f"{date_str} {time_str}"
# 26:00 のような表記を処理
if ':' in time_str:
hours, minutes = map(int, time_str.split(':'))
if hours >= 24:
date_obj = datetime.strptime(date_str, "%m/%d")
date_obj += timedelta(days=1)
hours -= 24
datetime_str = f"{date_obj.strftime('%m/%d')} {hours:02d}:{minutes:02d}"
event_date = datetime.strptime(datetime_str, "%m/%d %H:%M")
event_date = event_date.replace(year=datetime.now().year)
return pytz.timezone('Asia/Tokyo').localize(event_date)
except ValueError as e:
print(f"Error parsing date/time: {e} for {date_str} {time_str}")
return None
async def get_hirose_events():
url = "https://hirose-fx.co.jp/contents/ecclndr/"
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'
}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as response:
print(f"Status Code: {response.status}")
html_content = await response.text()
print(f"Content length: {len(html_content)}")
soup = BeautifulSoup(html_content, 'html.parser')
events = []
rows = soup.select('table.layout-top.row-data.font-small tr')
current_date = None
for row in rows:
cells = row.select('td')
if len(cells) >= 6:
date_cell = cells[0].text.strip()
time = cells[1].text.strip()
currency = cells[2].find('img')['title'] if cells[2].find('img') else ''
event_name = cells[3].text.strip()
if date_cell:
current_date = date_cell.split()[0]
if current_date and time and currency in ['日本', '米国']:
event_date = parse_time(current_date, time)
if event_date:
events.append({
"name": f"{currency} - {event_name}",
"time": event_date
})
print(f"Total JPY and USD events found: {len(events)}")
return events
async def report_upcoming_events():
fx_events = await get_hirose_events()
now = datetime.now(pytz.timezone('Asia/Tokyo'))
target_time = now + timedelta(minutes=30)
upcoming_events = []
for event in fx_events:
time_until_event = event['time'] - now
if timedelta(minutes=29) <= time_until_event <= timedelta(minutes=30):
upcoming_events.append(event)
if upcoming_events:
message = "30分後に予定されている日本と米国のイベント:\n"
for event in upcoming_events:
message += f"{event['time'].strftime('%Y-%m-%d %H:%M')} - {event['name']}\n"
# Discordに接続してメッセージを送信
client = discord.Client(intents=discord.Intents.default())
@client.event
async def on_ready():
try:
channel = await client.fetch_channel(CHANNEL_ID)
await channel.send(message)
print(f"Message sent to channel {CHANNEL_ID}")
finally:
await client.close()
await client.start(DISCORD_TOKEN)
else:
print("No events scheduled in the next 30 minutes.")
async def main():
await report_upcoming_events()
if __name__ == "__main__":
asyncio.run(main())
