(jsonl_file: str)
| 137 | |
| 138 | |
| 139 | def collect_links(jsonl_file: str) -> None: |
| 140 | page = 1 |
| 141 | progress_bar = None |
| 142 | with jsonlines.open(jsonl_file, mode="w") as writer: |
| 143 | while progress_bar is None or progress_bar.n < progress_bar.total: |
| 144 | try: |
| 145 | results = search_zenodo(page=page, sort="-mostrecent") |
| 146 | except Exception as e: |
| 147 | tqdm.write(f"Error {e}, current page: {page}") |
| 148 | continue |
| 149 | if progress_bar is None: |
| 150 | progress_bar = tqdm( |
| 151 | initial=(page - 1) * 500, |
| 152 | total=results["hits"]["total"], |
| 153 | desc="Processing pages", |
| 154 | ) |
| 155 | progress_bar.update(len(results["hits"]["hits"])) |
| 156 | records = [] |
| 157 | for record in results["hits"]["hits"]: |
| 158 | for file in record["files"]: |
| 159 | if not file["key"].endswith(".pptx"): |
| 160 | continue |
| 161 | license = record["metadata"].get("license", {"id": "unknown"})["id"] |
| 162 | if license == "unknow": |
| 163 | continue |
| 164 | records.append( |
| 165 | { |
| 166 | "filename": file["key"], |
| 167 | "size": file["size"], |
| 168 | "url": file["links"]["self"], |
| 169 | "license": license, |
| 170 | "title": record["title"], |
| 171 | "created": record["created"], |
| 172 | "updated": record["updated"], |
| 173 | "doi": record.get("doi", "unknown"), |
| 174 | "checksum": file["checksum"], |
| 175 | } |
| 176 | ) |
| 177 | writer.write_all(records) |
| 178 | page += 1 |
| 179 | |
| 180 | |
| 181 | async def download_links(jsonl_file: str) -> None: |
no test coverage detected