Mobile versionAPI Nitrado-Dokumentation: https://doc.nitradado.net/#api-gameserv ... ileSupload
https://api.nitrado.net/services/:./GameSververs /File_Server/Upload
, die Antwort ist Erfolg (200.201 Code). Danach erhalte ich die URL für die Datenübertragung und das generierte Token (
"URL": "https://fileserver.nitrado.net/ MS2312 /Upload /",
" Token ":" 4307ee5a-6bfe-4913-950C-C422AC733CC0 "
) [Das Token ist jedes Mal variabel, ich weiß, dass es einzigartig ist, dies ist nur ein Beispiel] Ich erhalte auch die Antwort 200.201, was Erfolg bedeutet, aber wenn ich versuche, der Server hochzuladen /p>
Der Bot soll die Datei vom Server herunterladen, bearbeiten und erneut hochladen. Wenn Sie jedoch versuchen, hochzuladen Das ist mein Upload -Code: < /p>
Code: Select all
async def upload_file_to_url(upload_url, token, local_file_path, target_directory): # Added target_directory parameter
headers = {
"Authorization": f"Bearer {token}"
}
try:
async with aiohttp.ClientSession() as session:
form = aiohttp.FormData()
form.add_field(
'file',
open(local_file_path, 'rb'),
filename=os.path.basename(local_file_path),
content_type='multipart/form-data'
)
form.add_field(
'path',
target_directory
)
async with session.post(
upload_url,
headers=headers,
data=form,
timeout=30
) as response:
response_status = response.status
response_text = await response.text()
print(f"[DEBUG] Upload response status: {response_status}")
print(f"[DEBUG] Upload response text: {response_text}")
return response_status == 200
except Exception as e:
print(f"[ERROR] Upload failed with error: {str(e)}")
return False
async def upload_cfgeffectarea_to_server(local_file_path, target_directory, target_filename):
if not validate_json_file(local_file_path):
print("[ERROR] Cannot upload invalid JSON file.")
return False
upload_url = "https://api.nitrado.net/services/16605568/gameservers/file_server/upload"
headers = {
"Authorization": f"Bearer {NITRADO_API_KEY}"
}
if not target_directory.startswith('/'):
target_directory = '/' + target_directory
try:
async with aiohttp.ClientSession() as session:
async with session.post(
upload_url,
headers=headers,
json={
"path": target_directory,
"file": target_filename
},
timeout=30
) as response:
if response.status in (200, 201):
response_data = await response.json()
if 'data' in response_data and 'token' in response_data['data']:
token_data = response_data['data']['token']
upload_url = token_data.get('url')
token = token_data.get('token')
if not upload_url or not token:
print("[ERROR] Missing upload URL or token in response.")
return False
print(f"[INFO] Upload URL: {upload_url}")
print(f"[INFO] Upload token: {token}")
return await upload_file_to_url(upload_url, token, local_file_path, target_directory)
else:
print("[ERROR] Invalid response format from server")
print(f"Response data: {response_data}")
return False
else:
print(f"[ERROR] Failed to get upload token. Status: {response.status}")
print(await response.text())
return False
except Exception as e:
print(f"[ERROR] Request failed with error: {str(e)}")
return False
TEST FUNCTION
async def auto_update_cfg_effect_area():
print("[AUTO-UPDATE] Starting automatic test.")
try:
file_content = await download_cfgeffectarea_from_server()
if not file_content:
print("[AUTO-UPDATE] Failed to download file from server.")
return
data = file_content
test_area_name = f"AUTO_TEST_{uuid.uuid4().hex}"
new_area = {
"AreaName": test_area_name,
"Type": "TestType",
"Data": {
"Pos": [100.0, 100.0]
}
}
if "Areas" not in data:
data["Areas"] = []
data["Areas"].append(new_area)
print("[AUTO-UPDATE] Added new test area to JSON.")
local_file_path = "./cfgEffectArea.json"
with open(local_file_path, "w", encoding="utf-8") as file:
json.dump(data, file, indent=4)
target_directory = f"/games/{HOSTNAME}/ftproot/dayzxb_missions/dayzOffline.chernarusplus"
success = await upload_cfgeffectarea_to_server(local_file_path, target_directory, "cfgEffectArea.json")
if success:
print(f"[AUTO-UPDATE] Successfully uploaded updated file. Added area: {test_area_name}")
else:
print("[AUTO-UPDATE] Failed to upload updated file.")
except Exception as e:
print(f"[AUTO-UPDATE] Error during auto update: {e}")
