diff --git a/README.md b/README.md index 60914bd..3e8ab17 100644 --- a/README.md +++ b/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' diff --git a/apex_otp_import.py b/apex_otp_import.py index 2493d05..fa243ee 100644 --- a/apex_otp_import.py +++ b/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) diff --git a/qr_to_sheet.py b/qr_to_sheet.py index a34bde1..a8422d3 100644 --- a/qr_to_sheet.py +++ b/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): - 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) - - 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: +def collect_candidates(image_dir): + """Decode every QR image in a directory. Returns + (candidates, undecodable, unsupported) where candidates is a list of + (otpauth_uri, source_file).""" + candidates, undecodable, unsupported = [], [], [] + for path in scan_images(image_dir): 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: - for r in new_rows: +def run_qr_import(image_dir, csv_path, dry_run=False): + """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 - - if new_rows: - append_csv(args.csv, new_rows) - print(f"Appended {len(new_rows)} row(s) to {args.csv}.") + elif summary["added"]: + print(f"Appended {summary['added']} row(s) to {csv_path}.") 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 diff --git a/test_qr.py b/test_qr.py index 6392975..025655c 100644 --- a/test_qr.py +++ b/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()