Programmiererforum

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}")