Amal Graafstra
a8a405600c
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.
205 lines
7.3 KiB
Python
205 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Decode saved OTP QR-code screenshots into rows of the canonical CSV.
|
|
|
|
Stage 1 of the pipeline (see docs/qr-to-csv-design.md): decode each QR, dedup on
|
|
the otpauth:// data, and append only new credentials. The otpauth:// string is
|
|
the immutable source of truth; the issuer/account columns are editable labels
|
|
seeded from it.
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import glob
|
|
import os
|
|
import sys
|
|
|
|
from otpauth import parse_otpauth, canonical_key
|
|
|
|
CSV_COLUMNS = ["issuer", "account", "otpauth_uri", "source_file"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CSV + dedup (pure)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def read_csv_rows(path):
|
|
"""Read the canonical CSV into a list of dict rows ([] if it doesn't exist)."""
|
|
if not os.path.exists(path):
|
|
return []
|
|
with open(path, newline="", encoding="utf-8") as f:
|
|
return list(csv.DictReader(f))
|
|
|
|
|
|
def append_csv(path, rows):
|
|
"""Append rows to the CSV, writing the header first if the file is new.
|
|
Existing rows are never rewritten or reordered."""
|
|
new_file = not os.path.exists(path)
|
|
with open(path, "a", newline="", encoding="utf-8") as f:
|
|
w = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
|
|
if new_file:
|
|
w.writeheader()
|
|
for r in rows:
|
|
w.writerow({k: r.get(k, "") for k in CSV_COLUMNS})
|
|
|
|
|
|
def existing_keys(rows):
|
|
"""Canonical dedup keys for every row that carries an otpauth_uri."""
|
|
keys = set()
|
|
for r in rows:
|
|
uri = (r.get("otpauth_uri") or "").strip()
|
|
if uri:
|
|
keys.add(canonical_key(parse_otpauth(uri)))
|
|
return keys
|
|
|
|
|
|
def dedup_new(candidates, existing):
|
|
"""From (otpauth_uri, source_file) candidates, return (new_rows, dup_count).
|
|
Skips any whose canonical otpauth is already present or seen earlier in the
|
|
batch."""
|
|
seen = set(existing)
|
|
new_rows = []
|
|
dups = 0
|
|
for uri, source in candidates:
|
|
key = canonical_key(parse_otpauth(uri))
|
|
if key in seen:
|
|
dups += 1
|
|
continue
|
|
seen.add(key)
|
|
o = parse_otpauth(uri)
|
|
new_rows.append({
|
|
"issuer": o.issuer,
|
|
"account": o.account,
|
|
"otpauth_uri": uri,
|
|
"source_file": source,
|
|
})
|
|
return new_rows, dups
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# QR decoding (I/O)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def decode_qr_image(path):
|
|
"""Decode a QR image to its payload string, or None. Multi-pass to cope with
|
|
scaling and thresholding differences across screenshots."""
|
|
import cv2
|
|
detector = cv2.QRCodeDetector()
|
|
img = cv2.imread(path)
|
|
if img is None:
|
|
return None
|
|
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) if img.ndim == 3 else img
|
|
attempts = [img, gray]
|
|
for scale in (2, 3, 4, 0.5):
|
|
h, w = gray.shape[:2]
|
|
attempts.append(cv2.resize(gray, (int(w * scale), int(h * scale)),
|
|
interpolation=cv2.INTER_CUBIC))
|
|
_, otsu = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
|
|
attempts.append(otsu)
|
|
attempts.append(cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
|
|
cv2.THRESH_BINARY, 31, 5))
|
|
for a in attempts:
|
|
try:
|
|
data, _, _ = detector.detectAndDecode(a)
|
|
except cv2.error:
|
|
continue
|
|
if data:
|
|
return data
|
|
return None
|
|
|
|
|
|
def scan_images(image_dir):
|
|
"""Return sorted image paths under image_dir."""
|
|
paths = []
|
|
for ext in ("*.png", "*.jpg", "*.jpeg", "*.PNG", "*.JPG"):
|
|
paths.extend(glob.glob(os.path.join(image_dir, ext)))
|
|
return sorted(set(paths))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Orchestration (reused by this CLI and by apex_otp_import --import-qr)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DEFAULT_CSV = "otp-keys.csv"
|
|
|
|
|
|
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:
|
|
undecodable.append(name)
|
|
elif data.startswith("otpauth://"):
|
|
candidates.append((data, name))
|
|
elif data.startswith("otpauth-migration://"):
|
|
unsupported.append(name)
|
|
else:
|
|
undecodable.append(name)
|
|
return candidates, undecodable, unsupported
|
|
|
|
|
|
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}
|
|
|
|
|
|
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']})")
|
|
elif summary["added"]:
|
|
print(f"Appended {summary['added']} row(s) to {csv_path}.")
|
|
else:
|
|
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
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|