|
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: |
|
|
|
|
|
|
|
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 = "hubei-hunan/games" |
|
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 not games: |
|
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 |
|
|
