73 lines
2.9 KiB
Python
73 lines
2.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Laedt Dateien ueber die Gitea-Contents-API hoch (create oder update)."""
|
|
import base64, json, os, sys, urllib.request, urllib.error
|
|
|
|
def env():
|
|
d = {}
|
|
for line in open('.env', encoding='utf-8'):
|
|
line = line.strip()
|
|
if not line or line.startswith('#') or '=' not in line: continue
|
|
k, v = line.split('=', 1)
|
|
v = v.strip()
|
|
if v.startswith('"') and v.endswith('"'): v = v[1:-1].replace('\\$', '$').replace('\\"', '"')
|
|
d[k] = v
|
|
return d
|
|
|
|
E = env()
|
|
BASE = E['GITEA_BASE_URL'].rstrip('/')
|
|
REPO = f"{BASE}/api/v1/repos/{E['GITEA_OWNER']}/{E['GITEA_REPO']}/contents"
|
|
KOPF = {'Authorization': 'token ' + E['GITEA_TOKEN'], 'Content-Type': 'application/json'}
|
|
|
|
def ruf(methode, url, daten=None):
|
|
r = urllib.request.Request(url, method=methode, headers=KOPF,
|
|
data=json.dumps(daten).encode() if daten else None)
|
|
try:
|
|
with urllib.request.urlopen(r, timeout=60) as a:
|
|
return a.status, json.loads(a.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read().decode() or '{}')
|
|
|
|
def hoch(pfad, nachricht):
|
|
with open(pfad, 'rb') as f:
|
|
inhalt = base64.b64encode(f.read()).decode()
|
|
ziel = pfad.replace(os.sep, '/')
|
|
code, _ = ruf('GET', f"{REPO}/{ziel}?ref={E.get('branch','main')}")
|
|
sha = None
|
|
if code == 200:
|
|
code_g, body = ruf('GET', f"{REPO}/{ziel}?ref=main")
|
|
sha = body.get('sha')
|
|
daten = {'content': inhalt, 'message': nachricht, 'branch': 'main'}
|
|
if sha:
|
|
daten['sha'] = sha
|
|
c, b = ruf('PUT', f"{REPO}/{ziel}", daten)
|
|
else:
|
|
c, b = ruf('POST', f"{REPO}/{ziel}", daten)
|
|
ok = c in (200, 201)
|
|
print(f" {'ok ' if ok else 'FEHLER'} {ziel}" + ('' if ok else f" -> {c} {str(b)[:160]}"))
|
|
return ok
|
|
|
|
AUS = {'.git', '.env', 'node_modules', '__pycache__'}
|
|
|
|
# Diese Pfade schreiben die Workflows selbst. Wuerde das Skript sie hochladen,
|
|
# ueberschreibt ein lokaler Altstand die Ergebnisse eines Laufs.
|
|
NICHT_UEBERSCHREIBEN = {'artikel-index.json', 'status-importe.json'}
|
|
NICHT_UEBERSCHREIBEN_ORDNER = ('briefe' + os.sep, 'nachbessern' + os.sep, 'artikel' + os.sep)
|
|
dateien = []
|
|
for wurzel, ordner, namen in os.walk('.'):
|
|
ordner[:] = [o for o in ordner if o not in AUS]
|
|
for n in namen:
|
|
p = os.path.relpath(os.path.join(wurzel, n), '.')
|
|
if p in AUS or p.startswith('.env') and p != '.env.example': continue
|
|
if p == '.env': continue
|
|
if p in NICHT_UEBERSCHREIBEN or p.startswith(NICHT_UEBERSCHREIBEN_ORDNER):
|
|
continue
|
|
dateien.append(p)
|
|
|
|
nachricht = sys.argv[1] if len(sys.argv) > 1 else 'Update'
|
|
print("Uebersprungen (von den Workflows gepflegt): "
|
|
+ ', '.join(sorted(NICHT_UEBERSCHREIBEN)) + ', ' + ', '.join(NICHT_UEBERSCHREIBEN_ORDNER))
|
|
fehler = 0
|
|
for p in sorted(dateien):
|
|
if not hoch(p, nachricht): fehler += 1
|
|
print(f"\n{len(dateien)} Dateien, {fehler} Fehler")
|