BetaWmManager.py

import asyncio
from collections import defaultdict
import regex
from datetime import datetime, timedelta
import time
from BetaManager import Manager
from utils import get_similar_key, type_like_human

import json


class WmManager(Manager):

    def __init__(self, context, odds_store, browser_manager):
        super().__init__(context, odds_store, browser_manager)
        self.plateform = "wm"
        self.base_link = "https://www.winamax.fr/"

    async def get_match_locators(self, page):

        matches = await page.locator("div[data-testid^='match-card-']").all()
        return matches

    async def accept_cookies(self, page):
        await asyncio.sleep(3)
        if self.accepted_cookies == False:
            await page.get_by_role("button", name="Tout accepter").click(timeout=1000)
            self.accepted_cookies = True

    async def close_pop_up(self, page):
        try:
            # Try to find and click the "Fermer" button within 3 seconds
            await page.get_by_role("button", name="Fermer").click(timeout=1000)
        except:
            # If not found within 3 seconds, just continue
            pass

    async def wait_for_title_stabilization(self, page, timeout=3):
        previous_title = await page.title()
        stable_time = 0
        interval = 0.2  # How often to check (in seconds)

        for _ in range(int(timeout / interval)):
            await asyncio.sleep(interval)
            current_title = await page.title()
            if current_title != previous_title:
                return  # Title is stable

    async def get_match_info(self, e):
        odd_buttons = e.locator("div[data-testid^='odd-button']")

        # can have 3 buttons in the middle for draw
        first_text = await odd_buttons.first.inner_text()

        text_only = [
            line
            for line in first_text.splitlines()
            if regex.search(r"\p{L}", line, regex.UNICODE)
        ]

        first = text_only[0]

        second_text = await odd_buttons.last.inner_text()

        text_only = [
            line
            for line in second_text.splitlines()
            if regex.search(r"\p{L}", line, regex.UNICODE)
        ]

        second = text_only[0]
        return first, second

    def is_less_than(self, n_hours, day_str="", time_str=""):

        # handle live matches exclude for the moment
        if "MT" in time_str:
            return False
        # French weekday mapping to English
        french_weekdays = {
            "Lun.": 0,
            "Mar.": 1,
            "Mer.": 2,
            "Jeu.": 3,
            "Ven.": 4,
            "Sam.": 5,
            "Dim.": 6,
        }
        french_months = {
            "Janv.": "Jan",
            "Févr.": "Feb",
            "Mars": "Mar",
            "Avr.": "Apr",
            "Mai": "May",
            "Juin": "Jun",
            "Juil.": "Jul",
            "Août": "Aug",
            "Sept.": "Sep",
            "Oct.": "Oct",
            "Nov.": "Nov",
            "Déc.": "Dec",
        }
        now = datetime.now()

        # à 19h45  = 19:45 today
        hour, minute = time_str.split(":")
        if day_str == "":
            target_time = now.replace(
                hour=int(hour), minute=int(minute), second=0, microsecond=0
            )
        elif day_str == "Demain":
            target_time = (now + timedelta(days=1)).replace(
                hour=int(hour), minute=int(minute), second=0, microsecond=0
            )
        elif day_str in french_weekdays:
            today = datetime.today()
            target_weekday = french_weekdays[
                day_str
            ]  # Convert to weekday number (Monday = 0, ..., Sunday = 6)
            days_ahead = (
                target_weekday - today.weekday()
            ) % 7  # Days until pevious occurrence

            # If today is the same as the target weekday, move to next week
            if days_ahead == 0:
                days_ahead = 7

            next_date = today + timedelta(days=days_ahead)
            day_str = next_date.strftime("%d/%m/%Y")
            target_time = datetime.strptime(
                f"{day_str}  {hour}:{minute}", "%d/%m/%Y %H:%M"
            )
        else:
            # Replace French month names with English equivalents
            for fr, en in french_months.items():
                day_str = day_str.replace(fr, en)
            target_time = datetime.strptime(
                f"{day_str} 2025 {hour}:{minute}", "%d %b %Y %H:%M"
            )

        return target_time - now < timedelta(hours=n_hours)

    async def get_day_time(self, e):
        """
        e is a match locator
        """
        time_div = e.locator("//div[contains(text(), ':')]").first
        time_str = await time_div.inner_text()

        day_str = await time_div.evaluate(
            """
                el => el.parentElement && el.parentElement.children.length > 0 
                    ? el.parentElement.children[0].innerText 
                    : null
            """
        )
        if day_str is None or day_str == time_str:
            day_str = ""

        return day_str, time_str

    async def get_balance(self, page):
        if not self.is_connected:
            return 0
        balance = page.locator("div#money-block  div.value").first
        try:
            amount = await balance.inner_text()
            amount = float(amount.split()[0].replace(",", "."))
            return amount
        except Exception as e:
            self.log(f"Unable to get balance amount : {e}")

        return 0

    async def get_bets(self, link, matches, sport):

        page = await self.context.new_page()

        if not self.is_connected:
            await self.connect()

        await page.goto(link)
        self.browser_manager.balances[self.plateform] = await self.get_balance(page)

        match_locators = await self.get_match_locators(page)
        found = 0
        self.log(
            f"Got  {  len(match_locators) if match_locators else 0  } matches in {link} : looking for {matches}"
        )
        # filter matches here  to get opponents pay attention to live or not

        for e in match_locators:
            matched_opponents = ""
            try:
                first, second = await self.get_match_info(e)
                matched_opponents = get_similar_key(
                    f"{first}_{second}", self.odds_store.data
                )
                self.log(f"{first}_{second} mapped to {matched_opponents}")
            except Exception as e:
                self.log(e)
            if matched_opponents in matches:

                async with self.context.expect_page() as new_page_info:

                    await e.click(modifiers=["Control"])
                    new_page = await new_page_info.value
                    await new_page.wait_for_load_state("load")

                    self.pages[matched_opponents] = new_page
                    # to be enhanced
                    for bet_name in self.bets:

                        bet_locators = self.get_bet_locators(sport, new_page, bet_name)

                        for index, loc in enumerate(bet_locators):
                            odd_identifier = index
                            odd = await loc.text_content()

                            await self.odds_store.update(
                                matched_opponents,
                                bet_name,
                                int(odd_identifier),
                                float(odd.replace(",", ".")),
                                self.plateform,
                            )
                            self.locators[matched_opponents][bet_name][
                                odd_identifier
                            ] = loc
                            task = self.fire_and_log(
                                self.monitor_element_changes(
                                    matched_opponents,
                                    bet_name,
                                    odd_identifier,
                                    loc,
                                )
                            )
                            self.monitors[matched_opponents].append(task)

                    # fire a background task to monitor the match status
                    task = self.fire_and_log(
                        self.monitor_match(matched_opponents, new_page)
                    )

                    found += 1
                    if found == len(matches):
                        break

        await page.close()

    # @profile
    async def place_bet(self, matched_opponents, bet_name, odd_identifier, stake):

        try:
            self.log(
                f"Recieved bet execution order {matched_opponents}  {bet_name}  {odd_identifier}  {stake}"
            )
            # clean
            page = self.pages[matched_opponents]
            main_div = page.locator("div[data-testid='sticky']")
            # clean
            delete_button = main_div.locator("svg").nth(0)
            await delete_button.click(timeout=2000, force=True)
            # click on odd
            await self.locators[matched_opponents][bet_name][odd_identifier].click(
                timeout=2000, force=True
            )

            # fill stake

            currency_input = main_div.locator("input").nth(0)
            stake = str(stake).replace(".", ",")
            await currency_input.click(timeout=2000, force=True)
            await currency_input.press("Control+A")  # Select all
            await currency_input.press("Backspace")  # Delete selected text
            await currency_input.fill(stake)

            bet_button = main_div.locator("button")

            # if all concerned bet excuters are ready then execute
            # while true check the readiness
            duration = 0
            while True:
                # check that the opportunity still exists then click on the button TODO click two times ??
                data = self.odds_store.data[matched_opponents][bet_name]
                inv_sum = 0
                odds = []
                concerned = False
                for outcome_name, outcome in data.outcomes.items():
                    odd = outcome.odd
                    platform = outcome.platform
                    inv_sum += 1 / odd
                    odds.append(odd)
                    if platform == self.plateform and outcome_name == odd_identifier:
                        concerned = True

                if (
                    inv_sum < 1
                    and concerned
                    # and await self.check_availability(
                    #     bet_button,
                    #     matched_opponents,
                    #     bet_name,
                    #     odd_identifier,
                    #     name="bet button",
                    # )
                ):
                    self.log(
                        f"{matched_opponents}  {bet_name}  {odd_identifier}  {stake}  Bet button is available and ready to be clicked "
                    )
                    self.browser_manager.readiness_flags[odd_identifier] = True
                else:
                    self.browser_manager.readiness_flags[odd_identifier] = False
                    self.log(
                        f" {matched_opponents}  {bet_name}  {odd_identifier}  {stake}  The odds changed New odds {odds} or the odd button is not clickable!"
                    )
                    return (
                        False,
                        f"{self.plateform}: opportunity not kept ",
                    )

                flags = self.browser_manager.readiness_flags.values()

                all_responded = all(f != "" for f in flags)
                all_ready = all(list(flags))
                if all_responded and all_ready:
                    if self.browser_manager.place_bets:
                        if "Accepter les changements" in await bet_button.inner_text():
                            await bet_button.click()
                        await bet_button.click()

                        await page.get_by_role(
                            "button", name=re.compile("fermer", re.IGNORECASE)
                        ).click()
                    self.log(
                        f" {matched_opponents}  {bet_name}  {odd_identifier}  {stake}  Bet executed !"
                    )
                    return (
                        True,
                        f"{self.plateform}: every thing available , opportunity kept and bet executed",
                    )
                await asyncio.sleep(0.2)
                duration += 0.2
                if duration > 2:
                    return (
                        False,
                        f"{self.plateform}: other bet executers didn't get ready quickly {self.browser_manager.readiness_flags}",
                    )

        except Exception as e:
            self.log(f"Error when executing bet: {e}")

            await self.odds_store.update(
                matched_opponents,
                bet_name,
                int(odd_identifier),
                1,
                self.plateform,
            )
            return (
                False,
                f"{self.plateform}: Error when executing bet: {e} ",
            )