spawnrys/flanabot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add SteamBot

Browse Source
This commit is contained in:
AlberLC
2024-07-04 03:03:54 +02:00
parent 71a5cb74d9
commit 5f027e665a
7 changed files with 423 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
flanabot/bots/__init__.py
View File

@@ -5,5 +5,6 @@ from flanabot.bots.flana_tele_bot import *
from flanabot.bots.penalty_bot import *
from flanabot.bots.poll_bot import *
from flanabot.bots.scraper_bot import *
from flanabot.bots.steam_bot import *
from flanabot.bots.ubereats_bot import *
from flanabot.bots.weather_bot import *

3
flanabot/bots/flana_bot.py
View File

@@ -19,6 +19,7 @@ from flanabot.bots.connect_4_bot import Connect4Bot
from flanabot.bots.penalty_bot import PenaltyBot
from flanabot.bots.poll_bot import PollBot
from flanabot.bots.scraper_bot import ScraperBot
from flanabot.bots.steam_bot import SteamBot
from flanabot.bots.ubereats_bot import UberEatsBot
from flanabot.bots.weather_bot import WeatherBot
from flanabot.models import Action, BotAction, ButtonsGroup, Chat, Message
@@ -27,7 +28,7 @@ from flanabot.models import Action, BotAction, ButtonsGroup, Chat, Message
# ----------------------------------------------------------------------------------------------------- #
# --------------------------------------------- FLANA_BOT --------------------------------------------- #
# ----------------------------------------------------------------------------------------------------- #
class FlanaBot(Connect4Bot, PenaltyBot, PollBot, ScraperBot, UberEatsBot, WeatherBot, MultiBot, ABC):
class FlanaBot(Connect4Bot, PenaltyBot, PollBot, ScraperBot, SteamBot, UberEatsBot, WeatherBot, MultiBot, ABC):
Chat = Chat
Message = Message

6
flanabot/bots/poll_bot.py
View File

@@ -25,8 +25,8 @@ class PollBot(MultiBot, ABC):
super()._add_handlers()
self.register(self._on_choose, keywords=constants.KEYWORDS['choose'], priority=2)
self.register(self._on_choose, keywords=constants.KEYWORDS['random'], priority=2)
self.register(self._on_choose, keywords=(constants.KEYWORDS['choose'], constants.KEYWORDS['random']), priority=2)
self.register(self._on_choose, keywords=multibot_constants.KEYWORDS['random'], priority=2)
self.register(self._on_choose, keywords=(constants.KEYWORDS['choose'], multibot_constants.KEYWORDS['random']), priority=2)
self.register(self._on_delete_votes, extra_kwargs={'all_': True}, keywords=(multibot_constants.KEYWORDS['deactivate'], multibot_constants.KEYWORDS['all'], constants.KEYWORDS['vote']))
self.register(self._on_delete_votes, extra_kwargs={'all_': True}, keywords=(multibot_constants.KEYWORDS['delete'], multibot_constants.KEYWORDS['all'], constants.KEYWORDS['vote']))
@@ -97,7 +97,7 @@ class PollBot(MultiBot, ABC):
discarded_words = {
*constants.KEYWORDS['choose'],
*constants.KEYWORDS['random'],
*multibot_constants.KEYWORDS['random'],
self.name.lower(), f'<@{self.id}>',
'entre', 'between'
}

355
flanabot/bots/steam_bot.py Normal file
View File

