30 lines
1.3 KiB
Python
30 lines
1.3 KiB
Python
import json, os, time, hashlib
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError
|
|
|
|
def request(url, headers=None, timeout=90, retries=4):
|
|
h={"User-Agent":"vulnerability-apex-project-mirror/1.0", "Accept":"application/json"}
|
|
h.update(headers or {})
|
|
for attempt in range(retries):
|
|
try:
|
|
with urlopen(Request(url,headers=h),timeout=timeout) as response:
|
|
return response.status, response.read(), dict(response.headers)
|
|
except HTTPError as err:
|
|
if err.code == 304: return 304,b'',dict(err.headers)
|
|
if err.code not in (429,500,502,503,504) or attempt==retries-1: raise
|
|
except (TimeoutError, OSError):
|
|
if attempt==retries-1: raise
|
|
time.sleep(min(30, 2**attempt*2))
|
|
raise RuntimeError('unreachable')
|
|
|
|
def store(path, raw):
|
|
path=Path(path); path.parent.mkdir(parents=True,exist_ok=True)
|
|
if path.exists() and hashlib.sha256(path.read_bytes()).digest()==hashlib.sha256(raw).digest(): return False
|
|
temp=path.with_name(path.name+'.tmp')
|
|
temp.write_bytes(raw); temp.replace(path)
|
|
return True
|
|
|
|
def state_file(path, value):
|
|
store(path,(json.dumps(value,indent=2,ensure_ascii=False,sort_keys=True)+'\n').encode())
|