pkgs/by-name/fa/factorio/update.py

   1 #!/usr/bin/env nix-shell
   2 #! nix-shell -i python -p "python3.withPackages (ps: with ps; [ ps.absl-py ps.requests ])"
   3
   4 from collections import defaultdict
   5 import copy
   6 from dataclasses import dataclass
   7 import json
   8 import os.path
   9 from typing import Callable, Dict
  10
  11 from absl import app
  12 from absl import flags
  13 from absl import logging
  14 import requests
  15
  16
  17 FACTORIO_RELEASES = "https://factorio.com/api/latest-releases"
  18 FACTORIO_HASHES = "https://factorio.com/download/sha256sums/"
  19
  20
  21 FLAGS = flags.FLAGS
  22
  23 flags.DEFINE_string("out", "", "Output path for versions.json.")
  24 flags.DEFINE_list(
  25     "release_type",
  26     "",
  27     "If non-empty, a comma-separated list of release types to update (e.g. alpha).",
  28 )
  29 flags.DEFINE_list(
  30     "release_channel",
  31     "",
  32     "If non-empty, a comma-separated list of release channels to update (e.g. experimental).",
  33 )
  34
  35
  36 @dataclass
  37 class System:
  38     nix_name: str
  39     url_name: str
  40     tar_name: str
  41
  42
  43 @dataclass
  44 class ReleaseType:
  45     name: str
  46     hash_filename_format: list[str]
  47     needs_auth: bool = False
  48
  49
  50 @dataclass
  51 class ReleaseChannel:
  52     name: str
  53
  54
  55 FactorioVersionsJSON = Dict[str, Dict[str, str]]
  56 OurVersionJSON = Dict[str, Dict[str, Dict[str, Dict[str, str]]]]
  57
  58 FactorioHashes = Dict[str, str]
  59
  60
  61 SYSTEMS = [
  62     System(nix_name="x86_64-linux", url_name="linux64", tar_name="x64"),
  63 ]
  64
  65 RELEASE_TYPES = [
  66     ReleaseType(
  67         "alpha",
  68         needs_auth=True,
  69         hash_filename_format=["factorio_linux_{version}.tar.xz"],
  70     ),
  71     ReleaseType("demo", hash_filename_format=["factorio_demo_x64_{version}.tar.xz"]),
  72     ReleaseType(
  73         "headless",
  74         hash_filename_format=[
  75             "factorio-headless_linux_{version}.tar.xz",
  76             "factorio_headless_x64_{version}.tar.xz",
  77         ],
  78     ),
  79     ReleaseType(
  80         "expansion",
  81         needs_auth=True,
  82         hash_filename_format=["factorio-space-age_linux_{version}.tar.xz"],
  83     ),
  84 ]
  85
  86 RELEASE_CHANNELS = [
  87     ReleaseChannel("experimental"),
  88     ReleaseChannel("stable"),
  89 ]
  90
  91
  92 def find_versions_json() -> str:
  93     if FLAGS.out:
  94         return FLAGS.out
  95     try_paths = ["pkgs/by-name/fa/factorio/versions.json", "versions.json"]
  96     for path in try_paths:
  97         if os.path.exists(path):
  98             return path
  99     raise Exception(
 100         "Couldn't figure out where to write versions.json; try specifying --out"
 101     )
 102
 103
 104 def fetch_versions() -> FactorioVersionsJSON:
 105     return json.loads(requests.get(FACTORIO_RELEASES).text)
 106
 107
 108 def fetch_hashes() -> FactorioHashes:
 109     resp = requests.get(FACTORIO_HASHES)
 110     resp.raise_for_status()
 111     out = {}
 112     for ln in resp.text.split("\n"):
 113         ln = ln.strip()
 114         if not ln:
 115             continue
 116         sha256, filename = ln.split()
 117         out[filename] = sha256
 118     return out
 119
 120
 121 def generate_our_versions(factorio_versions: FactorioVersionsJSON) -> OurVersionJSON:
 122     def rec_dd():
 123         return defaultdict(rec_dd)
 124
 125     output = rec_dd()
 126
 127     # Deal with times where there's no experimental version
 128     for rc in RELEASE_CHANNELS:
 129         if rc.name not in factorio_versions or not factorio_versions[rc.name]:
 130             factorio_versions[rc.name] = factorio_versions["stable"]
 131         for rt in RELEASE_TYPES:
 132             if (
 133                 rt.name not in factorio_versions[rc.name]
 134                 or not factorio_versions[rc.name][rt.name]
 135             ):
 136                 factorio_versions[rc.name][rt.name] = factorio_versions["stable"][
 137                     rt.name
 138                 ]
 139
 140     for system in SYSTEMS:
 141         for release_type in RELEASE_TYPES:
 142             for release_channel in RELEASE_CHANNELS:
 143                 version = factorio_versions[release_channel.name].get(release_type.name)
 144                 if version is None:
 145                     continue
 146                 this_release = {
 147                     "name": f"factorio_{release_type.name}_{system.tar_name}-{version}.tar.xz",
 148                     "url": f"https://factorio.com/get-download/{version}/{release_type.name}/{system.url_name}",
 149                     "version": version,
 150                     "needsAuth": release_type.needs_auth,
 151                     "candidateHashFilenames": [
 152                         fmt.format(version=version)
 153                         for fmt in release_type.hash_filename_format
 154                     ],
 155                     "tarDirectory": system.tar_name,
 156                 }
 157                 output[system.nix_name][release_type.name][release_channel.name] = (
 158                     this_release
 159                 )
 160     return output
 161
 162
 163 def iter_version(
 164     versions: OurVersionJSON,
 165     it: Callable[[str, str, str, Dict[str, str]], Dict[str, str]],
 166 ) -> OurVersionJSON:
 167     versions = copy.deepcopy(versions)
 168     for system_name, system in versions.items():
 169         for release_type_name, release_type in system.items():
 170             for release_channel_name, release in release_type.items():
 171                 release_type[release_channel_name] = it(
 172                     system_name, release_type_name, release_channel_name, dict(release)
 173                 )
 174     return versions
 175
 176
 177 def merge_versions(old: OurVersionJSON, new: OurVersionJSON) -> OurVersionJSON:
 178     """Copies already-known hashes from version.json to avoid having to re-fetch."""
 179
 180     def _merge_version(
 181         system_name: str,
 182         release_type_name: str,
 183         release_channel_name: str,
 184         release: Dict[str, str],
 185     ) -> Dict[str, str]:
 186         old_system = old.get(system_name, {})
 187         old_release_type = old_system.get(release_type_name, {})
 188         old_release = old_release_type.get(release_channel_name, {})
 189         if FLAGS.release_type and release_type_name not in FLAGS.release_type:
 190             logging.info(
 191                 "%s/%s/%s: not in --release_type, not updating",
 192                 system_name,
 193                 release_type_name,
 194                 release_channel_name,
 195             )
 196             return old_release
 197         if FLAGS.release_channel and release_channel_name not in FLAGS.release_channel:
 198             logging.info(
 199                 "%s/%s/%s: not in --release_channel, not updating",
 200                 system_name,
 201                 release_type_name,
 202                 release_channel_name,
 203             )
 204             return old_release
 205         if "sha256" not in old_release:
 206             logging.info(
 207                 "%s/%s/%s: not copying sha256 since it's missing",
 208                 system_name,
 209                 release_type_name,
 210                 release_channel_name,
 211             )
 212             return release
 213         if not all(
 214             old_release.get(k, None) == release[k] for k in ["name", "version", "url"]
 215         ):
 216             logging.info(
 217                 "%s/%s/%s: not copying sha256 due to mismatch",
 218                 system_name,
 219                 release_type_name,
 220                 release_channel_name,
 221             )
 222             return release
 223         release["sha256"] = old_release["sha256"]
 224         return release
 225
 226     return iter_version(new, _merge_version)
 227
 228
 229 def fill_in_hash(
 230     versions: OurVersionJSON, factorio_hashes: FactorioHashes
 231 ) -> OurVersionJSON:
 232     """Fill in sha256 hashes for anything missing them."""
 233
 234     def _fill_in_hash(
 235         system_name: str,
 236         release_type_name: str,
 237         release_channel_name: str,
 238         release: Dict[str, str],
 239     ) -> Dict[str, str]:
 240         for candidate_filename in release["candidateHashFilenames"]:
 241             if candidate_filename in factorio_hashes:
 242                 release["sha256"] = factorio_hashes[candidate_filename]
 243                 break
 244         else:
 245             logging.error(
 246                 "%s/%s/%s: failed to find any of %s in %s",
 247                 system_name,
 248                 release_type_name,
 249                 release_channel_name,
 250                 release["candidateHashFilenames"],
 251                 FACTORIO_HASHES,
 252             )
 253             return release
 254         if "sha256" in release:
 255             logging.info(
 256                 "%s/%s/%s: skipping fetch, sha256 already present",
 257                 system_name,
 258                 release_type_name,
 259                 release_channel_name,
 260             )
 261             return release
 262         return release
 263
 264     return iter_version(versions, _fill_in_hash)
 265
 266
 267 def main(argv):
 268     factorio_versions = fetch_versions()
 269     factorio_hashes = fetch_hashes()
 270     new_our_versions = generate_our_versions(factorio_versions)
 271     old_our_versions = None
 272     our_versions_path = find_versions_json()
 273     if our_versions_path:
 274         logging.info("Loading old versions.json from %s", our_versions_path)
 275         with open(our_versions_path, "r") as f:
 276             old_our_versions = json.load(f)
 277     if old_our_versions:
 278         logging.info("Merging in old hashes")
 279         new_our_versions = merge_versions(old_our_versions, new_our_versions)
 280     logging.info("Updating hashes from Factorio SHA256")
 281     new_our_versions = fill_in_hash(new_our_versions, factorio_hashes)
 282     with open(our_versions_path, "w") as f:
 283         logging.info("Writing versions.json to %s", our_versions_path)
 284         json.dump(new_our_versions, f, sort_keys=True, indent=2)
 285         f.write("\n")
 286
 287
 288 if __name__ == "__main__":
 289     app.run(main)