|
|
|
|
|
import json |
|
import urllib.error |
|
import urllib.parse |
|
import urllib.request |
|
from concurrent.futures import ThreadPoolExecutor, as_completed |
|
|
|
BASE_API_URL = "https://projects.blender.org/api/v1" |
|
|
|
|
|
def url_json_get(url, data=None): |
|
try: |
|
if data: |
|
data = json.dumps(data).encode('utf-8') |
|
request = urllib.request.Request(url, data=data, method='POST') |
|
request.add_header('Content-Type', 'application/json') |
|
else: |
|
request = urllib.request.Request(url) |
|
|
|
response = urllib.request.urlopen(request) |
|
response_data = json.loads(response.read()) |
|
return response_data |
|
|
|
except urllib.error.URLError as ex: |
|
print("Error making HTTP request:", ex) |
|
return None |
|
|
|
|
|
def url_json_get_all_pages(url, item_filter=None, limit=50, exclude=set(), verbose=False): |
|
assert limit <= 50, "50 is the maximum limit of items per page" |
|
|
|
url_for_page = f"{url}&limit={limit}&page=" |
|
|
|
with urllib.request.urlopen(url_for_page + '1') as response: |
|
headers_first = response.info() |
|
json_data_first = json.loads(response.read()) |
|
|
|
total_count = int(headers_first.get('X-Total-Count')) |
|
total_pages = (total_count + limit - 1) // limit |
|
|
|
def fetch_page(page): |
|
if page == 1: |
|
json_data = json_data_first |
|
else: |
|
json_data = url_json_get(url_for_page + str(page)) |
|
|
|
if verbose: |
|
print(f"Fetched page {page}") |
|
|
|
data = [] |
|
for item in json_data: |
|
if exclude and int(item["number"]) in exclude: |
|
continue |
|
data.append({k: item[k] for k in item_filter} |
|
if item_filter else item) |
|
|
|
return data |
|
|
|
with ThreadPoolExecutor() as executor: |
|
futures = [executor.submit(fetch_page, page) |
|
for page in range(1, total_pages + 1)] |
|
all_results = [future.result() for future in as_completed(futures)] |
|
|
|
return [item for sublist in all_results for item in sublist] |
|
|
|
|
|
def gitea_json_issue_get(owner, repo, number): |
|
""" |
|
Get issue/pull JSON data. |
|
""" |
|
url = f"{BASE_API_URL}/repos/{owner}/{repo}/issues/{number}" |
|
return url_json_get(url) |
|
|
|
|
|
def gitea_fetch_issues(owner, repo, state='all', labels='', issue_attr_filter=None, since=None, exclude=set()): |
|
query_params = { |
|
'labels': labels, |
|
'state': state, |
|
'type': 'issues'} |
|
|
|
if since: |
|
query_params['since'] = since |
|
|
|
base_url = f"{BASE_API_URL}/repos/{owner}/{repo}/issues" |
|
encoded_query_params = urllib.parse.urlencode(query_params) |
|
issues_url = f"{base_url}?{encoded_query_params}" |
|
return url_json_get_all_pages(issues_url, item_filter=issue_attr_filter, exclude=exclude, verbose=True) |
|
|
|
|
|
def gitea_issues_body_updated_at_get(issues, verbose=True): |
|
def fetch_issue(issue): |
|
number = issue['number'] |
|
if verbose: |
|
print(f"Fetched issue #{number}") |
|
|
|
json_data = url_json_get( |
|
f"https://projects.blender.org/blender/blender/issues/{number}/content-history/list") |
|
|
|
if json_data and json_data['results']: |
|
return json_data['results'][0]['name'].split('datetime="')[1].split('"')[0] |
|
else: |
|
return issue['created_at'] |
|
|
|
with ThreadPoolExecutor() as executor: |
|
futures = [executor.submit(fetch_issue, issue) for issue in issues] |
|
all_results = [future.result() for future in as_completed(futures)] |
|
|
|
return all_results |
|
|
|
|
|
def gitea_wiki_page_get(owner, repo, page_name, verbose=True): |
|
""" |
|
Get a wiki page. |
|
""" |
|
encoded_page_name = urllib.parse.quote(page_name, safe='') |
|
base_url = f"{BASE_API_URL}/repos/{owner}/{repo}/wiki/page/{encoded_page_name}" |
|
return url_json_get(base_url) |
|
|
|
|
|
def gitea_wiki_pages_get(owner, repo, verbose=True): |
|
""" |
|
Get all wiki pages. |
|
""" |
|
base_url = f"{BASE_API_URL}/repos/{owner}/{repo}/wiki/pages" |
|
return url_json_get(base_url) |
|
|
