Python Doc & Data Automation

Move Processed Files to Archive Folders with Python

The document job finishes each file with os.rename(path, "archive/" + path.name), and three problems surface within a month. On the first day a second statement.pdf arrives and the rename silently replaces yesterday's archived copy on Linux — or raises FileExistsError: [WinError 183] Cannot create a file when that file already exists on Windows. When the archive is moved to a network share, the job fails with OSError: [Errno 18] Invalid cross-device link. And after a crash halfway through a batch, nobody can tell which files in the inbox were already processed.

Root Cause

os.rename is a thin wrapper around the operating system's rename call, and its behaviour differs by platform and by where the destination lives. On POSIX systems it replaces an existing destination without warning; on Windows it refuses. It only works within one file system: a destination on another disk, mount or network share needs a copy followed by a delete, which os.rename does not do. shutil.move handles the cross-device case, but by copying and then deleting — two steps, not atomic — and it still overwrites existing files on POSIX when given a full destination path. Using the original filename as the archive name assumes names are unique, and they almost never are: invoice.pdf, export.csv and scan_0001.pdf repeat constantly. Finally, moving the file is the pipeline's record of completion, so moving it before the processing result is safely stored — or not at all on failure — breaks the guarantee that every input ends up in exactly one place.

Minimal Diagnostic

Check the conditions that decide how a move will behave: whether source and destination share a device, whether names already collide, and how many files are sitting in the inbox with no record of processing.

# stdlib only
import os
from collections import Counter
from pathlib import Path

INBOX = Path("drop/incoming")
ARCHIVE = Path("/mnt/archive/documents")        # e.g. a network share

def diagnose(inbox: Path, archive: Path) -> None:
    archive.mkdir(parents=True, exist_ok=True)
    same_device = inbox.stat().st_dev == archive.stat().st_dev
    print(f"same device: {same_device}  ({'atomic rename possible' if same_device else 'copy + delete needed'})")
    names = Counter(p.name for p in inbox.iterdir() if p.is_file())
    archived = {p.name for p in archive.rglob("*") if p.is_file()}
    repeats_in_inbox = {n: c for n, c in names.items() if c > 1}
    collisions = sorted(set(names) & archived)
    print(f"inbox files: {sum(names.values())}, names already in archive: {len(collisions)}")
    if collisions:
        print("   e.g.", collisions[:5])
    stale = [p.name for p in inbox.iterdir() if p.is_file() and
             (os.path.getmtime(p) < __import__("time").time() - 3600)]
    print(f"files older than 1 hour still in inbox: {len(stale)}")

if __name__ == "__main__":
    diagnose(INBOX, ARCHIVE)

same device: False  (copy + delete needed)
inbox files: 42, names already in archive: 17
   e.g. ['export.csv', 'invoice.pdf', 'scan_0001.pdf', 'scan_0002.pdf', 'statement.pdf']
files older than 1 hour still in inbox: 6

A cross-device archive rules out plain renames, seventeen names would overwrite or fail, and six old files suggest a previous crash left work unaccounted for.

How move functions behave os.rename overwrites an existing destination on Linux and raises FileExistsError on Windows, fails across devices, and is atomic within one file system. os.replace overwrites on both platforms, fails across devices, and is atomic within one file system. shutil.move overwrites when given a full destination path, works across devices by copying and deleting, and is not atomic across devices. None of them prevents name collisions, which must be handled by generating unique names. Function Existing dest Cross-device Atomic os.rename Linux overwrites; Win raises fails same FS only os.replace overwrites silently fails same FS only shutil.move overwrites (full path) copy + delete not across devices

Fix: Unique Destination, Staged Copy, Then Delete the Source

Build a collision-free destination name, move within the same file system with an atomic rename when possible, and across devices copy to a temporary name on the destination, verify, rename into place, and only then delete the source. Changed lines are commented.

# stdlib only
import hashlib
import os
import shutil
from datetime import datetime
from pathlib import Path

def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as fh:
        for block in iter(lambda: fh.read(1 << 20), b""):
            h.update(block)
    return h.hexdigest()

def unique_destination(folder: Path, name: str, digest: str) -> Path:
    stem, suffix = Path(name).stem, Path(name).suffix
    candidate = folder / f"{stem}__{digest[:10]}{suffix}"                  # changed: content-derived name
    n = 1
    while candidate.exists():
        if sha256(candidate) == digest:
            return candidate                                               # changed: identical file already archived
        candidate = folder / f"{stem}__{digest[:10]}_{n}{suffix}"
        n += 1
    return candidate