@@ -0,0 +1,355 @@
__all__ = ['SteamBot']
import asyncio
import base64
import itertools
import os
import random
import re
import urllib.parse
from abc import ABC
from collections import defaultdict
from collections.abc import Awaitable, Callable, Iterable, Iterator
from contextlib import asynccontextmanager
from typing import Any
import aiohttp
import flanautils
import playwright.async_api
import playwright.async_api
import plotly
from flanautils import Media, MediaType, Source
from multibot import LimitError, MultiBot, RegisteredCallback, constants as multibot_constants
import constants
from flanabot.models import Message, SteamRegion
# ----------------------------------------------------------------------------------------------------- #
# --------------------------------------------- constants.STEAM_BOT --------------------------------------------- #
# ----------------------------------------------------------------------------------------------------- #
class SteamBot(MultiBot, ABC):
# -------------------------------------------------------- #
# ------------------- PROTECTED METHODS ------------------ #
# -------------------------------------------------------- #
def _add_handlers(self):
super()._add_handlers()
self.register(self._on_steam, keywords='steam', priority=3)
async def _add_app_price(
self,
app_prices: dict[str, dict[str, float]],
steam_region: SteamRegion,
app_ids: Iterable[str],
session: aiohttp.ClientSession
) -> None:
for ids_batch_batch in itertools.batched(
itertools.batched(app_ids, constants.STEAM_IDS_BATCH), constants.STEAM_MAX_CONCURRENT_REQUESTS
):
gather_results = await asyncio.gather(
*(self._get_app_data(session, ids_batch, steam_region.code) for ids_batch in ids_batch_batch)
)
for gather_result in gather_results:
for app_id, app_data in gather_result.items():
if (
(data := app_data.get('data'))
and
(price := data.get('price_overview', {}).get('final')) is not None
):
app_prices[app_id][steam_region.code] = price / 100 / steam_region.eur_conversion_rate
@staticmethod
@asynccontextmanager
async def _create_browser_context(
browser: playwright.async_api.Browser
) -> Iterator[playwright.async_api.BrowserContext]:
async with await browser.new_context(
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.5615.49 Safari/537.36',
screen={
'width': 1920,
'height': 1080
},
viewport={
'width': 1920,
'height': 945
},
device_scale_factor=1,
is_mobile=False,
has_touch=False,
default_browser_type='chromium',
locale='es-ES'
) as context:
yield context
async def _get_app_data(
self,
session: aiohttp.ClientSession,
ids_batch: Iterable[str],
country_code: str
) -> dict[str, Any]:
async with session.get(
constants.STEAM_APP_ENDPOINT_TEMPLATE.format(ids=','.join(ids_batch), country_code=country_code)
) as response:
if response.status != 200:
raise LimitError('🚫 Steam ban: espera 5 minutos antes de consultar de nuevo.')
return await response.json()
@staticmethod
async def _get_exchange_data(session: aiohttp.ClientSession) -> dict[str, Any]:
async with session.get(
constants.STEAM_EXCHANGERATE_API_ENDPOINT.format(api_key=os.environ['EXCHANGERATE_API_KEY'])
) as response:
exchange_data = await response.json()
return exchange_data
async def _get_most_games_ids(self, browser: playwright.async_api.Browser) -> set[str]:
app_ids = set()
re_pattern = fr'{urllib.parse.urlparse(constants.STEAM_MOST_URLS[0]).netloc}/\w+/(\d+)'
async with self._create_browser_context(browser) as context:
context.set_default_timeout(10000)
page = await context.new_page()
for url in constants.STEAM_MOST_URLS:
await page.goto(url)
try:
await page.wait_for_selector('tr td a[href]')
except playwright.async_api.TimeoutError:
raise LimitError('🚫 Steam ban: espera 5 minutos antes de consultar de nuevo.')
for td in await page.locator('tr td a[href]').all():
href = await td.get_attribute('href')
app_ids.add(re.search(re_pattern, href).group(1))
return app_ids
@staticmethod
def _insert_exchange_rates(
steam_regions: dict[str, SteamRegion],
exchange_data: dict[str, Any]
) -> dict[str, SteamRegion]:
for steam_region in steam_regions.values():
alpha_3_code = constants.STEAM_REGION_CODE_MAPPING[steam_region.code]
steam_region.eur_conversion_rate = exchange_data['conversion_rates'][alpha_3_code]
return steam_regions
async def _scrape_steam_data(
self,
update_state: Callable[[str], Awaitable[Message]],
update_steam_regions=False,
most_games=True
) -> tuple[dict[str, SteamRegion], set[str]]:
steam_regions = {steam_region.code: steam_region for steam_region in SteamRegion.find()}
most_games_ids = set()
if update_steam_regions or most_games:
async with playwright.async_api.async_playwright() as playwright_:
async with await playwright_.chromium.launch() as browser:
if update_steam_regions:
await update_state('Actualizando las regiones de Steam...')
SteamRegion.delete_many_raw({})
steam_regions = await self._update_steam_regions(browser)
if most_games:
await update_state('Obteniendo los juegos más jugados y vendidos de Steam...')
most_games_ids = await self._get_most_games_ids(browser)
return steam_regions, most_games_ids
else:
return steam_regions, most_games_ids
async def _update_steam_regions(self, browser: playwright.async_api.Browser) -> dict[str, SteamRegion]:
steam_regions = {}
for app_id in constants.STEAM_APP_IDS_FOR_SCRAPE_COUNTRIES:
async with self._create_browser_context(browser) as context:
page = await context.new_page()
await page.goto(f'{constants.STEAM_DB_URL}/app/{app_id}/')
for td in await page.locator("#prices td[class='price-line']").all():
src = (await td.locator('img').get_attribute('src'))
name = (await td.text_content()).strip()
flag_url = f'{constants.STEAM_DB_URL}{src}'
region_code = await td.get_attribute('data-cc')
steam_region = SteamRegion(region_code, name, flag_url)
steam_region.save()
steam_regions[steam_region.code] = steam_region
await asyncio.sleep(2)
return steam_regions
# ---------------------------------------------- #
# HANDLERS #
# ---------------------------------------------- #
async def _on_steam(
self,
message: Message,
update_steam_regions: bool | None = None,
most_games: bool | None = None,
last_games: bool | None = None,
random_games: bool | None = None
):
if message.chat.is_group and not self.is_bot_mentioned(message):
return
if update_steam_regions is None:
update_steam_regions = bool(
self._parse_callbacks(
message.text,
[
RegisteredCallback(
...,
keywords=(multibot_constants.KEYWORDS['update'], constants.KEYWORDS['region'])
)
]
)
)
if most_games is None:
most_games = bool(
flanautils.cartesian_product_string_matching(
message.text,
('jugados', 'played', 'sellers', 'selling', 'vendidos'),
multibot_constants.PARSER_MIN_SCORE_DEFAULT
)
)
if last_games is None:
last_games = bool(
flanautils.cartesian_product_string_matching(
message.text,
multibot_constants.KEYWORDS['last'] + ('new', 'novedades', 'nuevos'),
multibot_constants.PARSER_MIN_SCORE_DEFAULT
)
)
if random_games is None:
random_games = bool(
flanautils.cartesian_product_string_matching(
message.text,
multibot_constants.KEYWORDS['random'],
multibot_constants.PARSER_MIN_SCORE_DEFAULT
)
)
if not any((most_games, last_games, random_games)):
most_games = True
last_games = True
random_games = True
update_state = self.create_message_updater(message, delete_user_message=True)
chart_title_parts = []
steam_regions, selected_app_ids = await self._scrape_steam_data(update_state, update_steam_regions, most_games)
if most_games:
chart_title_parts.append(f'los {len(selected_app_ids)} más jugados/vendidos')
async with aiohttp.ClientSession() as session:
if last_games or random_games:
await update_state('Obteniendo todas las aplicaciones de Steam...')
async with session.get(constants.STEAM_ALL_APPS_ENDPOINT) as response:
all_apps_data = await response.json()
app_ids = [
str(app_id) for app_data in all_apps_data['applist']['apps'] if (app_id := app_data['appid'])
]
if last_games:
selected_app_ids.update(app_ids[-constants.STEAM_LAST_GAMES:])
chart_title_parts.append(f'los {constants.STEAM_LAST_GAMES} más nuevos')
if random_games:
selected_app_ids.update(random.sample(app_ids, constants.STEAM_RANDOM_GAMES))
chart_title_parts.append(f'{constants.STEAM_RANDOM_GAMES} aleatorios')
exchange_data = await self._get_exchange_data(session)
steam_regions = self._insert_exchange_rates(steam_regions, exchange_data)
bot_state_message = await update_state('Obteniendo los precios para todas las regiones...')
apps_prices = defaultdict(dict)
try:
await asyncio.gather(
*(
self._add_app_price(apps_prices, steam_region, selected_app_ids, session)
for steam_region in steam_regions.values()
)
)
except LimitError:
await self.delete_message(bot_state_message)
raise
apps_prices = {k: v for k, v in apps_prices.items() if len(v) == len(steam_regions)}
for app_prices in apps_prices.values():
for region_code, price in app_prices.items():
steam_regions[region_code].total_price += price
await update_state('Creando gráfico...')
steam_regions_values = sorted(steam_regions.values(), key=lambda steam_region: steam_region.total_price)
region_names = []
region_total_prices = []
bar_colors = []
bar_line_colors = []
images = []
for i, steam_region in enumerate(steam_regions_values):
region_names.append(steam_region.name)
region_total_prices.append(steam_region.total_price)
bar_colors.append(steam_region.bar_color)
bar_line_colors.append(steam_region.bar_line_color)
async with session.get(steam_region.flag_url) as response:
images.append({
'source': f'data:image/svg+xml;base64,{base64.b64encode(await response.read()).decode()}',
'xref': 'x',
'yref': 'paper',
'x': i,
'y': 0.04,
'sizex': 0.425,
'sizey': 0.0225,
'xanchor': 'center',
'opacity': 1,
'layer': 'above'
})
figure = plotly.graph_objs.Figure(
[
plotly.graph_objs.Bar(
x=region_names,
y=region_total_prices,
marker={'color': bar_colors},
)
]
)
figure.update_layout(
width=1920,
height=945,
margin={'t': 20, 'r': 20, 'b': 20, 'l': 20},
title={
'text': f'{len(apps_prices)} juegos<br><sup>(solo los comparables entre {flanautils.join_last_separator(chart_title_parts, ', ', ' y ')})</sup>',
'xref': 'paper',
'yref': 'paper',
'x': 0.5,
'y': 0.93,
'font': {
'size': 35
}
},
images=images,
xaxis={'tickfont': {
'size': 14
}},
yaxis={'ticksuffix': ' €', 'tickfont': {
'size': 18
}}
)
bot_state_message = await update_state('Enviando...')
await self.send(Media(figure.to_image(), MediaType.IMAGE, 'png', Source.LOCAL), message)
await self.delete_message(bot_state_message)
# -------------------------------------------------------- #
# -------------------- PUBLIC METHODS -------------------- #
# -------------------------------------------------------- #

