Add --import-qr to run the QR->CSV step before importing

Wrap the QR decode/append step behind --import-qr on the importer so a
single command runs the whole pipeline: decode QR images, append only
new credentials to the CSV (deduped on otpauth://, existing entries
silently skipped), then validate and import. QR folder defaults to the
CSV's folder; override with --qr-images. Refactor qr_to_sheet into
reusable collect_candidates / apply_new_candidates / run_qr_import.

README.md

@@ -72,6 +72,11 @@ python apex_otp_import.py --input ... --skip-existing
 # Target a specific reader, supply a password, and profile memory
 python apex_otp_import.py --input ... --reader VivoKey --password '...' --profile-memory
 
+# One-shot pipeline: decode new QRs into the CSV (only new, deduped on
+# otpauth://), then validate and import. QR folder defaults to the CSV's folder.
+python apex_otp_import.py --input otp-keys.csv --import-qr --reader VivoKey
+python apex_otp_import.py --input otp-keys.csv --import-qr --qr-images /qr/folder --overwrite-all
+
 # Import, then set an access password (only if the applet has none yet)
 python apex_otp_import.py --input ... --set-password 'yourpassword'
 

apex_otp_import.py

@@ -675,6 +675,11 @@ def parse_args(argv=None):
     p.add_argument("--memlog", default=DEFAULT_MEMLOG,
                    help=f"profiling CSV path (default: {DEFAULT_MEMLOG})")
     p.add_argument("--password", help="applet password (else prompted if locked)")
