ERROR:Herobrain:_find_game: Could not get game info from server (), retrying...

[LalitMohanSinghBohra]

[freshollie]

Hey @LalitMohanSinghBohra, I've not looked at what HQs new API is, nor do I have time to spend this project. Feel free to find the new API and send a PR or fork the project.

Your first start would either be looking at someone else's project which works, or decompiling their android APK.

[LalitMohanSinghBohra]

aye aye captain

[LalitMohanSinghBohra]

HackQ-Trivia is working correctly cant i just copy their HQ API and paste it here.

[freshollie]

Are you not able to edit the code yourself? You have the source running.

[LalitMohanSinghBohra]

no, I cannot edit code myself I am not that good at coding. I am just started to learn to code. Help please If you have some free time

[freshollie]

Well here's your start. You are already halfway there by being able to run the app. Edit it and see what happens, you don't learn till you try.

I'm not willing to spend time on this project, I am happy to answer any specific questions.

[LalitMohanSinghBohra]

i will try my level best Senpai arigatōgozaimasu

[LalitMohanSinghBohra]

import json import logging

import colorama import lomond from unidecode import unidecode

from hackq_trivia.config import config from hackq_trivia.question_handler import QuestionHandler

class LiveShow: async def aenter(self): self.question_handler = QuestionHandler() return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
    await self.question_handler.close()

def __init__(self, headers):
    self.headers = headers
    self.show_question_summary = config.getboolean("LIVE", "ShowQuestionSummary")
    self.show_chat = config.getboolean("LIVE", "ShowChat")
    self.block_chat = False  # Block chat while question is active
    self.logger = logging.getLogger(__name__)
    self.logger.info("LiveShow initialized.")

async def connect(self, uri):
    websocket = lomond.WebSocket(uri)
    for header, value in self.headers.items():
        websocket.add_header(str.encode(header), str.encode(value))

    for event in websocket.connect(ping_rate=5):
        if event.name == "text":
            message = json.loads(event.text)
            self.logger.debug(message)

            if "error" in message and message["error"] == "Auth not valid":
                raise ConnectionRefusedError("User ID/Bearer invalid. Please check your settings.ini.")
            elif message["type"] == "interaction" and self.show_chat and not self.block_chat:
                self.logger.info(f"{message['metadata']['username']}: {message['metadata']['message']}")
            elif message["type"] == "question":
                question = unidecode(message["question"])
                choices = [unidecode(choice["text"]) for choice in message["answers"]]

                self.logger.info("\n" * 5)
                self.logger.info(f"Question {message['questionNumber']} out of {message['questionCount']}")
                self.logger.info(question, extra={"pre": colorama.Fore.BLUE})
                self.logger.info(f"Choices: {', '.join(choices)}", extra={"pre": colorama.Fore.BLUE})

                await self.question_handler.answer_question(question, choices)

                self.block_chat = True
            elif self.show_question_summary and message["type"] == "questionSummary":
                question = unidecode(message["question"])
                self.logger.info(f"Question summary: {question}", extra={"pre": colorama.Fore.BLUE})

                for answer in message["answerCounts"]:
                    ans_str = unidecode(answer["answer"])

                    self.logger.info(f"{ans_str}:{answer['count']}:{answer['correct']}",
                                     extra={"pre": colorama.Fore.GREEN if answer['correct'] else colorama.Fore.RED})

                self.logger.info(f"{message['advancingPlayersCount']} players advancing")
                self.logger.info(f"{message['eliminatedPlayersCount']} players eliminated\n")
            elif self.show_chat and self.block_chat and message["type"] == "questionClosed":
                self.block_chat = False
                self.logger.info("\n" * 5)

    self.logger.info("Disconnected.")

Is this the correct HQ API and where do i put it if this is correct

[LalitMohanSinghBohra]

import asyncio import json.decoder import time from datetime import datetime

import colorama import jwt import nltk import requests import logging import logging.config

from hackq_trivia.config import config from hackq_trivia.live_show import LiveShow

class BearerError(Exception): """Raise when bearer token is invalid/expired"""

class HackQ: HQ_URL = f"https://api-quiz.hype.space/shows/schedule?type=hq"

def __init__(self):
    HackQ.download_nltk_resources()
    colorama.init()

    self.bearer = config.get("CONNECTION", "BEARER")
    self.timeout = config.getfloat("CONNECTION", "Timeout")
    self.show_next_info = config.getboolean("MAIN", "ShowNextShowInfo")
    self.exit_if_offline = config.getboolean("MAIN", "ExitIfShowOffline")
    self.show_bearer_info = config.getboolean("MAIN", "ShowBearerInfo")
    self.headers = {"User-Agent": "Android/1.40.0",
                    "x-hq-client": "Android/1.40.0",
                    "x-hq-country": "US",
                    "x-hq-lang": "en",
                    "x-hq-timezone": "America/New_York",
                    "Authorization": f"Bearer {self.bearer}",
                    "Connection": "close"}

    self.session = requests.Session()
    self.session.headers.update(self.headers)

    self.init_root_logger()
    self.logger = logging.getLogger(__name__)

    # Find local UTC offset
    now = time.time()
    self.local_utc_offset = datetime.fromtimestamp(now) - datetime.utcfromtimestamp(now)

    self.validate_bearer()
    self.logger.info("HackQ-Trivia initialized.\n", extra={"pre": colorama.Fore.GREEN})

