app_util.py

import json

from gradio_client import Client
from pgsoft.pgdate.date_utils import beijing
from pgsoft.pgfile import download, upload, list_files
from pgsoft.pghash.md5 import md5
from time import sleep
from huggingface_hub import HfApi
import os


def call_logger(log_info, caller, hf_token) -> None:
    #######################
    # logging
    #######################
    calling_start = beijing()
    print(f"calling logger starts at {beijing()}")
    #################################################
    urls = [
        "https://hubei-hunan-logger.hf.space",
        "https://hubei-hunan-logger2.hf.space",
    ]
    for url in urls:
        try:
            client = Client(
                url,
                hf_token=hf_token,
                verbose=False,
            )
            client.submit(json.dumps(log_info), caller)
            print(f"[logging to {url}] OK")
        except Exception as e:
            print(f"[logging to {url}] error: {e}")
    #################################################
    calling_end = beijing()
    timecost = calling_end.timestamp() - calling_start.timestamp()
    print(f"calling logger ends at {calling_end}, costs {timecost:.2f}s")


dataset_id = "pgsoft/game"
tempdir = "game"
gamename = "house"
localdir = os.path.join(tempdir, gamename)
if not os.path.exists(localdir):
    os.makedirs(localdir)
hf_api = HfApi()


def file_service(service, arg: str, token: str):
    """download game, upload game, or list games"""
    if service == "download game":
        filename = arg.strip() + ".json"
        remotepath = "/".join(["house", filename[:2], filename])
        res = download(
            dataset_id,
            remotepath=remotepath,
            repo_type="dataset",
            localdir=tempdir,
            token=token,
        )
        if not res:
            return None
        with open(res, "r") as f:
            outp = json.load(f)
        print(f"[{service}] OK")
        return outp
    elif service == "upload game":
        try:
            game = json.loads(arg)
        except json.JSONDecodeError as e:
            print(f"[{service}] {type(e)}: {e}")
            return None

        if not isinstance(game, dict):
            print(f"[{service}] not a dict")
            return None

        needed_keys = ["game-file", "device-id"]
        for key in needed_keys:
            if key not in game:
                print(f'[{service}] error: missed "{key}"')
                return None
        if not isinstance(game["device-id"], str):
            print(f'[{service}] error: "device-id" is not a str')
            return None
        if not isinstance(game["game-file"], dict):
            print(f'[{service}] error: "game-file" is not a dict')
            return None

        obj = {
            "upload-time": beijing().__str__(),
            "game-file": game["game-file"],
        }

        maxtry = 5
        for retry in range(maxtry):
            md5code = md5(obj)
            remotepath = "/".join([gamename, md5code[:2], md5code + ".json"])
            if not hf_api.file_exists(
                repo_id=dataset_id,
                filename=remotepath,
                repo_type="dataset",
                token=token,
            ):
                break
            sleep(0.1)
            obj["upload-time"] = beijing().__str__()
            maxtry -= 1
        if not maxtry and hf_api.file_exists(
            repo_id=dataset_id,
            filename=remotepath,
            repo_type="dataset",
            token=token,
        ):
            print(f"[{service}] error: file exists")
            return None
        filedir = os.path.join(localdir, md5code[:2])
        if not os.path.exists(filedir):
            os.mkdir(filedir)
        filepath = os.path.join(filedir, md5code + ".json")
        content = json.dumps(game, indent=4)
        with open(filepath, "w") as f:
            f.write(content)
        res = upload(
            filepath,
            remotepath,
            dataset_id,
            "dataset",
            token,
            f"Updated at {beijing()}",
        )
        if not res:
            print(f"[{service}] error: upload failed")
            return None
        print(f"[{service}] OK")
        return md5code
    elif service == "list games":
        games = list_files(
            repo_id=dataset_id,
            repo_type="dataset",
            token=token,
        )
        if games is None:
            return None
        games = {
            item.split(".")[0][-32:]: item
            for item in games
            if item.endswith(".json") and item.startswith("house")
        }
        print(f"[{service}] OK")
        return games
    else:
        print(f"[{service}] error: unknown service")
        return None