def archive(path: Path, root: Path, when: datetime | None = None) -> Path:
    when = when or datetime.now()
    folder = root / f"{when:%Y}" / f"{when:%m}" / f"{when:%d}"              # changed: dated subfolders
    folder.mkdir(parents=True, exist_ok=True)
    digest = sha256(path)
    dest = unique_destination(folder, path.name, digest)
    if dest.exists():
        path.unlink()                                                      # changed: duplicate, drop source
        return dest
    if path.stat().st_dev == folder.stat().st_dev:
        os.link(path, dest) if hasattr(os, "link") else shutil.copy2(path, dest)   # changed: no overwrite
        path.unlink()
        return dest
    tmp = dest.with_name(f".{dest.name}.part")
    shutil.copy2(path, tmp)                                                # changed: copy with metadata
    with tmp.open("rb") as fh:
        os.fsync(fh.fileno())
    if sha256(tmp) != digest:                                              # changed: verify before delete
        tmp.unlink()
        raise OSError(f"copy of {path.name} does not match source")
    os.replace(tmp, dest)                                                  # changed: atomic on destination FS
    path.unlink()                                                          # changed: delete source last
    return dest

if __name__ == "__main__":
    for item in sorted(Path("drop/incoming").glob("*.pdf")):
        try:
            print(item.name, "->", archive(item, Path("/mnt/archive/documents")))
        except OSError as exc:
            print(f"{item.name}: left in place ({exc})")

Two details make the same-device path safe. os.link creates the destination as a hard link and fails with FileExistsError if the name exists, so it can never overwrite — and removing the source link afterwards completes the move. And the content hash in the name makes collisions between different files effectively impossible while letting an identical file be recognised as already archived. Hard links are not supported on some network and FAT file systems; os.link raises there, and the cross-device branch — which also refuses to overwrite because the name is unique — handles it.

Safe cross-device archive move Six steps. Hash the source file. Choose a unique destination name in a dated folder that includes part of the hash, reusing it if an identical file is already archived. Copy the file to a hidden temporary name on the destination file system and flush it to disk. Hash the copy and compare with the source. Rename the temporary file into place atomically. Delete the source only after the rename succeeded. A crash at any earlier point leaves the source in place and at most a hidden temporary file. Hash source sha256 Unique name dated folder + hash Copy to .part fsync Verify hash copy equals source Rename into place os.replace Delete source last step

Variant Fix 1: Failed Files with a Reason

Files that fail processing need to leave the inbox too — otherwise the watcher retries them forever — but they must carry the reason, or someone has to reproduce the failure to find out what went wrong. Write a small JSON sidecar next to the moved file:

# stdlib only
import json
import traceback
from datetime import datetime, timezone
from pathlib import Path

def quarantine(path: Path, root: Path, error: BaseException, stage: str) -> Path:
    dest = archive(path, root)                                   # same safe move, different root
    sidecar = dest.with_suffix(dest.suffix + ".reason.json")
    sidecar.write_text(json.dumps({
        "original_name": path.name,
        "failed_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "stage": stage,
        "error_type": type(error).__name__,
        "error": str(error),
        "traceback": traceback.format_exception(error)[-3:],
    }, indent=2), encoding="utf-8")
    return dest

Recording the stage — extract, validate, write — makes the failed folder triageable at a glance, and grouping reasons by error_type each morning shows whether one supplier's layout change caused twenty failures or twenty different problems occurred. To retry after a fix, move files from the failed folder back to the inbox; the content hash ensures they are not duplicated if some had actually succeeded.

Variant Fix 2: Retention and Cleanup

Archives grow without limit unless something removes old files. Delete by the dated folder structure rather than by file modification time — copy2 preserves the original mtime, so a file archived today may carry last year's timestamp:

# stdlib only
import shutil
from datetime import date, timedelta
from pathlib import Path

