zotero-addons-scraper/github_operations.py

import json
import time
import requests


def github_api_headers(**kwargs):
    result = {
        'Accept': 'application/vnd.github+json',
        'X-GitHub-Api-Version': '2022-11-28',
    }
    if github_token := kwargs.get('github_token'):
        result['Authorization'] = f'token {github_token}'
    return result


def report_issue(github_repository: str, title: str, body: str, **kwargs):
    if not github_repository:
        print('report issue repository not found')
        return
    try:
        response = requests.post(f'https://api.github.com/repos/{github_repository}/issues',
                                 headers=github_api_headers(github_token=kwargs.get('github_token')),
                                 json={
                                     'title': title,
                                     'body': body
                                 })

        if response.status_code == 201:
            print('Issue created successfully.')
            print('Issue URL:', response.json()['html_url'])
        else:
            print('Failed to create issue.')
            print('Response:', response.content)
    except Exception as e:
        print(f'report issue failed: {e}')


def upload_json_to_release(github_repository, release_id, upload_file_name, upload_file, **kwargs):
    headers = github_api_headers(github_token=kwargs.get('github_token'))
    headers["Content-Type"] = "application/octet-stream"
    upload_url = (f'https://uploads.github.com/repos/{github_repository}'
                  f'/releases/{release_id}/assets?name={upload_file_name}')
    try:
        with open(upload_file, "rb") as file:
            upload_resp = requests.post(upload_url, data=file, headers=headers)
            if upload_resp.status_code != 201:
                print(f'upload release assets code: {upload_resp.status_code}')
            else:
                print('upload release assets succeed')
    except Exception as e:
        print(f'upload release assets failed: {e}')


def create_release(github_repository, **kwargs):
    create_release_url = f'https://api.github.com/repos/{github_repository}/releases'
    cur_time = int(time.time())
    param = {
        'tag_name': f'{cur_time}',
        'target_commitish': 'publish',
        'name': f'{cur_time}',
        'body': f'![](https://img.shields.io/github/downloads/{github_repository}/{cur_time}/total?label=downloads)\n'
                f'publish addon_infos.json',
        'draft': False,
        'prerelease': False,
        'generate_release_notes': False,
    }
    try:
        create_resp = requests.post(create_release_url, json=param,
                                    headers=github_api_headers(github_token=kwargs.get('github_token')))
        if create_resp.status_code == 201:
            create_release_info = json.loads(create_resp.content)
            release_id = create_release_info['id']
            return release_id
        else:
            print(f'create release code: {create_resp.status_code}')
    except Exception as e:
        print(f'create release failed: {e}')


def delete_cache(github_repository, github_token, remain_count=2):
    headers = github_api_headers(github_token=github_token)
    get_caches_url = (f'https://api.github.com/repos/{github_repository}'
                      f'/actions/caches?per_page=100&sort=last_accessed_at&direction=desc')
    try:
        caches_resp = requests.get(get_caches_url, headers=headers)
        caches = json.loads(caches_resp.content)
        if caches.get('total_count', 0) < remain_count:
            return
        delete_cache_url = f'https://api.github.com/repos/{github_repository}/actions/caches/'
        for cache in caches.get('actions_caches', [])[remain_count:]:
            cache_key = cache.get('key')
            if cache_id := cache.get('id'):
                try:
                    delete_cache_resp = requests.delete(f'{delete_cache_url}{cache_id}', headers=headers)
                    if delete_cache_resp.status_code == 204:
                        print(f'delete {cache_key} succeed')
                    else:
                        print(f'delete {cache_key} failed: {delete_cache_resp.text}')
                except Exception as e:
                    print(f'delete cache for {cache_key} failed: {e}')
    except Exception as e:
        print(f'get caches failed: {e}')


def delete_release(github_repository, github_token, remain_count=2):
    headers = github_api_headers(github_token=github_token)
    get_release_url = (f'https://api.github.com/repos/{github_repository}'
                       f'/releases?per_page=100&page=1')
    try:
        releases_resp = requests.get(get_release_url, headers=headers)
        releases = json.loads(releases_resp.content)
        if len(releases) < remain_count:
            return
        releases.sort(key=lambda release: release['tag_name'], reverse=True)
        delete_release_url = f'https://api.github.com/repos/{github_repository}/releases/'
        for release in releases[remain_count:]:
            release_tag = release.get('tag_name')
            if release_id := release.get('id'):
                try:
                    delete_release_resp = requests.delete(f'{delete_release_url}{release_id}', headers=headers)
                    if delete_release_resp.status_code == 204:
                        print(f'delete release {release_tag} succeed')
                    else:
                        print(f'delete release {release_tag} failed: {delete_release_resp.text}')
                except Exception as e:
                    print(f'delete release for {release_tag} failed: {e}')
    except Exception as e:
        print(f'get releases failed: {e}')


def delete_tag(github_repository, github_token, remain_count=2):
    headers = github_api_headers(github_token=github_token)
    get_tags_url = (f'https://api.github.com/repos/{github_repository}'
                    f'/git/refs/tags')
    try:
        tags_response = requests.get(get_tags_url, headers=headers)
        tags = json.loads(tags_response.content)
        if len(tags) < remain_count:
            return
        tags.sort(key=lambda tag: tag['ref'], reverse=True)
        delete_tag_url = f'https://api.github.com/repos/{github_repository}/git/refs/tags/'
        for tag in tags[remain_count:]:
            if ref := tag.get('ref').replace('refs/tags/', ''):
                if len(ref) < 10:
                    continue
                try:  # automatically tag name is timestamp
                    int(ref)
                except ValueError:
                    continue
                try:
                    delete_tag_resp = requests.delete(f'{delete_tag_url}{ref}', headers=headers)
                    if delete_tag_resp.status_code == 204:
                        print(f'delete tag {ref} succeed')
                    else:
                        print(f'delete tag {ref} failed: {delete_tag_resp.text}')
                except Exception as e:
                    print(f'delete tag for {ref} failed: {e}')
    except Exception as e:
        print(f'get tags failed: {e}')


def rate_limit(github_token):
    try:
        resp = requests.get('https://api.github.com/rate_limit', headers=github_api_headers(github_token=github_token))
        rate = json.loads(resp.content)
        print(f'token rate {rate.get("rate")}')
    except Exception as e:
        print(f'get rate limit failed: {e}')