Warthunder Server Replay Downloader

download.py

import aiohttp
import asyncio
import socket
import time
import os

def getUrls(rid, count, directory):
    url = f"http://wt-game-replays.warthunder.com/{rid}/"
    urlList = []
    pathList = []
    for i in range(count):
        fname = f'{i:04d}.wrpl'
        link = f'{url}{fname}'
        path = os.path.join(directory, fname)
        urlList.append(link)
        pathList.append(path)
    return urlList, pathList

async def download_one(session, url, path):
    # create get request
    print(f'downloading:{url}')
    async with session.get(url) as response:
        # wait for response
        content = await response.read()
    print(f'downloaded -> {path}')
    with open(path, 'wb') as f:
        f.write(content)

async def download_all(rid, count, directory):
    urlList, pathList = getUrls(rid, count, directory)
    conn = aiohttp.TCPConnector(
        family=socket.AF_INET,
        ssl=False,
    )
    tasks =  []
    async with aiohttp.ClientSession(connector=conn) as session:
        for url, path in zip(urlList, pathList):
            task = asyncio.create_task(download_one(session, url, path))
            tasks.append(task)
        results = await asyncio.gather(*tasks)
    print('download finished')
    
directory = os.path.join(os.getcwd(), 'replay')
if not os.path.exists(directory):
    os.mkdir(directory)
print(f'download directory：{directory}')
replayId = input("find your rid：\nhttp://wt-game-replays.warthunder.com/{rid}\ninput rid:")
count = int(input("input number of wrpl files:"))

t0 = time.time()
asyncio.run(download_all(replayId, count, directory))
print(f'time elapse:{time.time() - t0}')
_ = input("press any key to exit")