Using SHA-256 as an Idempotency Key in a File Anchoring Pipeline

import sqlite3
from pathlib import Path

DB_PATH = Path("anchor_state.db")

def init_db(conn: sqlite3.Connection) -> None:
    conn.execute("""
        CREATE TABLE IF NOT EXISTS anchored (
            sha256      TEXT PRIMARY KEY,
            proof_id    TEXT NOT NULL,
            anchored_at TEXT
        )
    """)
    conn.commit()

def already_anchored(conn: sqlite3.Connection, sha256: str) -> dict | None:
    row = conn.execute(
        "SELECT proof_id, anchored_at FROM anchored WHERE sha256 = ?",
        (sha256,)
    ).fetchone()
    if row:
        return {"proof_id": row[0], "anchored_at": row[1]}
    return None

def record_anchor(
    conn: sqlite3.Connection,
    sha256: str,
    proof_id: str,
    anchored_at: str,
) -> None:
    conn.execute(
        "INSERT OR IGNORE INTO anchored (sha256, proof_id, anchored_at) VALUES (?, ?, ?)",
        (sha256, proof_id, anchored_at),
    )
    conn.commit()

import hashlib
import requests
import time

def hash_file(filepath: str) -> str:
    h = hashlib.sha256()
    with open(filepath, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()

def call_anchor_api(sha256: str, filename: str, api_key: str) -> dict:
    while True:
        resp = requests.post(
            "https://proofledger.io/api/v1/proof",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"sha256": sha256, "filename": filename},
            timeout=30,
        )
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", "60"))
            print(f"Rate limited. Waiting {retry_after}s.")
            time.sleep(retry_after)
            continue
        resp.raise_for_status()
        return resp.json()

from pathlib import Path

def process_file(
    filepath: str,
    api_key: str,
    conn: sqlite3.Connection,
) -> dict:
    sha256 = hash_file(filepath)
    filename = Path(filepath).name

    # Layer 1: local check
    existing = already_anchored(conn, sha256)
    if existing:
        return {
            "file": filename,
            "sha256": sha256,
            "status": "skipped",
            "proof_id": existing["proof_id"],
        }

    # Layer 2: API call
    try:
        result = call_anchor_api(sha256, filename, api_key)
    except requests.exceptions.HTTPError as e:
        return {"file": filename, "sha256": sha256, "status": "error", "error": str(e)}

    proof_id = result.get("id", "")
    anchored_at = result.get("created_at", "")
    duplicate_of = result.get("duplicate_of")

    if duplicate_of:
        # The canonical anchor predates this one. Record the original.
        record_anchor(conn, sha256, duplicate_of, anchored_at)
        return {
            "file": filename,
            "sha256": sha256,
            "status": "duplicate",
            "canonical_proof_id": duplicate_of,
        }

    record_anchor(conn, sha256, proof_id, anchored_at)
    return {
        "file": filename,
        "sha256": sha256,
        "status": "anchored",
        "proof_id": proof_id,
    }

import os
import sqlite3
from pathlib import Path

API_KEY = os.environ.get("PROOFLEDGER_API_KEY", "sk_YOUR_KEY_HERE")
INCOMING = Path("./incoming")

def run(directory: Path, api_key: str) -> None:
    conn = sqlite3.connect(str(DB_PATH))
    init_db(conn)

    files = [f for f in directory.iterdir() if f.is_file()]
    print(f"Processing {len(files)} files.")

    counts: dict[str, int] = {"anchored": 0, "skipped": 0, "duplicate": 0, "error": 0}

    for filepath in files:
        result = process_file(str(filepath), api_key, conn)
        status = result["status"]
        counts[status] = counts.get(status, 0) + 1
        ref = (
            result.get("proof_id")
            or result.get("canonical_proof_id")
            or result.get("error", "")
        )
        print(f"  [{status}] {result['file']} -> {ref}")

    conn.close()
    print(
        f"\nAnchored: {counts['anchored']}, Skipped: {counts['skipped']}, "
        f"Duplicate: {counts['duplicate']}, Errors: {counts['error']}"
    )

if __name__ == "__main__":
    run(INCOMING, API_KEY)

from verify_proof import hash_file as vp_hash_file, verify_proof, load_proof

# proof.json downloaded from the certificate page or stored by your pipeline
proof = load_proof("evidence_photo_proof.json")
file_hash = vp_hash_file("evidence_photo.jpg")
result = verify_proof(file_hash, proof)

if result["verified"]:
    print(f"Verified on {result['blockchain']} at {result['anchored_at']}")
else:
    print(f"Not verified: {result['error']}")