|
| 1 | +import time |
| 2 | +from typing import List |
| 3 | + |
| 4 | +PKGS = [ |
| 5 | + 'computer-languages', |
| 6 | + 'data-languages', |
| 7 | + 'find-project-root', |
| 8 | + 'get-min-py', |
| 9 | + 'is-legacy-terminal', |
| 10 | + 'is-unicode-supported', |
| 11 | + 'latin-locales', |
| 12 | + 'markup-languages', |
| 13 | + 'non-latin-locales', |
| 14 | + 'programming-languages', |
| 15 | + 'project-markers', |
| 16 | + 'prose-languages', |
| 17 | + 'remove-json-keys', |
| 18 | + 'translate-messages' |
| 19 | +] |
| 20 | +STATS_API_URL = 'https://pypistats.org/api/packages/{pkg}/overall' |
| 21 | + |
| 22 | +def format_total(num: int) -> str: # abbr ints to e.g. 1.5k, 2b |
| 23 | + return f'{num / 1000000000:.1f}B' if num >= 1000000000 \ |
| 24 | + else f'{num / 1000000:.1f}M' if num >= 1000000 \ |
| 25 | + else f'{num / 1000:.1f}K' if num >= 1000 \ |
| 26 | + else str(num) |
| 27 | + |
| 28 | +def get_downloads(pkg: str, max_retries: int = 5, get_delay: int = 2) -> int: |
| 29 | + import json |
| 30 | + from urllib.request import urlopen |
| 31 | + from urllib.error import HTTPError |
| 32 | + url = STATS_API_URL.format(pkg=pkg) |
| 33 | + for idx in range(max_retries): |
| 34 | + try: |
| 35 | + with urlopen(url) as resp: |
| 36 | + return sum(item['downloads'] for item in json.load(resp)['data']) |
| 37 | + except HTTPError as err: |
| 38 | + if err.code == 429: # Rate limited |
| 39 | + retry_delay = (idx +1) *2 # Exponential backoff |
| 40 | + print(f'{pkg}: Rate limited. Retrying in {retry_delay}s...') |
| 41 | + time.sleep(retry_delay) |
| 42 | + else: |
| 43 | + print(f'{pkg}: ERROR ({err.code})') |
| 44 | + return 0 |
| 45 | + except Exception as err: |
| 46 | + print(f'{pkg}: Exception: {err}') |
| 47 | + time.sleep(get_delay) |
| 48 | + print(f'{pkg}: Failed after {max_retries} retries') |
| 49 | + return 0 |
| 50 | + |
| 51 | +def read_file(file_path: str) -> List[str]: |
| 52 | + with open(file_path, 'r', encoding='utf-8') as file: |
| 53 | + return file.readlines() |
| 54 | + |
| 55 | +def write_file(file_path: str, lines: List[str]) -> None: |
| 56 | + with open(file_path, 'w', encoding='utf-8') as file: |
| 57 | + file.writelines(lines) |
| 58 | + |
| 59 | +def update_downloads_shield(readme_path: str, downloads: int) -> None: |
| 60 | + import re |
| 61 | + lines = read_file(readme_path) |
| 62 | + shield_re = r'(<img[^>]+src="https://img.shields.io/badge/Downloads-)([\d\.kM]+)(-[a-f0-9]{6})' |
| 63 | + formatted_downloads = format_total(downloads) |
| 64 | + downloads_str = f'{formatted_downloads.lower()}' |
| 65 | + for idx, line in enumerate(lines): |
| 66 | + shield_match = re.search(shield_re, line) |
| 67 | + if shield_match: |
| 68 | + new_line = re.sub(shield_match.group(2), downloads_str, line) |
| 69 | + lines[idx] = new_line |
| 70 | + print(f'>>> {new_line.strip()}') |
| 71 | + write_file(readme_path, lines) |
| 72 | + |
| 73 | +def main() -> None: |
| 74 | + grand_total_dls = 0 |
| 75 | + for pkg in PKGS: # get downloads |
| 76 | + downloads = get_downloads(pkg) |
| 77 | + grand_total_dls += downloads |
| 78 | + print(f'{pkg:30} {downloads:,}') |
| 79 | + time.sleep(1) |
| 80 | + print('-' *45) |
| 81 | + print(f"{'TOTAL DOWNLOADS':20} {grand_total_dls:,}\n") |
| 82 | + README_PATH = 'docs/README.md' |
| 83 | + print(f'Updating {README_PATH}...') |
| 84 | + update_downloads_shield(README_PATH, grand_total_dls) |
| 85 | + print('Done!') |
| 86 | + |
| 87 | +if __name__ == '__main__' : main() |
0 commit comments