From b916165cab49801f6d613968a96082155725a470 Mon Sep 17 00:00:00 2001 From: bob walker Date: Thu, 10 Sep 2026 12:07:43 +0100 Subject: [PATCH] Read credentials and settings from config files Reuse the flickr key flickrtouchr already holds and the immich url and key immich-sync uses, so nothing secret is duplicated. Everything else comes from a typed settings file with flag > environment > file > default precedence, so a cron line can be the bare script. Co-Authored-By: Claude Fable 5.1 --- rgl-photo-backup | 162 +++++++++++++++++++++++++++++++++++++++ test_rgl_photo_backup.py | 104 +++++++++++++++++++++++++ 2 files changed, 266 insertions(+) diff --git a/rgl-photo-backup b/rgl-photo-backup index 9826355..08af211 100755 --- a/rgl-photo-backup +++ b/rgl-photo-backup @@ -110,5 +110,167 @@ def debug(msg): print("debug: %s" % msg, file=sys.stderr) +# +# All three config files are flat "name: value" yaml. Parsing it properly +# would mean a dependency, so read exactly that shape. A missing file is an +# empty one. +# +def readyaml(path): + values = {} + try: + with open(os.path.expanduser(path), encoding="utf-8") as fh: + for line in fh: + if line.lstrip().startswith("#"): + continue + found = re.match(r'^\s*([A-Za-z_][\w-]*)\s*:\s*(.*?)\s*$', line) + if found: + values[found.group(1)] = found.group(2).strip().strip("\"'") + except FileNotFoundError: + pass + except OSError as e: + raise BackupError("%s: cannot be read: %s" % (path, e.strerror or e)) + return values + +def loadflickr(path): + key = readyaml(path).get("key") + if not key: + raise BackupError("%s: needs a flickr api 'key'" % path) + return key + +def loadimmich(path): + values = readyaml(path) + url = values.get("url", "").rstrip("/") + key = values.get("key") + if not url or not key: + raise BackupError("%s: needs both a url and a key" % path) + return (url, key) + + +SETTING_KINDS = { + "directory": "path", + "flickr_auth": "path", + "immich_auth": "path", + "index_url": str, + "fetch": bool, + "sync": bool, + "dry_run": bool, + "limit": int, + "pace": float, + "quiet": bool, + "attempts": int, + "backoff": float, + "timeout": float, + "user_agent": str, + "debug": bool, +} + +BOOLEANS = {"true": True, "yes": True, "on": True, "1": True, + "false": False, "no": False, "off": False, "0": False} + +def coerce(kind, raw): + if kind == "path": + return os.path.expanduser(raw) + if kind is bool: + return BOOLEANS[raw.lower()] + return kind(raw) + +def loadsettings(path): + path = os.path.expanduser(path) + settings = {} + for (name, raw) in readyaml(path).items(): + kind = SETTING_KINDS.get(name) + if kind is None: + warn("%s: unknown setting '%s' ignored" % (path, name)) + continue + try: + settings[name] = coerce(kind, raw) + except (KeyError, ValueError): + wanted = {bool: "true or false", int: "a whole number", + float: "a number"}.get(kind, "text") + raise BackupError("%s: '%s' must be %s" % (path, name, wanted)) + return settings + +def parseargs(argv=None): + parser = argparse.ArgumentParser( + description="Back up the photos the Randomness Guide to London shows " + "to a directory, then mirror them into an immich album.", + epilog="Every option can also live in the settings file; a flag beats " + "the environment, which beats the file.") + parser.add_argument("directory", nargs="?", + help="the backup directory (default: 'directory:' in " + "the settings file)") + parser.add_argument("--fetch-only", action="store_true", default=None, + help="RGL and flickr to disk, no immich") + parser.add_argument("--sync-only", action="store_true", default=None, + help="disk to immich, no RGL or flickr calls") + parser.add_argument("--dry-run", action="store_true", default=None, + help="report what would be fetched and written, change nothing") + parser.add_argument("--limit", type=int, metavar="N", + help="stop after N photos, for a trial run") + parser.add_argument("--pace", type=float, metavar="SECONDS", + help="delay between flickr calls") + parser.add_argument("--quiet", "-q", action="store_true", default=None, + help="print only failures, for cron") + parser.add_argument("--index-url", metavar="URL", + help="the wiki index endpoint (default: RGL)") + parser.add_argument("--flickr-config", metavar="FILE", + help="flickr key file (default: %s)" % FLICKR_CONFIG) + parser.add_argument("--immich-config", metavar="FILE", + help="immich url and key file (default: %s)" % IMMICH_CONFIG) + parser.add_argument("--settings", metavar="FILE", default=SETTINGS_FILE, + help="settings file (default: %s)" % SETTINGS_FILE) + parser.add_argument("--debug", action="store_true", default=None, + help="log every request to stderr") + return parser.parse_args(argv) + +# Settings the environment may also supply, by key +ENVIRONMENT = {"pace": "PACE", "attempts": "ATTEMPTS", "backoff": "BACKOFF", + "timeout": "TIMEOUT", "user_agent": "USER_AGENT", "debug": "DEBUG"} + +def settings(args, environ=None): + environ = os.environ if environ is None else environ + filed = loadsettings(args.settings) + + def pick(name, flag=None, default=None): + if flag is not None: + return flag + env = ENVIRONMENT.get(name) + if env and environ.get(ENV + env): + try: + return coerce(SETTING_KINDS[name], environ[ENV + env]) + except (KeyError, ValueError): + raise BackupError("%s%s: %r is not a valid %s" + % (ENV, env, environ[ENV + env], name)) + return filed.get(name, default) + + conf = { + "directory": pick("directory", args.directory), + "flickr_auth": pick("flickr_auth", args.flickr_config, + os.path.expanduser(DEFAULTS["flickr_auth"])), + "immich_auth": pick("immich_auth", args.immich_config, + os.path.expanduser(DEFAULTS["immich_auth"])), + "index_url": pick("index_url", args.index_url, DEFAULTS["index_url"]), + "fetch": pick("fetch", False if args.sync_only else None, DEFAULTS["fetch"]), + "sync": pick("sync", False if args.fetch_only else None, DEFAULTS["sync"]), + "dry_run": pick("dry_run", args.dry_run, DEFAULTS["dry_run"]), + "limit": pick("limit", args.limit, DEFAULTS["limit"]), + "pace": pick("pace", args.pace, DEFAULTS["pace"]), + "quiet": pick("quiet", args.quiet, DEFAULTS["quiet"]), + "attempts": pick("attempts", None, DEFAULTS["attempts"]), + "backoff": pick("backoff", None, DEFAULTS["backoff"]), + "timeout": pick("timeout", None, DEFAULTS["timeout"]), + "user_agent": pick("user_agent", None, DEFAULTS["user_agent"]), + "debug": pick("debug", args.debug, DEFAULTS["debug"]), + } + if conf["directory"]: + conf["directory"] = os.path.expanduser(conf["directory"]) + else: + raise BackupError("no backup directory: give it on the command line, or " + "as 'directory:' in %s" % os.path.expanduser(args.settings)) + if not conf["fetch"] and not conf["sync"]: + raise BackupError("--fetch-only and --sync-only together leave nothing to do") + return conf + + if __name__ == "__main__": sys.exit(0) diff --git a/test_rgl_photo_backup.py b/test_rgl_photo_backup.py index ed0b5b1..ee133de 100644 --- a/test_rgl_photo_backup.py +++ b/test_rgl_photo_backup.py @@ -132,3 +132,107 @@ class Output(TestCase): e = rb.HTTPFailure("ctx: HTTP 404", code=404) self.assertEqual(e.code, 404) self.assertIsInstance(e, rb.BackupError) + + +class Config(TestCase): + def test_readyaml_reads_flat_pairs_and_ignores_comments(self): + path = self.write("a.yml", "# note\nkey: 'abc'\nurl: https://immich.example/api \n") + self.assertEqual(rb.readyaml(path), {"key": "abc", "url": "https://immich.example/api"}) + + def test_readyaml_missing_file_is_empty(self): + self.assertEqual(rb.readyaml(os.path.join(self.dir, "none.yml")), {}) + + def test_readyaml_unreadable_path_names_it(self): + with self.assertRaises(rb.BackupError) as cm: + rb.readyaml(self.dir) # a directory, not a file + self.assertIn(self.dir, str(cm.exception)) + + def test_loadflickr_needs_key(self): + path = self.write("f.yml", "secret: s\n") + with self.assertRaises(rb.BackupError): + rb.loadflickr(path) + path = self.write("f2.yml", "key: k\nsecret: s\n") + self.assertEqual(rb.loadflickr(path), "k") + + def test_loadimmich_needs_url_and_key_and_strips_slash(self): + path = self.write("i.yml", "url: https://immich.example/api/\nkey: abc\n") + self.assertEqual(rb.loadimmich(path), ("https://immich.example/api", "abc")) + with self.assertRaises(rb.BackupError): + rb.loadimmich(self.write("i2.yml", "url: x\n")) + + +class Settings(TestCase): + def test_flag_beats_env_beats_file_beats_default(self): + path = self.write("s.yml", "directory: /from/file\npace: 3\nquiet: true\n") + args = rb.parseargs(["--settings", path]) + conf = rb.settings(args, environ={}) + self.assertEqual(conf["directory"], "/from/file") + self.assertEqual(conf["pace"], 3.0) + self.assertTrue(conf["quiet"]) + conf = rb.settings(args, environ={rb.ENV + "PACE": "7"}) + self.assertEqual(conf["pace"], 7.0) + args = rb.parseargs(["--settings", path, "--pace", "9", "/cli"]) + conf = rb.settings(args, environ={rb.ENV + "PACE": "7"}) + self.assertEqual(conf["pace"], 9.0) + self.assertEqual(conf["directory"], "/cli") + + def test_defaults(self): + args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"]) + conf = rb.settings(args, environ={}) + self.assertEqual(conf["index_url"], rb.INDEX_URL) + self.assertTrue(conf["fetch"] and conf["sync"]) + self.assertFalse(conf["dry_run"]) + self.assertIsNone(conf["limit"]) + self.assertEqual(conf["attempts"], 6) + self.assertEqual(conf["flickr_auth"], os.path.expanduser(rb.FLICKR_CONFIG)) + + def test_defaults_are_literals_not_moving_constants(self): + rb.PACE, rb.ATTEMPTS, rb.USER_AGENT = 99.0, 99, "mangled" + args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"]) + conf = rb.settings(args, environ={}) + self.assertEqual(conf["pace"], rb.DEFAULTS["pace"]) + self.assertEqual(conf["attempts"], rb.DEFAULTS["attempts"]) + self.assertEqual(conf["user_agent"], rb.DEFAULTS["user_agent"]) + self.assertEqual((rb.DEFAULTS["pace"], rb.DEFAULTS["attempts"]), (1.0, 6)) + + def test_debug_is_a_setting(self): + path = self.write("s.yml", "directory: /d\ndebug: true\n") + self.assertEqual(self.stderr.getvalue(), "") # no "unknown setting" + args = rb.parseargs(["--settings", path]) + self.assertTrue(rb.settings(args, environ={})["debug"]) + args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"]) + self.assertFalse(rb.settings(args, environ={})["debug"]) + self.assertTrue(rb.settings(args, environ={rb.ENV + "DEBUG": "yes"})["debug"]) + args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "--debug", "/d"]) + self.assertTrue(rb.settings(args, environ={})["debug"]) + + def test_a_bad_environment_value_is_a_configuration_error(self): + args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"]) + with self.assertRaises(rb.BackupError) as cm: + rb.settings(args, environ={rb.ENV + "ATTEMPTS": "lots"}) + self.assertIn("ATTEMPTS", str(cm.exception)) + + def test_phase_flags(self): + base = ["--settings", os.path.join(self.dir, "none"), "/d"] + self.assertFalse(rb.settings(rb.parseargs(base + ["--sync-only"]), environ={})["fetch"]) + self.assertFalse(rb.settings(rb.parseargs(base + ["--fetch-only"]), environ={})["sync"]) + with self.assertRaises(rb.BackupError): + rb.settings(rb.parseargs(base + ["--fetch-only", "--sync-only"]), environ={}) + + def test_missing_directory_is_an_error(self): + args = rb.parseargs(["--settings", os.path.join(self.dir, "none")]) + with self.assertRaises(rb.BackupError): + rb.settings(args, environ={}) + + def test_unknown_key_warns_and_wrong_type_fails(self): + path = self.write("s.yml", "directory: /d\nbogus: 1\n") + rb.loadsettings(path) + self.assertIn("bogus", self.stderr.getvalue()) + path = self.write("s2.yml", "directory: /d\nlimit: many\n") + with self.assertRaises(rb.BackupError) as cm: + rb.loadsettings(path) + self.assertIn("limit", str(cm.exception)) + + def test_tilde_expanded_in_paths(self): + path = self.write("s.yml", "directory: ~/photos\n") + self.assertEqual(rb.loadsettings(path)["directory"], os.path.expanduser("~/photos"))