List Unique Files

uniq_file.py

from collections import deque
from pathlib import Path
from queue import Queue
from dataclasses import dataclass, field
from typing import List

import concurrent.futures
import xxhash

import os
import sys

@dataclass
class _UniqFile:
    name: str
    hvalue: str
    count: int = 0
    occurrences: List[str] = field(default_factory = list)


class UniqFile:
    def __init__(self, directory, max_count=2):
        self.q, self.d, self.max_count = deque([]), directory, int(max_count)
        self.pool = concurrent.futures.ThreadPoolExecutor(max_workers=os.cpu_count() - 1)
        self.queue = Queue()

    def _analyze(self, filename, chunk_size=65536):
        if filename.stat().st_size == 0:
            return

        hasher = xxhash.xxh3_64()
        with open(filename, "rb") as file:
            # print("reading file", filename)
            chunk = file.read(chunk_size)
            while chunk:
                hasher.update(chunk)
                chunk = file.read(chunk_size)

        self.queue.put((hasher.hexdigest(), str(filename)))

    def _analyse_result(self):
        fmap = {}
        while not self.queue.empty():
            entry = self.queue.get()
            hvalue, filename = entry
            if hvalue in fmap:
                file = fmap[hvalue]
                file.count += 1
                file.occurrences.append(filename)
            else:
                file = _UniqFile(name=filename.split("/")[-1], hvalue=hvalue, count=1)
                file.occurrences.append(filename)
                fmap[hvalue] = file

        sorted_by_count = sorted(fmap.values(), key=lambda x: x.count, reverse=True)
        for i in sorted_by_count:
            if i.count > self.max_count:
                print("\n")
                print(i.name, i.count)
                print("    ->", ", ".join(i.occurrences))

    def list(self):
        root, visited = Path(self.d), set()
        if not root.is_dir():
            raise ValueError(f"{self.d} is not a directory")

        self.q.append(root)

        while self.q:
            item = self.q.popleft()
            if item.resolve() in visited:
                continue

            try:
                visited.add(item.resolve())
                for item in item.iterdir():
                    if item.is_file():
                        self.pool.submit(self._analyze, item)
                    elif item.is_dir():
                        self.q.append(item)
            except Exception as e:
                print(e)

        self.pool.shutdown(wait=True)
        self._analyse_result()


if len(sys.argv) < 2:
    print("Please provide path to directory as script argument")
    exit(1)

if __name__ == "__main__":
    processor = UniqFile(sys.argv[1].strip(), sys.argv[2].strip())
    processor.list()