Spaces:
Sleeping
Sleeping
| 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/watermelon" | |
| local_dir = "game" | |
| if not os.path.exists(local_dir): | |
| os.mkdir(local_dir) | |
| hf_api = HfApi() | |
| def file_service(service, arg: str, token: str): | |
| """download game, upload game, or list games""" | |
| if service == "download game": | |
| filepath = arg.strip() + ".json" | |
| res = download( | |
| dataset_id, | |
| filepath, | |
| repo_type="dataset", | |
| localdir=local_dir, | |
| 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) | |
| if not hf_api.file_exists( | |
| repo_id=dataset_id, | |
| filename=md5code + ".json", | |
| repo_type="dataset", | |
| token=token, | |
| ): | |
| break | |
| sleep(0.1) | |
| obj["upload-time"] = beijing().__str__() | |
| maxtry -= 1 | |
| filename = md5code + ".json" | |
| if not maxtry and hf_api.file_exists( | |
| repo_id=dataset_id, | |
| filename=md5code + ".json", | |
| repo_type="dataset", | |
| token=token, | |
| ): | |
| print(f"[{service}] error: file exists") | |
| return None | |
| localpath = os.sep.join([local_dir, filename]) | |
| content = json.dumps(game, indent=4) | |
| with open(localpath, "w") as f: | |
| f.write(content) | |
| res = upload( | |
| localpath, | |
| filename, | |
| 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]: item for item in games if len(item) >= 32} | |
| print(f"[{service}] OK") | |
| return games | |
| else: | |
| print(f"[{service}] error: unknown service") | |
| return None | |