@staticmethod
def download_nltk_resources():
    for resource in {"stopwords", "averaged_perceptron_tagger", "punkt"}:
        nltk.download(resource, quiet=True)

@staticmethod
def init_root_logger():
    import os

    class LogFilterColor(logging.Filter):
        def filter(self, record):
            if "hackq" not in record.name and "__main__" not in record.name:
                return None

            if not hasattr(record, "pre"):
                record.pre = ""
                record.post = ""
            elif not hasattr(record, "post"):
                record.post = colorama.Style.RESET_ALL

            return record

    log_filename = config.get("LOGGING", "FILE")
    script_dir = os.path.dirname(os.path.abspath(__file__))
    if not os.path.isabs(log_filename):
        log_filename = os.path.join(script_dir, log_filename)

    with open(os.path.join(script_dir, "logging_config.json")) as log_conf_file:
        log_conf_dict = json.load(log_conf_file)
        log_conf_dict["handlers"]["fileHandler"]["filename"] = log_filename
        log_conf_dict["filters"]["LogFilterColor"]["()"] = LogFilterColor

        logging.config.dictConfig(log_conf_dict)

def validate_bearer(self):
    try:
        bearer_info = jwt.decode(self.bearer, verify=False)
    except jwt.exceptions.DecodeError:
        raise BearerError("Bearer invalid. Please check your settings.ini.")

    expiration_time = datetime.utcfromtimestamp(bearer_info["exp"])
    issue_time = datetime.utcfromtimestamp(bearer_info["iat"])

    if datetime.utcnow() > expiration_time:
        raise BearerError("Bearer expired. Please obtain another from your device.")

    if self.show_bearer_info:
        exp_local = expiration_time + self.local_utc_offset
        iat_local = issue_time + self.local_utc_offset

        self.logger.info("Bearer info:")
        self.logger.info(f"    Username: {bearer_info['username']}")
        self.logger.info(f"    Issuing time: {iat_local.strftime('%Y-%m-%d %I:%M %p')}")
        self.logger.info(f"    Expiration time: {exp_local.strftime('%Y-%m-%d %I:%M %p')}")

async def __connect_show(self, uri):
    async with LiveShow(self.headers) as show:
        await show.connect(uri)

def connect(self):
    while True:
        websocket_uri = self.get_next_show_info()

        if websocket_uri is not None:
            self.logger.info("Found WebSocket, connecting...\n", extra={"pre": colorama.Fore.GREEN})
            asyncio.run(self.__connect_show(websocket_uri))

def get_next_show_info(self):
    """
    Gets info of upcoming shows from HQ, prints it out if ShowNextShowInfo is True
    :return: The show's WebSocket URI if it is live, else None
    """
    try:
        response = self.session.get(self.HQ_URL, timeout=self.timeout).json()
        self.logger.debug(response)
    except json.decoder.JSONDecodeError:
        self.logger.info("Server response not JSON, retrying...", extra={"pre": colorama.Fore.RED})
        time.sleep(1)
        return None

    if "error" in response:
        if response["error"] == "Auth not valid":
            raise BearerError("Bearer invalid. Please check your settings.ini.")
        else:
            self.logger.warning(f"Error in server response: {response['error']}")
            time.sleep(1)
            return None

    next_show = response["shows"][0]
    if self.show_next_info:  # If desired, print info of next show
        start_time = datetime.strptime(next_show["startTime"], "%Y-%m-%dT%H:%M:%S.%fZ")
        start_time_local = start_time + self.local_utc_offset

        self.logger.info("Upcoming show:")
        self.logger.info(f"{next_show['display']['title']} - {next_show['display']['summary']}")
        self.logger.info(next_show["display"]["description"])
        if "subtitle" in next_show["display"]:
            self.logger.info(f"Subtitle: {next_show['display']['subtitle']}")
        self.logger.info(f"Prize: ${(next_show['prizeCents'] / 100):0,.2f} {next_show['currency']}")
        self.logger.info(f"Show start time: {start_time_local.strftime('%Y-%m-%d %I:%M %p')}")

    if "live" in next_show:  # Return found WebSocket URI
        return next_show["live"]["socketUrl"].replace("https", "wss")
    else:
        self.logger.info("Show not live.\n", extra={"pre": colorama.Fore.RED})
        if self.exit_if_offline:
            exit()

        time.sleep(5)
        return None

if name == "main": HackQ().connect() or is it this one ?

[freshollie]

HQ_URL = f"https://api-quiz.hype.space/shows/schedule?type=hq" looks right, also you might want to consider replacing the headers with the ones in HackQ Trivia.

[LalitMohanSinghBohra]

it not running now giving error when I tried to edit it.