(
session: aiohttp.ClientSession, filepath: str, url: str, pbar: tqdm = None
)
| 121 | |
| 122 | |
| 123 | async def download_file( |
| 124 | session: aiohttp.ClientSession, filepath: str, url: str, pbar: tqdm = None |
| 125 | ) -> None: |
| 126 | os.makedirs(os.path.dirname(filepath), exist_ok=True) |
| 127 | async with session.get( |
| 128 | url, params={"access_token": os.environ["ZENODO_TOKEN"]} |
| 129 | ) as response: |
| 130 | if response.status == 200: |
| 131 | with open(filepath, "wb") as f: |
| 132 | f.write(await response.read()) |
| 133 | elif response.status != 404: |
| 134 | raise Exception(f"Failed to download {filepath}: {response.status}") |
| 135 | if pbar is not None: |
| 136 | pbar.update(1) |
| 137 | |
| 138 | |
| 139 | def collect_links(jsonl_file: str) -> None: |
no outgoing calls
no test coverage detected