Abruf der Frontier-API (Texte aller Sprachen + Bilder) nach galnet_raw/. Läuft bei Oliver, auch für das monatliche Update.
Ablage beim Arbeiten: galnet_fetch.py. Herunterladen: auf den Dateinamen über dem Code klicken.
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ galnet_fetch.py - Galnet vollstaendig von der Frontier-API sichern. Speichert die rohe API-Antwort pro Sprache und Monat: galnet_raw/EN/3302-01.json galnet_raw/DE/3302-01.json ... und baut daraus galnet_raw/index.csv (ein Eintrag pro Artikel, alle Sprachen): frontier_id;date;date_ingame;number;page_id;languages;title_en;slug_en date = Echtzeit (Spieljahr - 1286) number = 01, 02 ... innerhalb eines Tages, nach Frontier-ID aufsteigend page_id = news:galnet:JJJJ:MM:TT_NN (Echtzeit-Datum) Verhalten: * Erster Lauf: holt alles (Standard 3300-01 bis zum aktuellen Spielmonat). * Weitere Laeufe: holt nur Monate, die fehlen oder beim letzten Mal gescheitert sind, plus die letzten --recent Monate (Standard 2) neu. -> fuer das monatliche Aktualisieren einfach ohne Optionen aufrufen. * Aenderungen gegenueber dem letzten Stand (neue / geaenderte / entfernte Artikel) werden angezeigt und an galnet_raw/changes.log angehaengt. * Gescheiterte Monate werden am Ende aufgelistet, Rueckgabewert 1. Einfach erneut starten - sie werden dann nachgeholt. Aufruf: python3 galnet_fetch.py # erster Lauf / monatliches Update python3 galnet_fetch.py --force # alles neu holen python3 galnet_fetch.py --lang EN DE # nur bestimmte Sprachen python3 galnet_fetch.py --index-only # nur index.csv neu bauen python3 galnet_fetch.py --images-only # nur fehlende Bilder laden (galnet_raw/images/) """ import argparse, base64, csv, fcntl, hashlib, json, os, re, sys, time, urllib.parse, urllib.request from collections import defaultdict from datetime import datetime, timedelta, timezone from pathlib import Path API_BASE = 'https://www.elitedangerous.com/elite-webstore-proxy/v1/galnet/articles' UA = 'Mozilla/5.0 (X11; Linux x86_64; rv:154.0) Gecko/20100101 Firefox/154.0' LANGUAGES = [('EN', 'en'), ('EN-US', 'en-US'), ('DE', 'de-DE'), ('ES', 'es-ES'), ('FR', 'fr-FR'), ('PT-BR', 'pt-BR'), ('RU', 'ru-RU')] YEAR_OFFSET = 1286 TIMEOUT, RETRIES, DELAY = 30, 4, 0.3 OUT_DIR = Path(__file__).resolve().parent / 'galnet_raw' # ---------------------------------------------------------------- Abruf def fetch(url, language): headers = {'User-Agent': UA, 'Accept': 'application/json', 'Accept-Language': language} last = None for n in range(RETRIES): try: with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=TIMEOUT) as r: raw = r.read() payload = json.loads(raw.decode('utf-8')) articles_of(payload) # Format pruefen, sonst Wiederholung return payload except Exception as e: last = e if n + 1 < RETRIES: time.sleep(2 * (n + 1)) raise last def month_url(year, month): start = datetime(year, month, 1) end = datetime(year + (month == 12), month % 12 + 1, 1) - timedelta(days=1) return API_BASE + '?' + urllib.parse.urlencode({ 'date_from': start.strftime('%Y-%m-%d'), 'date_to': end.strftime('%Y-%m-%d')}) def articles_of(payload): if isinstance(payload, list): return payload if isinstance(payload, dict) and isinstance(payload.get('articles'), list): return payload['articles'] raise RuntimeError('Unbekanntes API-Antwortformat') def write_json(path, payload): path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(f'{path.name}.{os.getpid()}.part') tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding='utf-8') tmp.replace(path) def read_articles(path): try: return articles_of(json.loads(path.read_text(encoding='utf-8'))) except Exception: return None # ---------------------------------------------------------------- Hilfen def first(a, *keys): for k in keys: if a.get(k): return str(a[k]) return '' def art_id(a): return first(a, 'id', 'uuid', 'article_id') def fingerprint(a): return hashlib.sha1(json.dumps(a, sort_keys=True, ensure_ascii=False).encode()).hexdigest() def ingame_date(a): m = re.match(r'(\d{4})-(\d{2})-(\d{2})', first(a, 'dated_at')) return tuple(map(int, m.groups())) if m else None def current_ingame_month(): now = datetime.now(timezone.utc) return now.year + YEAR_OFFSET, now.month def month_list(start, end): (sy, sm), (ey, em) = start, end return [(y, m) for y in range(sy, ey + 1) for m in range(1, 13) if (sy, sm) <= (y, m) <= (ey, em)] def compare(old, new): """Liste von (art, id, titel) fuer neu / geaendert / entfernt.""" o = {art_id(a): a for a in (old or [])} n = {art_id(a): a for a in new} out = [] for i, a in n.items(): if i not in o: out.append(('NEU', i, first(a, 'title'))) elif fingerprint(a) != fingerprint(o[i]): out.append(('GEAENDERT', i, first(a, 'title'))) for i, a in o.items(): if i not in n: out.append(('ENTFERNT', i, first(a, 'title'))) return out # ---------------------------------------------------------------- Index def build_index(out_dir, languages): entries = {} langs = defaultdict(set) for lang, _ in languages: for f in sorted((out_dir / lang).glob('*.json')): for a in read_articles(f) or []: i = art_id(a) if not i: continue langs[i].add(lang) # EN bestimmt Datum/Titel; sonst erste Sprache, die ihn hat if i not in entries or lang == 'EN': entries[i] = a by_day = defaultdict(list) no_date = [] for i, a in entries.items(): d = ingame_date(a) (by_day[d] if d else no_date).append(i) rows = [] order = [l for l, _ in languages] for d in sorted(by_day): y, m, day = d for n, i in enumerate(sorted(by_day[d]), 1): # Frontier-ID aufsteigend a = entries[i] rows.append({ 'frontier_id': i, 'date': f'{y - YEAR_OFFSET:04d}-{m:02d}-{day:02d}', 'date_ingame': f'{y:04d}-{m:02d}-{day:02d}', 'number': f'{n:02d}', 'page_id': f'news:galnet:{y - YEAR_OFFSET:04d}:{m:02d}:{day:02d}_{n:02d}', 'languages': ' '.join(l for l in order if l in langs[i]), 'title_en': re.sub(r'\s+', ' ', first(a, 'title')).strip(), 'slug_en': first(a, 'slug'), }) fields = ['frontier_id', 'date', 'date_ingame', 'number', 'page_id', 'languages', 'title_en', 'slug_en'] tmp = out_dir / 'index.csv.part' with tmp.open('w', encoding='utf-8', newline='') as f: w = csv.DictWriter(f, fieldnames=fields, delimiter=';') w.writeheader() w.writerows(rows) tmp.replace(out_dir / 'index.csv') print(f'\nindex.csv: {len(rows)} Artikel') per_lang = defaultdict(int) for s in langs.values(): for l in s: per_lang[l] += 1 print(' pro Sprache: ' + ', '.join(f'{l} {per_lang[l]}' for l in order)) if no_date: print(f' WARNUNG: {len(no_date)} Artikel ohne gueltiges dated_at: {", ".join(no_date[:10])}') # ---------------------------------------------------------------- Bilder def image_url(path): """Frontier-Bilddienst: Original-Pfad aus der API -> WebP in voller Breite.""" spec = {'bucket': 'frontier-platform-hosting', 'key': path.lstrip('/'), 'edits': {'webp': {'quality': 85}, 'toFormat': 'webp', 'resize': {'width': 3840, 'fit': 'contain'}}} token = base64.urlsafe_b64encode(json.dumps(spec, separators=(',', ':')).encode()).decode() return 'https://image-service.zaonce.net/' + token def fetch_images(out_dir): """Laedt alle in den API-Daten genannten Bilder, die noch fehlen, nach galnet_raw/images/.""" paths = set() for f in out_dir.glob('*/*.json'): for a in read_articles(f) or []: for img in a.get('images') or []: if img.get('path') and Path(img['path']).stem: paths.add(img['path']) img_dir = out_dir / 'images' img_dir.mkdir(exist_ok=True) todo = [p for p in sorted(paths) if not (img_dir / (Path(p).stem + '.webp')).exists()] print(f'\nBilder: {len(paths)} bekannt, {len(todo)} fehlen') failed = [] for n, p in enumerate(todo, 1): target = img_dir / (Path(p).stem + '.webp') try: req = urllib.request.Request(image_url(p), headers={'User-Agent': UA, 'Referer': 'https://www.elitedangerous.com/'}) with urllib.request.urlopen(req, timeout=TIMEOUT) as r: data = r.read() ctype = r.headers.get('Content-Type', '') if not data or 'html' in ctype: raise RuntimeError(f'ungueltige Antwort ({ctype})') tmp = target.with_name(f'{target.name}.{os.getpid()}.part') tmp.write_bytes(data) tmp.replace(target) print(f' [{n}/{len(todo)}] {target.name} ({len(data) // 1024} KB)') except Exception as e: print(f' [{n}/{len(todo)}] {Path(p).name}: FEHLER {e}') failed.append(f'Bild {Path(p).name}') time.sleep(DELAY) return failed # ---------------------------------------------------------------- Hauptteil def main(): p = argparse.ArgumentParser(description='Galnet vollstaendig von der Frontier-API sichern') p.add_argument('--from', dest='start', default='3300-01', help='Start JJJJ-MM (Spielzeit)') p.add_argument('--to', dest='end', help='Ende JJJJ-MM (Standard: aktueller Spielmonat)') p.add_argument('--recent', type=int, default=2, help='so viele letzte Monate immer neu holen') p.add_argument('--force', action='store_true', help='alle Monate neu holen') p.add_argument('--lang', nargs='+', help='nur diese Sprachen (z. B. EN DE)') p.add_argument('--out', default=str(OUT_DIR), help='Zielordner') p.add_argument('--index-only', action='store_true', help='nur index.csv neu bauen') p.add_argument('--no-images', action='store_true', help='keine Bilder laden') p.add_argument('--images-only', action='store_true', help='nur fehlende Bilder laden') args = p.parse_args() out_dir = Path(args.out) out_dir.mkdir(parents=True, exist_ok=True) languages = [(l, c) for l, c in LANGUAGES if not args.lang or l in [x.upper() for x in args.lang]] # Sperre: nur ein Lauf gleichzeitig lock = open(out_dir / '.lock', 'w') try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: print('Es laeuft bereits ein anderer galnet_fetch.py - bitte warten, bis er fertig ist.') return 2 # Reste abgebrochener Laeufe entfernen for leftover in out_dir.glob('*/*.part'): leftover.unlink(missing_ok=True) if args.index_only: build_index(out_dir, LANGUAGES) return 0 if args.images_only: return 1 if fetch_images(out_dir) else 0 start = tuple(map(int, args.start.split('-'))) end = tuple(map(int, args.end.split('-'))) if args.end else current_ingame_month() months = month_list(start, end) recent = set(months[-args.recent:]) if args.recent > 0 else set() print('=' * 60) print('GALNET FETCH') print('=' * 60) print(f'Ziel : {out_dir}') print(f'Zeitraum: {start[0]}-{start[1]:02d} bis {end[0]}-{end[1]:02d} ({len(months)} Monate)') print(f'Sprachen: {" ".join(l for l, _ in languages)}') failed, changes = [], [] fetched = skipped = 0 stamp = datetime.now().strftime('%Y-%m-%d %H:%M') try: for lang, code in languages: print(f'\n--- {lang} ---') for (y, m) in months: label = f'{y}-{m:02d}' path = out_dir / lang / f'{label}.json' old = read_articles(path) if path.exists() else None if old is not None and not args.force and (y, m) not in recent: skipped += 1 continue try: payload = fetch(month_url(y, m), code) except Exception as e: print(f' {label}: FEHLER {e}') failed.append(f'{lang} {label}') continue new = articles_of(payload) write_json(path, payload) fetched += 1 diff = compare(old, new) if old is not None else [] for kind, i, t in diff: changes.append(f'{stamp}\t{lang}\t{label}\t{kind}\t{i}\t{t}') note = f' | {len(diff)} Aenderung(en)' if diff else '' if old is None: note = ' | neu gesichert' print(f' {label}: {len(new)} Artikel{note}') time.sleep(DELAY) except KeyboardInterrupt: print('\nABGEBROCHEN - bereits gesicherte Monate bleiben erhalten.') if changes: with (out_dir / 'changes.log').open('a', encoding='utf-8') as f: f.write('\n'.join(changes) + '\n') build_index(out_dir, LANGUAGES) if not args.no_images: failed += fetch_images(out_dir) print('\n' + '=' * 60) print(f'Abgerufen : {fetched} Monatsdateien') print(f'Unveraendert: {skipped} (schon vorhanden)') if changes: print(f'Aenderungen : {len(changes)} (siehe changes.log)') for c in changes[:30]: print(' ' + c.split('\t', 1)[1].replace('\t', ' ')) if len(changes) > 30: print(f' ... und {len(changes) - 30} weitere') if failed: print(f'FEHLGESCHLAGEN ({len(failed)}): {", ".join(failed)}') print('-> Skript einfach erneut starten, diese Monate werden nachgeholt.') return 1 print('Alles vollstaendig.') return 0 if __name__ == '__main__': sys.exit(main())