75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
"""
|
||
代码托管平台 API 的统一网络层（GitHub / Gitea）。
|
||
|
||
- 所有出站请求集中在此，按 flavor 注入对应平台的推荐请求头与鉴权方式；
|
||
- 代理支持显式传入（GitHub 用 GH_PROXY、Gitea 用 GITEA_PROXY）；
|
||
- 未显式配置代理时 requests 仍会自动遵循容器的 HTTPS_PROXY/HTTP_PROXY 环境变量。
|
||
"""
|
||
import requests
|
||
|
||
from . import app_settings
|
||
|
||
GITHUB_API = 'https://api.github.com'
|
||
API_VERSION = '2026-03-10'
|
||
DEFAULT_TIMEOUT = 15
|
||
|
||
|
||
def proxies(proxy=None):
|
||
"""返回 requests 使用的 proxies 字典。
|
||
|
||
proxy=None 时沿用 GitHub 的 GH_PROXY（历史默认）；
|
||
显式传 '' 表示直连（环境变量仍生效），传具体地址则使用该代理。
|
||
"""
|
||
if proxy is None:
|
||
proxy = app_settings.get_value('GH_PROXY').strip()
|
||
if proxy:
|
||
return {'http': proxy, 'https': proxy}
|
||
return {}
|
||
|
||
|
||
def build_headers(token=None, json_content=False, flavor='github'):
|
||
"""按平台风格构建请求头。
|
||
|
||
- github：application/vnd.github+json + 版本头 + Bearer 鉴权；
|
||
- gitea：application/json + Gitea 官方文档主推的 token 鉴权方案。
|
||
"""
|
||
flavor = (flavor or 'github').lower()
|
||
if flavor == 'gitea':
|
||
headers = {'Accept': 'application/json'}
|
||
if token:
|
||
headers['Authorization'] = f'token {token}'
|
||
else:
|
||
headers = {
|
||
'Accept': 'application/vnd.github+json',
|
||
'X-GitHub-Api-Version': API_VERSION,
|
||
}
|
||
if token:
|
||
headers['Authorization'] = f'Bearer {token}'
|
||
if json_content:
|
||
headers['Content-Type'] = 'application/json'
|
||
return headers
|
||
|
||
|
||
def request(method, url, *, token=None, timeout=DEFAULT_TIMEOUT, headers=None,
|
||
flavor='github', proxy=None, **kwargs):
|
||
"""发起平台 API 请求，自动注入代理、超时与标准请求头。"""
|
||
kwargs.setdefault('proxies', proxies(proxy))
|
||
kwargs.setdefault('timeout', timeout)
|
||
merged_headers = build_headers(
|
||
token=token,
|
||
json_content=kwargs.get('json') is not None,
|
||
flavor=flavor,
|
||
)
|
||
if headers:
|
||
merged_headers.update(headers)
|
||
kwargs['headers'] = merged_headers
|
||
return requests.request(method, url, **kwargs)
|
||
|
||
|
||
def get(url, **kwargs):
|
||
return request('GET', url, **kwargs)
|
||
|
||
|
||
def post(url, **kwargs):
|
||
return request('POST', url, **kwargs)
|