#!/usr/bin/env python3 """Synchronize the published Workshop version, or inspect a local Workbench bundle.""" import argparse import json from pathlib import Path import re import sys import urllib.request from html.parser import HTMLParser ROOT = Path(__file__).resolve().parents[1] VERSION_LINE = 'Latest published Workshop version:' DEFAULT_MANIFEST = (Path.home() / '.local/share/Steam/steamapps/compatdata/1874910/pfx' / 'drive_c/users/steamuser/AppData/Local/Temp/Arma Reforger Workbench' / 'Publishing/2528466DE220D91C/manifest.json') def valid_version(value): return (isinstance(value, str) and re.fullmatch(r'\d+\.\d+\.\d+', value) is not None and all(int(part) <= 32000 for part in value.split('.'))) def bundled_version(manifest, project): guid = re.search(r'\bGUID\s+"([A-Fa-f0-9]{16})"', project) asset = manifest.get('asset', {}) if isinstance(manifest, dict) else {} if (not guid or not isinstance(asset, dict) or not isinstance(asset.get('id'), str) or asset['id'].upper() != guid[1].upper()): raise ValueError('Publishing manifest does not belong to this addon.') version = asset.get('version') # Top-level "version" is the manifest schema, not the mod version. if not valid_version(version): raise ValueError('Publishing manifest has no valid addon version.') return version class WorkshopPage(HTMLParser): def __init__(self): super().__init__() self.active = False self.payloads = [] def handle_starttag(self, tag, attrs): if tag == 'script' and dict(attrs).get('id') == '__NEXT_DATA__': self.active = True self.payloads.append('') def handle_endtag(self, tag): if tag == 'script': self.active = False def handle_data(self, data): if self.active: self.payloads[-1] += data def online_version(page, guid): parser = WorkshopPage() parser.feed(page) if len(parser.payloads) != 1: raise ValueError('Workshop page has no unique version metadata.') data = json.loads(parser.payloads[0]) asset = data['props']['pageProps']['asset'] if asset.get('id') != guid or asset.get('private') is True or asset.get('blocked') is True: raise ValueError('Workshop page does not identify the public addon.') version = asset.get('currentVersionNumber') if not valid_version(version) or not any(v.get('version') == version and v.get('assetId') == guid and v.get('published') is True and v.get('approved') is True for v in asset.get('versions', [])): raise ValueError('Workshop current version is not confirmed as published and approved.') return version def prevent_downgrade(text, version): existing = re.search(r'^Latest published Workshop version: \*\*(\d+\.\d+\.\d+)\*\*$', text, re.MULTILINE) if existing and tuple(map(int, version.split('.'))) < tuple(map(int, existing[1].split('.'))): raise ValueError('Workshop response is older than the recorded publication; refusing stale metadata.') def sync_version(project, text, native_compat=None): match = re.search(r'\bGUID\s+"([A-Fa-f0-9]{16})"', project) if not match: raise ValueError('Addon GUID is missing.') guid = match[1].upper() if native_compat: from workshop_query import native_version version = native_version(ROOT, native_compat, guid) else: request = urllib.request.Request('https://reforger.armaplatform.com/workshop/' + guid, headers={'Cache-Control': 'no-cache'}) with urllib.request.urlopen(request, timeout=20) as response: page = response.read(2 * 1024 * 1024 + 1) if len(page) > 2 * 1024 * 1024: raise ValueError('Workshop response exceeds the expected size.') version = online_version(page.decode('utf-8'), guid) if not valid_version(version): raise ValueError('Workshop returned an invalid version.') prevent_downgrade(text, version) return version def update_changelog(text, version): if not valid_version(version): raise ValueError('Invalid Workshop version.') if not text.startswith('# Changelog\n'): raise ValueError('Expected # Changelog at the start of CHANGELOG.md.') line = VERSION_LINE + ' **' + version + '**' existing = list(re.finditer(r'^Latest published Workshop version:.*$', text, re.MULTILINE)) if len(existing) > 1: raise ValueError('Changelog contains multiple Workshop version records.') if existing: match = existing[0] if match.start() > text.find('\n## ') >= 0: raise ValueError('Workshop version record must be above the changelog entries.') return text[:match.start()] + line + text[match.end():] return text.replace('# Changelog\n', '# Changelog\n\n' + line + '\n', 1) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--manifest', type=Path, default=DEFAULT_MANIFEST) action = parser.add_mutually_exclusive_group() action.add_argument('--sync', action='store_true', help='Fetch the live published version and update CHANGELOG.md') parser.add_argument('--native-compat', type=Path, help='Use the live game catalogue with this disposable prefix instead of the website') action.add_argument('--published', action='store_true', help='Confirm the bundle was successfully uploaded; update CHANGELOG.md') args = parser.parse_args() if args.native_compat and not args.sync: parser.error('--native-compat requires --sync') try: if args.sync: path = ROOT / 'CHANGELOG.md' before = path.read_text() version = sync_version((ROOT / 'addons/core/addon.gproj').read_text(), before, args.native_compat) after = update_changelog(before, version) if before != after: path.write_text(after) print('Verified published Workshop version ' + version + '; CHANGELOG.md synchronized.') return 0 manifest = json.loads(args.manifest.read_text(encoding='utf-8-sig')) version = bundled_version(manifest, (ROOT / 'addons/core/addon.gproj').read_text()) if args.published: path = ROOT / 'CHANGELOG.md' before = path.read_text() after = update_changelog(before, version) if before != after: path.write_text(after) print('Recorded published Workshop version ' + version + ' in CHANGELOG.md.') else: print('Workbench bundle version: ' + version) print('Bundling alone does not confirm publication. Use --published after a successful upload.') except (OSError, ValueError, AttributeError, KeyError, TypeError) as error: print('Cannot read/record Workshop version: ' + str(error), file=sys.stderr) return 1 return 0 if __name__ == '__main__': sys.exit(main())