Files
APWorldTester/src/archipelago_tester/pipeline/batch/discontinued.py
2026-09-06 15:12:38 +02:00

70 lines
2.6 KiB
Python

"""Flagging test records whose game left the sheet."""
from archipelago_tester.core.model.name import Name
from archipelago_tester.core.state.keys import StateKeys
class Discontinued:
"""A world taken off the sheet stops being one this pipeline
tracks: it should leave the status page and stop being deployed.
The record is kept and labelled rather than deleted, so its history
survives and a row that comes back - or that vanished because of a
bad sheet parse - is one flag away from being live again.
"""
def __init__(self, store):
self.store = store
@staticmethod
def sheet_name(key, record):
"""The sheet row a record's key came from.
A download- or core-stage record is keyed on the bare sheet
name, since nothing was ever fetched for it. Everything else is
keyed "<folder>/<file>.apworld". The record's own "game" is
deliberately not used: that is the name Archipelago registered
the world under, which can differ from the sheet's wording.
"""
if (record or {}).get("stage") in ("download", "core"):
return key
return key.split("/", 1)[0]
def apply(self, tests, live):
"""Set or clear the flag, reporting only what moved.
Comparison is on the folded directory name of both sides, which
is what turned the sheet name into the download folder in the
first place, so punctuation and case cannot make a still-listed
game look discontinued. An unchanged run leaves the state file
byte-identical.
"""
flag = StateKeys.DISCONTINUED
changed = {}
for key, record in tests.items():
name = self.sheet_name(key, record)
gone = Name(name).directory not in live
if gone != bool(record.get(flag)):
changed[key] = gone
if gone:
record[flag] = True
else:
record.pop(flag, None)
return changed
def mark(self, sheet_names):
"""Flag every record whose game is no longer listed.
Does nothing when handed no sheet names: an empty list means
the sheet was not read, not that every world on earth was
retired, and acting on it would empty the page.
"""
if not sheet_names:
return {}
state = self.store.load()
tests = state.get(StateKeys.APWORLD_TESTS, {})
changed = self.apply(
tests, {Name(name).directory for name in sheet_names})
if changed:
self.store.save(state)
return changed