vollständige umstellung auf gitea repo
This commit is contained in:
Simon Daniel Streich committed
1 parent
3a807da54c
commit
079e69963d
9 files changed
+426
-39
No files matched your search
@@ -0,0 +1,225 @@
|
||||
"""Update timestamp and release manifests using only Python 3.10+.
|
||||
|
||||
Without --write, calculate changes without saving them. Never commit or push.
|
||||
PowerShell statements in manifests are not executed.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
from datetime import timezone
|
||||
from email.utils import parsedate_to_datetime
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.parse import urldefrag, urlsplit
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
|
||||
BUCKET = Path(__file__).resolve().parents[1] / 'bucket'
|
||||
TIMESTAMP_REGEX = r'(?<version>\d{8}T\d{6}Z)'
|
||||
|
||||
|
||||
def open_url(url, method='GET', headers=None):
|
||||
parsed = urlsplit(url)
|
||||
if parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password:
|
||||
raise ValueError('Nur HTTPS-URLs ohne eingebettete Zugangsdaten sind erlaubt.')
|
||||
return urlopen(Request(url, method=method, headers={
|
||||
'User-Agent': 'Medelexis-Scoop-Updater', 'Cache-Control': 'no-cache',
|
||||
'Accept-Encoding': 'identity', **(headers or {})}), timeout=180)
|
||||
|
||||
|
||||
def metadata(headers):
|
||||
return {key: headers.get(key, '') for key in ('Last-Modified', 'ETag')}
|
||||
|
||||
|
||||
def read_head(url):
|
||||
with open_url(url, 'HEAD') as response:
|
||||
return metadata(response.headers)
|
||||
|
||||
|
||||
def release_version(checkver):
|
||||
if checkver.get('jsonpath') != '$.tag_name' or not checkver.get('regex'):
|
||||
raise ValueError('Unterstützt werden Last-Modified oder jsonpath $.tag_name mit regex.')
|
||||
with open_url(checkver['url']) as response:
|
||||
tag = json.load(response)['tag_name']
|
||||
pattern = checkver['regex'].replace('(?<version>', '(?P<version>')
|
||||
match = re.search(pattern, tag)
|
||||
if not match:
|
||||
raise ValueError('Release-Tag passt nicht zum Versionsmuster.')
|
||||
version = match.groupdict().get('version') or match.group(1)
|
||||
if not re.fullmatch(r'[0-9A-Za-z][0-9A-Za-z._+-]*', version):
|
||||
raise ValueError('Ungültige Release-Version.')
|
||||
return version
|
||||
|
||||
|
||||
def download_file(url, observed, target):
|
||||
conditions = {}
|
||||
if observed['ETag'] and not observed['ETag'].startswith('W/'):
|
||||
conditions['If-Match'] = observed['ETag']
|
||||
elif observed['Last-Modified']:
|
||||
conditions['If-Unmodified-Since'] = observed['Last-Modified']
|
||||
size = 0
|
||||
prefix = b''
|
||||
with open_url(url, headers=conditions) as response:
|
||||
if metadata(response.headers) != observed:
|
||||
raise ValueError('Datei hat sich zwischen HEAD und GET geändert; erneut versuchen.')
|
||||
if response.headers.get('Content-Type', '').split(';')[0].lower() in (
|
||||
'text/html', 'application/json'):
|
||||
raise ValueError('Download liefert HTML/JSON statt einer Programmdatei.')
|
||||
expected = response.headers.get('Content-Length')
|
||||
total = int(expected) if expected is not None else None
|
||||
last_report = time.monotonic()
|
||||
if total is not None:
|
||||
print(f' Download gestartet: {total / 1048576:.1f} MiB.', flush=True)
|
||||
else:
|
||||
print(' Download gestartet; Gesamtgrösse unbekannt.', flush=True)
|
||||
with target.open('wb') as output:
|
||||
while chunk := response.read(1024 * 1024):
|
||||
if not size:
|
||||
prefix = chunk[:4]
|
||||
output.write(chunk)
|
||||
size += len(chunk)
|
||||
now = time.monotonic()
|
||||
if now - last_report >= 5:
|
||||
progress = f'{size / 1048576:.1f} MiB'
|
||||
if total:
|
||||
progress += f' / {total / 1048576:.1f} MiB ({size / total:.0%})'
|
||||
print(f' Heruntergeladen: {progress}', flush=True)
|
||||
last_report = now
|
||||
if not size or (expected is not None and size != int(expected)):
|
||||
raise ValueError('Leerer oder unvollständiger Download.')
|
||||
suffix = Path(urlsplit(url).path).suffix.lower()
|
||||
if suffix == '.exe' and not prefix.startswith(b'MZ'):
|
||||
raise ValueError('Download hat keine Windows-EXE-Signatur.')
|
||||
if suffix == '.zip' and not prefix.startswith(b'PK'):
|
||||
raise ValueError('Download hat keine ZIP-Signatur.')
|
||||
print(f' Download abgeschlossen: {size / 1048576:.1f} MiB.', flush=True)
|
||||
return size
|
||||
|
||||
|
||||
def hash_file(path):
|
||||
total = path.stat().st_size
|
||||
print(f' SHA-256-Berechnung gestartet: {total / 1048576:.1f} MiB.', flush=True)
|
||||
digest = hashlib.sha256()
|
||||
size = 0
|
||||
last_report = time.monotonic()
|
||||
with path.open('rb') as source:
|
||||
while chunk := source.read(1024 * 1024):
|
||||
digest.update(chunk)
|
||||
size += len(chunk)
|
||||
now = time.monotonic()
|
||||
if now - last_report >= 5:
|
||||
print(f' Gehasht: {size / 1048576:.1f} / {total / 1048576:.1f} MiB '
|
||||
f'({size / total:.0%})', flush=True)
|
||||
last_report = now
|
||||
print(' SHA-256-Berechnung abgeschlossen.', flush=True)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def download_hash(url, observed):
|
||||
with tempfile.TemporaryDirectory(prefix='scoop-update-') as directory:
|
||||
target = Path(directory) / 'download.bin'
|
||||
size = download_file(url, observed, target)
|
||||
return hash_file(target), size
|
||||
|
||||
|
||||
def save_manifest(path, manifest):
|
||||
original = path.read_bytes()
|
||||
newline = '\r\n' if b'\r\n' in original else '\n'
|
||||
content = json.dumps(manifest, ensure_ascii=False, indent=4) + '\n'
|
||||
temporary = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(dir=path.parent, suffix='.tmp', delete=False) as output:
|
||||
temporary = Path(output.name)
|
||||
output.write(content.replace('\n', newline).encode('utf-8'))
|
||||
temporary.replace(path)
|
||||
finally:
|
||||
if temporary is not None:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
|
||||
def update_manifest(path, write=False):
|
||||
manifest = json.loads(path.read_text(encoding='utf-8-sig'))
|
||||
checkver, autoupdate = manifest.get('checkver'), manifest.get('autoupdate')
|
||||
if not checkver or not autoupdate:
|
||||
print(f'{path.stem}: kein automatisches Update konfiguriert; übersprungen.')
|
||||
return False
|
||||
if not isinstance(checkver, dict) or not isinstance(autoupdate.get('url'), str):
|
||||
raise ValueError('Diese checkver/autoupdate-Konfiguration wird noch nicht unterstützt.')
|
||||
|
||||
template = autoupdate['url']
|
||||
observed = None
|
||||
if checkver.get('regex') == TIMESTAMP_REGEX:
|
||||
source = urldefrag(manifest['url'])[0]
|
||||
print(f'{path.stem}: HEAD-Anfrage an {source} ...', flush=True)
|
||||
observed = read_head(source)
|
||||
if not observed['Last-Modified']:
|
||||
raise ValueError('Last-Modified-Header fehlt.')
|
||||
date = parsedate_to_datetime(observed['Last-Modified'])
|
||||
if date.tzinfo is None:
|
||||
raise ValueError('Last-Modified enthält keine Zeitzone.')
|
||||
version = date.astimezone(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
|
||||
if urldefrag(template.replace('$version', version))[0] != source:
|
||||
raise ValueError('Zeitstempel-Updates benötigen eine feste Download-URL.')
|
||||
elif 'script' not in checkver:
|
||||
print(f'{path.stem}: Release-Version abfragen: {checkver.get("url", "")} ...', flush=True)
|
||||
version = release_version(checkver)
|
||||
else:
|
||||
raise ValueError('Dieses checkver-Skript wird nicht ausgeführt oder interpretiert.')
|
||||
|
||||
if version == manifest['version']:
|
||||
print(f'{path.stem}: {version} unverändert; kein Download.')
|
||||
return False
|
||||
url = template.replace('$version', version)
|
||||
if '$' in url:
|
||||
raise ValueError('Unterstützt wird nur der Platzhalter $version.')
|
||||
source = urldefrag(url)[0]
|
||||
if observed is None:
|
||||
print(f'{path.stem}: HEAD-Anfrage an {source} ...', flush=True)
|
||||
observed = read_head(source)
|
||||
print(f'{path.stem}: neue Version {version}; GET-Download ...', flush=True)
|
||||
print(' Für den SHA-256-Hash muss die gesamte Datei heruntergeladen werden. '
|
||||
'Danach wird der Hash berechnet. Bei grossen Dateien kann dies mehrere Minuten dauern, '
|
||||
'auch im Probelauf.', flush=True)
|
||||
digest, size = download_hash(source, observed)
|
||||
print(f'{path.stem}: {manifest["version"]} -> {version}; {size} Bytes; SHA-256 {digest}')
|
||||
manifest.update(version=version, url=url, hash=digest)
|
||||
if write:
|
||||
save_manifest(path, manifest)
|
||||
print(f'{path.stem}: Manifest gespeichert.')
|
||||
else:
|
||||
print(f'{path.stem}: Probelauf; Manifest bleibt unverändert.')
|
||||
return True
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--bucket', type=Path, default=BUCKET)
|
||||
parser.add_argument('--app', action='append', help='Manifestname; mehrfach möglich, sonst alle.')
|
||||
parser.add_argument('--write', action='store_true', help='Geprüfte Änderungen speichern.')
|
||||
args = parser.parse_args(argv)
|
||||
if not args.bucket.is_dir():
|
||||
parser.error('Bucket-Verzeichnis existiert nicht.')
|
||||
if args.app and any(not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]*', app) for app in args.app):
|
||||
parser.error('--app erwartet einen Manifestnamen ohne Pfad oder .json-Endung.')
|
||||
paths = ([args.bucket / f'{app}.json' for app in args.app] if args.app else
|
||||
sorted(args.bucket.glob('*.json')))
|
||||
errors = 0
|
||||
for path in paths:
|
||||
try:
|
||||
update_manifest(path, args.write)
|
||||
except HTTPError as error:
|
||||
print(f'{path.stem}: HTTP {error.code}; Manifest unverändert.', file=sys.stderr)
|
||||
errors += 1
|
||||
except (ValueError, KeyError, TypeError, OSError, URLError) as error:
|
||||
print(f'{path.stem}: {error}; Manifest unverändert.', file=sys.stderr)
|
||||
errors += 1
|
||||
return int(errors > 0)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user