sort_downloads.py

#!/opt/homebrew/bin/uv run

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "arrow",
# ]
# ///

import errno
import filecmp
import os
import shutil
from subprocess import check_output
from typing import Optional

import arrow

IGNORE_FILES = [
    ".DS_Store",
    ".idata",
    ".localized",
    ".rdata",
    ".reloc",
    ".rsrc",
    ".rsrc_1",
    ".tls",
    "By week",
    "This week",
    "Last week",
]


def mkdirp(path):
    try:
        os.makedirs(path)
    except OSError as e:
        if e.errno == errno.EEXIST and os.path.isdir(path):
            pass
        else:
            raise


def files_match(a, b):
    if os.path.isdir(a) and os.path.isdir(b):
        return filecmp.dircmp(a, b).left_only == [] and filecmp.dircmp(a, b).right_only == [] and filecmp.dircmp(a, b).diff_files == []
    if os.path.isfile(a) and os.path.isfile(b):
        return filecmp.cmp(a, b, shallow=False)
    return False


def suffixed(destination, filename):
    base, ext = os.path.splitext(filename)
    n = 2
    while os.path.exists(os.path.join(destination, base + f" ({n})" + ext)):
        n += 1
    return base + f" ({n})" + ext


COMMAND_OPTIONS = [
    "mdls",
    "-raw",
    "-nullMarker",
    "",
    "-name",
    "kMDItemDateAdded",
]


def date_added(filename) -> Optional[arrow.Arrow]:
    try:
        added_string = check_output(
            [*COMMAND_OPTIONS, filename],
            universal_newlines=True,
        )
    except Exception:
        print("failed:", " ".join([*COMMAND_OPTIONS, filename]))
        return None

    try:
        return arrow.get(added_string, "YYYY-MM-DD HH:mm:ss Z")
    except arrow.parser.ParserMatchError:
        return None


def format_week(week):
    year, week, _ = week.isocalendar()

    return ("By week", "{}-{:02d}".format(year, week))


def link_week(downloads, week, title):
    week_path = os.path.join(downloads, *format_week(week))

    if not os.path.exists(week_path):
        mkdirp(week_path)

    link_path = os.path.join(downloads, title)

    if os.path.islink(link_path):
        # the link already points to the right place
        if os.readlink(link_path) == week_path:
            return

        os.unlink(link_path)

    os.symlink(week_path, link_path)


def sort_downloads():
    downloads = os.path.expanduser("~/Downloads")

    os.chdir(downloads)

    link_week(downloads, arrow.now(), "This week")
    link_week(downloads, arrow.now().shift(weeks=-1), "Last week")

    one_day_ago = arrow.now().shift(days=-1)

    for filename in os.listdir(downloads):
        if filename in IGNORE_FILES:
            continue

        file_date = date_added(filename)

        if not file_date:
            continue

        # file is not yet old enough to move
        if file_date > one_day_ago:
            continue

        destination = os.path.join(downloads, *format_week(file_date))

        if not os.path.exists(destination):
            mkdirp(destination)

        source = os.path.join(downloads, filename)
        dest_path = os.path.join(destination, filename)

        if os.path.exists(dest_path):
            if files_match(source, dest_path):
                if os.path.isdir(source) and not os.path.islink(source):
                    shutil.rmtree(source)
                else:
                    os.remove(source)
            else:
                shutil.move(source, os.path.join(destination, suffixed(destination, filename)))
            continue

        shutil.move(source, destination)


if __name__ == "__main__":
    sort_downloads()