def prune_archive(root: Path, keep_days: int, dry_run: bool = True) -> list[Path]:
    cutoff = date.today() - timedelta(days=keep_days)
    removed = []
    for day_dir in sorted(root.glob("[0-9][0-9][0-9][0-9]/[0-9][0-9]/[0-9][0-9]")):
        try:
            y, m, d = (int(part) for part in day_dir.parts[-3:])
            folder_date = date(y, m, d)
        except ValueError:
            continue                                      # not a dated folder
        if folder_date < cutoff:
            removed.append(day_dir)
            if not dry_run:
                shutil.rmtree(day_dir)
    return removed

Run pruning with dry_run=True and log the result for a week before enabling deletion. Retention periods for financial and HR documents are often set by regulation or policy; agree them with the document owner and keep failed files longer than successful ones, since they are the ones most likely to be asked about.

Archive layout on disk The archive root contains year, month and day folders so retention can prune whole days. Inside, processed files are named with the original stem, part of the content hash and the original extension. The failed root mirrors the dated structure and stores each failed file together with a reason json sidecar describing the stage and error. Hidden .part files only exist briefly during cross-device copies and indicate an interrupted move if they persist. archive/2026/09/17/ dated day folders prune by folder date, not mtime invoice__3fa91c2e7b.pdf stem + hash + extension identical content recognised by hash failed/2026/09/17/ same dated structure kept longer than processed files scan__8b10d4a79c.pdf.reason.json stage, error, traceback read by the morning triage report .invoice__3fa91c2e7b.pdf.part only during a copy if it persists, a move was interrupted

Recovering After a Crash

Because the source is deleted only after the destination is complete, a crash leaves at most two artefacts: the source still in the inbox, and possibly a hidden .part file in the archive. A startup routine removes stale partial files and lets the normal pipeline pick the source up again:

# stdlib only
import time
from pathlib import Path

def clean_partial_copies(root: Path, older_than: float = 3600.0) -> int:
    removed = 0
    now = time.time()
    for part in root.rglob(".*.part"):
        if now - part.stat().st_mtime > older_than:
            part.unlink(missing_ok=True)
            removed += 1
    return removed

Only remove partial files older than the longest plausible copy; a partial file younger than that may belong to a move still in progress in another process. With processing state keyed on content hash, re-running a file that was already processed but not yet moved simply skips the processing and completes the move.

Verification

Test the three failure modes directly: a name collision with different content, a duplicate with identical content, and a copy that is interrupted.

# stdlib only
import tempfile
from pathlib import Path
from unittest import mock

def test_archive() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        tmp = Path(tmp)
        inbox, root = tmp / "in", tmp / "archive"
        inbox.mkdir()
        (inbox / "invoice.pdf").write_bytes(b"%PDF-A")
        first = archive(inbox / "invoice.pdf", root)
        (inbox / "invoice.pdf").write_bytes(b"%PDF-B")             # same name, different content
        second = archive(inbox / "invoice.pdf", root)
        assert first != second and first.read_bytes() == b"%PDF-A", "collision overwrote an archived file"
        (inbox / "invoice.pdf").write_bytes(b"%PDF-A")             # identical content again
        third = archive(inbox / "invoice.pdf", root)
        assert third == first and not (inbox / "invoice.pdf").exists(), "duplicate not recognised"
        (inbox / "report.pdf").write_bytes(b"%PDF-C" * 1000)
        with mock.patch("os.link", side_effect=OSError("disk full")):   # the move itself fails
            try:
                archive(inbox / "report.pdf", root)
            except OSError:
                pass
        assert (inbox / "report.pdf").exists(), "source deleted although the move failed"
    print("archive moves: no overwrite, duplicates recognised, source kept on failure")

if __name__ == "__main__":
    test_archive()

The interrupted-move case patches os.link to fail the way a full disk would; run the same test against a real second mount to exercise the cross-device branch. The assertion that matters is that the source file still exists. Losing the source on a failed move is the one outcome an archive step must never produce.

FAQ

Is shutil.move good enough on a single disk? It performs a rename and overwrites existing destinations on POSIX. With unique names it is acceptable; without them it loses files silently.

Why not keep the original filename and add a counter? Counters require listing the folder under concurrency and still collide when two workers pick the same number. Content hashes are deterministic and need no coordination.

Should I preserve timestamps with copy2? Yes, for audit — the original modification time is often the only record of when a scanner produced the file. Just do not use it for retention decisions.

How do I make the archive read-only? Remove write permission after the move (dest.chmod(0o444)) or use storage with object lock or WORM retention for regulated records.

Part of Watching Folders for Incoming Documents.