187 lines
8.2 KiB
Python
187 lines
8.2 KiB
Python
import json
|
|
import time
|
|
import requests
|
|
from datetime import datetime, timedelta, UTC
|
|
|
|
|
|
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:
|
|
if issue_check_id := kwargs.get('id'):
|
|
body = f'{body}\n----{issue_check_id}'
|
|
try:
|
|
response = requests.get(f'https://api.github.com/repos/{github_repository}/issues',
|
|
headers=github_api_headers(github_token=kwargs.get('github_token')),
|
|
params={
|
|
'state': 'open',
|
|
'since': (datetime.now(UTC) - timedelta(days=10)).strftime('%Y-%m-%dT%H:%M:%S') + 'Z',
|
|
'per_page': 100,
|
|
})
|
|
exist_issues = response.json()
|
|
for issue in exist_issues:
|
|
if issue.get('body').endswith(f'----{issue_check_id}'):
|
|
return
|
|
except Exception as e:
|
|
print(f'fetch from exist issues failed: {e}')
|
|
|
|
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'\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}')
|