tjh.dev
1#!/usr/bin/env nix-shell
2#! nix-shell -i python -p "python3.withPackages (ps: with ps; [ ps.absl-py ps.requests ])"
3
4from collections import defaultdict
5import copy
6from dataclasses import dataclass
7import json
8import os.path
9from typing import Callable, Dict
10
11from absl import app
12from absl import flags
13from absl import logging
14import requests
15
16
17FACTORIO_RELEASES = "https://factorio.com/api/latest-releases"
18FACTORIO_HASHES = "https://factorio.com/download/sha256sums/"
19
20
21FLAGS = flags.FLAGS
22
23flags.DEFINE_string("out", "", "Output path for versions.json.")
24flags.DEFINE_list(
25 "release_type",
26 "",
27 "If non-empty, a comma-separated list of release types to update (e.g. alpha).",
28)
29flags.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
37class System:
38 nix_name: str
39 url_name: str
40 tar_name: str
41
42
43@dataclass
44class ReleaseType:
45 name: str
46 hash_filename_format: list[str]
47 needs_auth: bool = False
48
49
50@dataclass
51class ReleaseChannel:
52 name: str
53
54
55FactorioVersionsJSON = Dict[str, Dict[str, str]]
56OurVersionJSON = Dict[str, Dict[str, Dict[str, Dict[str, str]]]]
57
58FactorioHashes = Dict[str, str]
59
60
61SYSTEMS = [
62 System(nix_name="x86_64-linux", url_name="linux64", tar_name="x64"),
63]
64
65RELEASE_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_linux_{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
86RELEASE_CHANNELS = [
87 ReleaseChannel("experimental"),
88 ReleaseChannel("stable"),
89]
90
91
92def 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
104def fetch_versions() -> FactorioVersionsJSON:
105 return json.loads(requests.get(FACTORIO_RELEASES).text)
106
107
108def 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
121def 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
163def 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
177def 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
229def 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
267def 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
288if __name__ == "__main__":
289 app.run(main)