Helper script for downgrading MSYS2

msys2_downgrade.py

#!/usr/bin/env python3

from argparse import ArgumentParser
from datetime import datetime, timedelta
from pathlib import Path
import json
import re
import subprocess
import sys
import tempfile
import urllib.request

PACKAGE_CACHE_FILE = Path(tempfile.gettempdir()).joinpath('msys2_downgrade.json')

PACKAGE_LIST_URLS = [
  'http://repo.msys2.org/msys/x86_64',
  'http://repo.msys2.org/mingw/i686',
  'http://repo.msys2.org/mingw/x86_64',
]

LIST_START_REGEX = re.compile('^.*<pre>')
LIST_END_REGEX   = re.compile('^.*</pre>')
LIST_ENTRY_REGEX = re.compile(r'''
  ^ \s* <a\ href=" (?P<name> [^"]+ ) ">[^<]+</a>
    \s+            (?P<date>   \S+ )
    \s+            (?P<time>   \S+ )
    \s+            (?P<size>   \S+ ) \s* $''', re.VERBOSE)

PACKAGE_FILE_REGEX = re.compile(r'''
  ^ (?P<package>                 .+? )
  - (?P<version>         [^-]+ - \d+ )
  - (?P<arch>    i686 | x86_64 | any ) \.pkg\.tar\.xz $''', re.VERBOSE)

def warn(*args, **kwargs):
  print(*args, **kwargs, file=sys.stderr)

def query_installed_packages():
  installed = {}

  output = subprocess.check_output(['pacman', '-Q'])
  for line in output.decode().splitlines():
    package, version = line.split(maxsplit=1)
    installed[package] = version

  return installed

def fetch_available_versions(cache, url):
  entries = []

  cached = cache.get(url)
  if cached:
    timestamp = datetime.fromisoformat(cached['timestamp'])
    if timestamp > datetime.now() - timedelta(hours=12):
      warn('info: using cached data for %s' % url)
      entries = cached['entries']

  if not entries:
    warn('info: fetching data from %s' % url)
    with urllib.request.urlopen(url) as fh:
      in_list = False
      for line in fh:
        line = line.decode()

        if not in_list:
          if LIST_START_REGEX.match(line):
            in_list = True
          continue

        if LIST_END_REGEX.match(line):
          break

        match = LIST_ENTRY_REGEX.match(line)
        if not match:
          warn('error: failed to parse line: %r' % line)
          continue

        filename = match['name']

        if not filename.endswith('.pkg.tar.xz'):
          continue

        timestamp = '%s %s' % (match['date'], match['time'])
        timestamp = datetime.strptime(timestamp, '%d-%b-%Y %H:%M')
        timestamp = timestamp.isoformat()

        entries.append([filename, timestamp])

    entries.sort(key=lambda entry: entry[1])

  available = {}

  for filename, timestamp in entries:
    match = PACKAGE_FILE_REGEX.match(filename)
    if not match:
      warn('error: failed to parse name: %r' % filename)
      continue

    package = match['package']
    version = match['version']
    download = '%s/%s' % (url, filename)
    timestamp = datetime.fromisoformat(timestamp)

    available.setdefault(package, []).append({
      'version': version,
      'download': download,
      'timestamp': timestamp,
    })

  cache[url] = {
    'timestamp': datetime.now().isoformat(),
    'entries': entries,
  }

  return available

def get_package_info():
  cache = {}

  if PACKAGE_CACHE_FILE.exists():
    warn('info: loading cache file %s' % PACKAGE_CACHE_FILE)
    with PACKAGE_CACHE_FILE.open('r') as fh:
      cache = json.load(fh)

  packages = {}

  installed = query_installed_packages()
  for package, version in installed.items():
    packages[package] = {'installed': version, 'available': []}

  for url in PACKAGE_LIST_URLS:
    available = fetch_available_versions(cache, url)
    for package, info in packages.items():
      versions = available.get(package)
      if versions:
        info['available'] = versions

  with PACKAGE_CACHE_FILE.open('w') as fh:
    json.dump(cache, fh)

  return packages

def print_package_info(package, info):
  print('===', package)
  print(' installed:', info['installed'])

  if not info['available']:
    print(' available: None')
    return

  first = True
  for available in info['available']:
    label = ' available:' if first else \
            '           '
    first = False
    print('%s %s (%s)' % (label, available['version'], available['timestamp']))

def show_package(package):
  packages = get_package_info()

  info = packages.get(package)
  if not info:
    warn('error: unknown package %r' % package)
    return

  print_package_info(package, info)

def search_packages(search):
  packages = get_package_info()
  matches = {}

  for package, info in packages.items():
    if search in package:
      matches[package] = info

  for package, info in sorted(matches.items()):
    print_package_info(package, info)

def print_package_list(packages, save_urls):
  urls = []

  for package, latest in sorted(packages.items()):
    urls.append(latest['download'])
    print('%s %s (%s)' % (package, latest['version'], latest['timestamp']))

  if save_urls:
    with save_urls.open('w', newline='\n') as fh:
      for url in urls:
        fh.write('%s\n' % url)

def list_upgraded_since(timestamp, save_urls):
  timestamp = datetime.fromisoformat(timestamp)
  packages = get_package_info()
  matches = {}

  for package, info in packages.items():
    if not info['available']:
      warn('warning: no versions available for %r' % package)
      continue

    latest = info['available'][-1]
    if latest['timestamp'] >= timestamp:
      matches[package] = latest

  print_package_list(matches, save_urls)

def list_downgrade_to(timestamp, save_urls):
  timestamp = datetime.fromisoformat(timestamp)
  packages = get_package_info()
  matches = {}

  for package, info in packages.items():
    if not info['available']:
      warn('warning: no versions available for %r' % package)
      continue

    latest = None
    for available in reversed(info['available']):
      if available['timestamp'] < timestamp:
        latest = available
        break

    if not latest:
      warn('warning: no suitable version for %r' % package)
      continue

    if info['installed'] != latest['version']:
      matches[package] = latest

  print_package_list(matches, save_urls)

def main():
  parser = ArgumentParser(description='Helpers for downgrading MSYS2 packages')
  parser.add_argument('--show', metavar='PACKAGE', help='print info for the specified package')
  parser.add_argument('--search', metavar='PACKAGE', help='print info for matching packages')
  parser.add_argument('--upgraded-since', metavar='DATE', help='list packages upgraded since date')
  parser.add_argument('--downgrade-to', metavar='DATE', help='list packages needed for downgrade to date')
  parser.add_argument('--save-urls', metavar='FILE', type=Path, help='save URLs for upgrade/downgrade to file')
  args = parser.parse_args()

  if args.show:
    return show_package(args.show)

  if args.search:
    return search_packages(args.search)

  if args.upgraded_since:
    return list_upgraded_since(args.upgraded_since, args.save_urls)

  if args.downgrade_to:
    return list_downgrade_to(args.downgrade_to, args.save_urls)

  parser.print_usage()

main()