+    p.add_argument("--import-qr", action="store_true",
+                   help="first decode QR images and append new credentials to the "
+                        "CSV (deduped on otpauth://), then import the CSV")
+    p.add_argument("--qr-images", metavar="DIR",
+                   help="QR image folder for --import-qr (default: the folder of --input)")
     p.add_argument("--reset", action="store_true",
                    help="wipe all credentials and any password, then exit")
     p.add_argument("--set-password", metavar="PASS",
@@ -707,6 +712,15 @@ def main(argv=None):
         print(f"Applet reset on '{reader}' (all credentials and any password wiped).")
         return 0
 
+    # 0. Optionally decode QR images into the CSV first (stage 1 of the pipeline).
+    if args.import_qr:
+        from qr_to_sheet import run_qr_import, print_qr_summary
+        images_dir = args.qr_images or os.path.dirname(os.path.abspath(args.input)) or "."
+        print(f"QR import: {images_dir} -> {args.input}")
+        summary = run_qr_import(images_dir, args.input, dry_run=args.dry_run)
+        print_qr_summary(summary, args.input, args.dry_run)
+        print()
+
     # 1. Load + validate BEFORE touching any reader.
     creds = load_credentials(args.input)
     problems = validate_input(creds)

qr_to_sheet.py

@@ -116,30 +116,18 @@ def scan_images(image_dir):
 
 
 # ---------------------------------------------------------------------------
-# CLI
+# Orchestration (reused by this CLI and by apex_otp_import --import-qr)
 # ---------------------------------------------------------------------------
 
 DEFAULT_CSV = "otp-keys.csv"
 
 
-def main(argv=None):
+def collect_candidates(image_dir):
-    p = argparse.ArgumentParser(
+    """Decode every QR image in a directory. Returns
-        description="Decode OTP QR screenshots and append new credentials to the "
+    (candidates, undecodable, unsupported) where candidates is a list of
-                    "canonical CSV (deduplicated on the otpauth:// data).")
+    (otpauth_uri, source_file)."""
-    p.add_argument("--images", default=".", help="directory of QR images")
+    candidates, undecodable, unsupported = [], [], []
-    p.add_argument("--csv", default=DEFAULT_CSV, help=f"CSV path (default: {DEFAULT_CSV})")
+    for path in scan_images(image_dir):
-    p.add_argument("--dry-run", action="store_true", help="report without writing")
-    args = p.parse_args(argv)
-
-    images = scan_images(args.images)
-    if not images:
-        print(f"No QR images found in {args.images}")
-        return 1
-
-    candidates = []      # (otpauth_uri, source_file)
-    undecodable = []
-    unsupported = []
-    for path in images:
         name = os.path.basename(path)
         data = decode_qr_image(path)
         if not data:
@@ -150,29 +138,65 @@ def main(argv=None):
             unsupported.append(name)
         else:
             undecodable.append(name)
+    return candidates, undecodable, unsupported
 
-    keys = existing_keys(read_csv_rows(args.csv))
+
+def apply_new_candidates(candidates, csv_path, dry_run=False):
+    """Append only candidates whose otpauth is not already in the CSV. Existing
+    rows are never modified; duplicates are silently skipped (counted only)."""
+    keys = existing_keys(read_csv_rows(csv_path))
     new_rows, dups = dedup_new(candidates, keys)
+    if new_rows and not dry_run:
+        append_csv(csv_path, new_rows)
+    return {"scanned": len(candidates), "added": len(new_rows),
+            "duplicates": dups, "new_rows": new_rows}
 
-    print(f"Scanned {len(images)} image(s): {len(candidates)} otpauth QR(s) decoded.")
-    print(f"  {len(new_rows)} new, {dups} duplicate(s) skipped.")
-    if unsupported:
-        print(f"  {len(unsupported)} otpauth-migration:// (unsupported): "
-              + ", ".join(unsupported))
-    if undecodable:
-        print(f"  {len(undecodable)} could not be decoded (add manually): "
-              + ", ".join(undecodable))
 
-    if args.dry_run:
+def run_qr_import(image_dir, csv_path, dry_run=False):
-        for r in new_rows:
+    """Decode a folder of QR images and append new credentials to the CSV."""
+    candidates, undecodable, unsupported = collect_candidates(image_dir)
+    summary = apply_new_candidates(candidates, csv_path, dry_run)
+    summary["undecodable"] = undecodable
+    summary["unsupported"] = unsupported
+    summary["images"] = len(candidates) + len(undecodable) + len(unsupported)
+    return summary
+
+
+def print_qr_summary(summary, csv_path, dry_run=False):
+    print(f"Scanned {summary['images']} image(s): "
+          f"{summary['scanned']} otpauth QR(s) decoded.")
+    print(f"  {summary['added']} new, {summary['duplicates']} duplicate(s) skipped.")
+    if summary["unsupported"]:
+        print("  otpauth-migration:// (unsupported): " + ", ".join(summary["unsupported"]))
+    if summary["undecodable"]:
+        print("  could not be decoded (add manually): " + ", ".join(summary["undecodable"]))
+    if dry_run:
+        for r in summary["new_rows"]:
             print(f"  + {r['issuer']}:{r['account']}  ({r['source_file']})")
-        return 0
+    elif summary["added"]:
-
+        print(f"Appended {summary['added']} row(s) to {csv_path}.")
-    if new_rows:
-        append_csv(args.csv, new_rows)
-        print(f"Appended {len(new_rows)} row(s) to {args.csv}.")
     else:
-        print("Nothing to append.")
+        print("Nothing new to append.")
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+def main(argv=None):
+    p = argparse.ArgumentParser(
+        description="Decode OTP QR screenshots and append new credentials to the "
+                    "canonical CSV (deduplicated on the otpauth:// data).")
+    p.add_argument("--images", default=".", help="directory of QR images")
+    p.add_argument("--csv", default=DEFAULT_CSV, help=f"CSV path (default: {DEFAULT_CSV})")
+    p.add_argument("--dry-run", action="store_true", help="report without writing")
+    args = p.parse_args(argv)
+
+    summary = run_qr_import(args.images, args.csv, dry_run=args.dry_run)
+    if summary["images"] == 0:
+        print(f"No QR images found in {args.images}")
+        return 1
+    print_qr_summary(summary, args.csv, args.dry_run)
     return 0
 
 

test_qr.py

@@ -4,7 +4,8 @@ import tempfile
 import unittest
 
 from otpauth import parse_otpauth, canonical_key
-from qr_to_sheet import existing_keys, dedup_new, append_csv, read_csv_rows
+from qr_to_sheet import (existing_keys, dedup_new, append_csv, read_csv_rows,
+                         apply_new_candidates)
 
 
 class TestParseOtpauth(unittest.TestCase):
@@ -83,5 +84,33 @@ class TestCsvIO(unittest.TestCase):
         self.assertEqual(read_csv_rows(os.path.join(tempfile.mkdtemp(), "nope.csv")), [])
 
 
+class TestApplyNewCandidates(unittest.TestCase):
+    def _csv_with_A(self):
+        p = os.path.join(tempfile.mkdtemp(), "otp-keys.csv")
+        append_csv(p, [{"issuer": "X", "account": "a", "otpauth_uri": A, "source_file": "a.png"}])
+        return p
+
+    def test_only_new_otpauth_is_appended(self):
+        p = self._csv_with_A()
+        summary = apply_new_candidates([(A, "a2.png"), (B, "b.png"), (B, "b2.png")], p)
+        self.assertEqual(summary["added"], 1)
+        self.assertEqual(summary["duplicates"], 2)
+        rows = read_csv_rows(p)
+        self.assertEqual(len(rows), 2)                      # existing A + new B only
+        self.assertEqual(rows[1]["otpauth_uri"], B)
+
+    def test_all_existing_appends_nothing(self):
+        p = self._csv_with_A()
+        summary = apply_new_candidates([(A, "a2.png")], p)
+        self.assertEqual((summary["added"], summary["duplicates"]), (0, 1))
+        self.assertEqual(len(read_csv_rows(p)), 1)          # unchanged
+
+    def test_dry_run_reports_but_does_not_write(self):
+        p = self._csv_with_A()
+        summary = apply_new_candidates([(B, "b.png")], p, dry_run=True)
+        self.assertEqual(summary["added"], 1)
+        self.assertEqual(len(read_csv_rows(p)), 1)          # not written
+
+
 if __name__ == "__main__":
     unittest.main()