otp-import/qr_to_sheet.py

#!/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))


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

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:
        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)

    keys = existing_keys(read_csv_rows(args.csv))
    new_rows, dups = dedup_new(candidates, keys)

    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:
            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}.")
    else:
        print("Nothing to append.")
    return 0


if __name__ == "__main__":
    sys.exit(main())