22
flanabot/constants.py
View File

@@ -23,6 +23,25 @@ PUNISHMENT_INCREMENT_EXPONENT = 6
PUNISHMENTS_RESET_TIME = datetime.timedelta(weeks=2)
RECOVERY_DELETED_MESSAGE_BEFORE = datetime.timedelta(hours=1)
SCRAPING_TIMEOUT_SECONDS = 10
STEAM_ALL_APPS_ENDPOINT = 'https://api.steampowered.com/ISteamApps/GetAppList/v2'
STEAM_APP_ENDPOINT_TEMPLATE = 'https://store.steampowered.com/api/appdetails?appids={ids}&cc={country_code}&filters=price_overview'
STEAM_APP_IDS_FOR_SCRAPE_COUNTRIES = (400, 620, 730, 210970, 252490, 292030, 427520, 1712350)
STEAM_DB_URL = 'https://steamdb.info'
STEAM_EXCHANGERATE_API_ENDPOINT = 'https://v6.exchangerate-api.com/v6/{api_key}/latest/EUR'
STEAM_IDS_BATCH = 750
STEAM_LAST_GAMES = 1500
STEAM_MAX_CONCURRENT_REQUESTS = 10
STEAM_MOST_URLS = (
'https://store.steampowered.com/charts/mostplayed',
'https://store.steampowered.com/charts/topselling/global'
)
STEAM_RANDOM_GAMES = 1000
STEAM_REGION_CODE_MAPPING = {'eu': 'EUR', 'ru': 'RUB', 'pk': 'USD', 'ua': 'UAH', 'za': 'ZAR', 'vn': 'VND', 'tw': 'TWD',
'id': 'IDR', 'my': 'MYR', 'ar': 'USD', 'tr': 'USD', 'ph': 'PHP', 'in': 'INR', 'cn': 'CNY',
'br': 'BRL', 'sa': 'SAR', 'th': 'THB', 'pe': 'PEN', 'cl': 'CLP', 'kw': 'KWD', 'az': 'USD',
'kz': 'KZT', 'co': 'COP', 'mx': 'MXN', 'qa': 'QAR', 'sg': 'SGD', 'jp': 'JPY', 'uy': 'UYU',
'ae': 'AED', 'kr': 'KRW', 'hk': 'HKD', 'cr': 'CRC', 'nz': 'NZD', 'ca': 'CAD', 'au': 'AUD',
'il': 'ILS', 'us': 'USD', 'no': 'NOK', 'uk': 'GBP', 'pl': 'PLN', 'ch': 'CHF'}
BANNED_POLL_PHRASES = (
'Deja de dar por culo {presser_name} que no puedes votar aqui',
@@ -106,7 +125,8 @@ KEYWORDS = {
'controla', 'destroy', 'destroza', 'duro', 'ejecuta', 'enseña', 'escarmiento', 'execute', 'fuck',
'fusila', 'hell', 'humos', 'infierno', 'jefe', 'jode', 'learn', 'leccion', 'lesson', 'manda', 'paliza',
'purgatorio', 'purgatory', 'sancion', 'shoot', 'teach', 'whip'),
'random': ('aleatorio', 'azar', 'random'),
'region': ('countries', 'country', 'pais', 'paises', 'region', 'regiones', 'regions', 'zona', 'zonas', 'zone',
'zones'),
'scraping': ('busca', 'contenido', 'content', 'descarga', 'descargar', 'descargues', 'download', 'envia', 'scrap',
'scrapea', 'scrapees', 'scraping', 'search', 'send'),
'self': (('contigo', 'contra', 'ti'), ('mismo', 'ti')),

1
flanabot/models/__init__.py
View File

@@ -5,4 +5,5 @@ from flanabot.models.heating_context import *
from flanabot.models.message import *
from flanabot.models.player import *
from flanabot.models.punishment import *
from flanabot.models.steam_region import *
from flanabot.models.weather_chart import *

40
flanabot/models/steam_region.py Normal file
View File

@@ -0,0 +1,40 @@
__all__ = ['SteamRegion']
from dataclasses import dataclass
from typing import Any
from flanautils import DCMongoBase, FlanaBase
@dataclass
class SteamRegion(DCMongoBase, FlanaBase):
collection_name = 'steam_region'
unique_keys = ('code',)
code: str
name: str
flag_url: str
eur_conversion_rate: float | None = None
total_price: int = 0
def __post_init__(self):
super().__post_init__()
match self.code:
case 'eu':
self.bar_color = '#2b6ced'
self.bar_line_color = '#0055ff'
case 'in':
self.bar_color = '#ffc091'
self.bar_line_color = '#ff6600'
case 'ar':
self.bar_color = '#8adcff'
self.bar_line_color = '#00b3ff'
case 'tr':
self.bar_color = '#ff7878'
self.bar_line_color = '#ff0000'
case _:
self.bar_color = '#68a9f2'
self.bar_line_color = '#68a9f2'
def _mongo_repr(self) -> Any:
return {k: v for k, v in super()._mongo_repr().items() if k in ('code', 'name', 'flag_url')}