Spaces:
Build error
Build error
File size: 4,435 Bytes
0f00fdc e63b9e1 0f00fdc e63b9e1 0f00fdc e63b9e1 0f00fdc 85db8c0 0f00fdc 85db8c0 0f00fdc 85db8c0 0f00fdc 85db8c0 0f00fdc 85db8c0 e63b9e1 0f00fdc 85db8c0 0f00fdc e63b9e1 0f00fdc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
from fastapi import APIRouter
import pgsoft.pgfile as pgfile
from pgsoft.pghash.md5 import md5
from pgsoft.pgdate.date_utils import beijing
from time import sleep
import json
import os
router = APIRouter(prefix="/file", tags=["File Service"])
admin_token_md5 = "01112732d503cd3f0898d185a11c8b92"
dataset_id = "pgsoft/game"
tempdir = "game"
db_token = os.environ.get("db_token")
if db_token:
print(db_token[:5])
@router.get("/download")
def download_file(game: str, token: str, gamecode: str):
if md5(token) != admin_token_md5:
print(f"[{beijing()}][download file] failed")
return {"status": "Failure", "detail": "Invalid token"}
game = game.strip().lower()
filename = gamecode.strip() + ".json"
remotepath = "/".join([game, filename[:2], filename])
res = pgfile.download(
dataset_id,
remotepath=remotepath,
repo_type="dataset",
localdir=tempdir,
token=db_token,
)
if not res:
print(f"[{beijing()}][download file] failed")
return {"status": "Failure", "detail": "File not found or server error"}
with open(res, "r") as f:
outp = json.load(f)
print(f"[{beijing()}][download file] OK")
return {"status": "OK", "result": outp}
@router.post("/upload")
def upload_file(game: str, token: str, content: dict):
if md5(token) != admin_token_md5:
print(f"[{beijing()}][upload file] failed")
return {"status": "Failure", "detail": "Invalid token"}
game = game.strip().lower()
needed_keys = ["game-file", "device-id"]
for key in needed_keys:
if key not in content:
print(f"[{beijing()}][upload file] failed, missed key {key}")
return {"status": "Failure", "detail": f"missed key {key}"}
if not isinstance(content["device-id"], str):
print(f'[{beijing()}][upload file] failed, "device-id" is not a str')
return {"status": "Failure", "detail": '"device-id" is not a str'}
if not isinstance(content["game-file"], dict):
print(f'[{beijing()}][upload file] failed, "game-file" is not a dict')
return {"status": "Failure", "detail": '"game-file" is not a dict'}
obj = {
"upload-time": beijing().__str__(),
"game-file": content["game-file"],
}
maxtry = 5
for retry in range(maxtry):
md5code = md5(obj)
remotepath = "/".join([game, md5code[:2], md5code + ".json"])
if not pgfile.api.file_exists(
repo_id=dataset_id,
filename=remotepath,
repo_type="dataset",
token=db_token,
):
break
sleep(0.1)
obj["upload-time"] = beijing().__str__()
maxtry -= 1
if not maxtry and pgfile.api.file_exists(
repo_id=dataset_id,
filename=remotepath,
repo_type="dataset",
token=db_token,
):
print(f"[{beijing()}][upload file] failed, timeout, please retry")
return {"status": "Failure", "detail": "timeout, please retry"}
filedir = os.path.join(tempdir, game, md5code[:2])
if not os.path.exists(filedir):
os.makedirs(filedir)
filepath = os.path.join(filedir, md5code + ".json")
content_indented = json.dumps(content, indent=4)
with open(filepath, "w") as f:
f.write(content_indented)
res = pgfile.upload(
filepath,
remotepath,
dataset_id,
"dataset",
db_token,
f"Updated at {beijing()}",
)
if not res:
print(f"[{beijing()}][upload file] failed")
return {"status": "Failure", "detail": "server error"}
print(f"[{beijing()}][upload file] OK")
return {"status": "OK", "result": md5code}
@router.get("/list")
def list_files(game: str, token: str):
if md5(token) != admin_token_md5:
print(f"[{beijing()}][list files] failed")
return {"status": "Failure", "detail": "Invalid token"}
game = game.strip().lower()
games = pgfile.list_files(
repo_id=dataset_id,
repo_type="dataset",
token=db_token,
)
if games is None:
print(f"[{beijing()}][list files] failed")
return {"status": "Failure", "detail": "server error"}
games = {
item.split(".")[0][-32:]: item
for item in games
if item.endswith(".json") and item.startswith(game)
}
print(f"[{beijing()}][list files] OK")
return {"status": "OK", "result": games}
|