Filters#

class Created(TimeFilter):
    """Matches files / folders by created date

    Attributes:
        years (int): specify number of years
        months (int): specify number of months
        weeks (float): specify number of weeks
        days (float): specify number of days
        hours (float): specify number of hours
        minutes (float): specify number of minutes
        seconds (float): specify number of seconds
        mode (str):
            either 'older' or 'newer'. 'older' matches files / folders created before
            the given time, 'newer' matches files / folders created within the given
            time. (default = 'older')

    Returns:
        `{created}` (datetime): The datetime the file / folder was created.
    """

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="created",
        files=True,
        dirs=True,
    )

    def get_datetime(self, path: Path) -> datetime:
        return read_created(path)

class DateAdded(TimeFilter):

    """Matches files by the time the file was added to a folder.

    **`date_added` is only available on macOS!**

    Attributes:
        years (int): specify number of years
        months (int): specify number of months
        weeks (float): specify number of weeks
        days (float): specify number of days
        hours (float): specify number of hours
        minutes (float): specify number of minutes
        seconds (float): specify number of seconds
        mode (str):
            either 'older' or 'newer'. 'older' matches files / folders last modified before
            the given time, 'newer' matches files / folders last modified within the given
            time. (default = 'older')

    Returns:
        `{date_added}`: The datetime the files / folders were added.
    """

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="date_added",
        files=True,
        dirs=True,
    )

    def __post_init__(self):
        if sys.platform != "darwin":
            raise EnvironmentError("date_added is only available on macOS")
        return super().__post_init__()

    def get_datetime(self, path: Path) -> datetime:
        return read_date_added(path)

class DateLastUsed(TimeFilter):

    """Matches files by the time the file was last used.

    **`date_lastused` is only available on macOS!**

    Attributes:
        years (int): specify number of years
        months (int): specify number of months
        weeks (float): specify number of weeks
        days (float): specify number of days
        hours (float): specify number of hours
        minutes (float): specify number of minutes
        seconds (float): specify number of seconds
        mode (str):
            either 'older' or 'newer'. 'older' matches files / folders last used before
            the given time, 'newer' matches files / folders last used within the given
            time. (default = 'older')

    Returns:
        {date_lastused}: The datetime the files / folders were added.
    """

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="date_lastused",
        files=True,
        dirs=True,
    )

    def __post_init__(self):
        if sys.platform != "darwin":
            raise EnvironmentError("date_added is only available on macOS")
        return super().__post_init__()

    def get_datetime(self, path: Path) -> datetime:
        return read_date_lastused(path)

@dataclass(config=ConfigDict(extra="forbid"))
class Duplicate:
    """A fast duplicate file finder.

    This filter compares files byte by byte and finds identical files with potentially
    different filenames.

    Attributes:
        detect_original_by (str):
            Detection method to distinguish between original and duplicate.
            Possible values are:

            - `"first_seen"`: Whatever file is visited first is the original. This
              depends on the order of your location entries.
            - `"name"`: The first entry sorted by name is the original.
            - `"created"`: The first entry sorted by creation date is the original.
            - `"lastmodified"`: The first file sorted by date of last modification is
               the original.

    You can reverse the sorting method by prefixing a `-`.

    So with `detect_original_by: "-created"` the file with the older creation date is
    the original and the younger file is the duplicate. This works on all methods, for
    example `"-first_seen"`, `"-name"`, `"-created"`, `"-lastmodified"`.

    **Returns:**

    `{duplicate.original}` - The path to the original
    """

    detect_original_by: DetectionMethod = "first_seen"
    hash_algorithm: str = "sha1"

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="duplicate", files=True, dirs=False
    )

    def __post_init__(self):
        # reverse original detection order if starting with "-"
        self._detect_original_by = self.detect_original_by
        self._detect_original_reverse = False
        if self.detect_original_by.startswith("-"):
            self._detect_original_by = self.detect_original_by[1:]
            self._detect_original_reverse = True

        self._files_for_size = defaultdict(list)
        self._files_for_chunk = defaultdict(list)
        self._file_for_hash = dict()

        # we keep track of the files we already computed the hashes for so we only do
        # that once.
        self._seen_files = set()
        self._first_chunk_known = set()
        self._hash_known = set()

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None, "Does not support standalone mode"
        # skip symlinks
        if res.path.is_symlink():
            return False

        # the exact same path has already been handled. This happens if multiple
        # locations emit this file in a single rule or if we follow symlinks.
        # We skip these.
        if res.path in self._seen_files:
            return False

        self._seen_files.add(res.path)

        # check for files with equal size
        file_size = read_file_size(path=res.path)
        same_size = self._files_for_size[file_size]
        same_size.append(res.path)
        if len(same_size) == 1:
            # the file is unique in size and cannot be a duplicate
            return False

        # for all other files with the same file size:
        # make sure we know their hash of their first 1024 byte chunk
        for f in same_size[:-1]:
            if f not in self._first_chunk_known:
                chunk_hash = hash_first_chunk(f, algo=self.hash_algorithm)
                self._first_chunk_known.add(f)
                self._files_for_chunk[chunk_hash].append(f)

        # check first chunk hash collisions with the current file
        chunk_hash = hash_first_chunk(res.path, algo=self.hash_algorithm)
        same_first_chunk = self._files_for_chunk[chunk_hash]
        same_first_chunk.append(res.path)
        self._first_chunk_known.add(res.path)
        if len(same_first_chunk) == 1:
            # the file has a unique small hash and cannot be a duplicate
            return False

        # Ensure we know the full hashes of all files with the same first chunk as
        # the investigated file
        for f in same_first_chunk[:-1]:
            if f not in self._hash_known:
                hash_ = hash(f, algo=self.hash_algorithm)
                self._hash_known.add(f)
                self._file_for_hash[hash_] = f

        # check full hash collisions with the current file
        hash_ = hash(res.path, algo=self.hash_algorithm)
        self._hash_known.add(res.path)
        known = self._file_for_hash.get(hash_)
        if known:
            original, duplicate = detect_original(
                known=known,
                new=res.path,
                method=self._detect_original_by,
                reverse=self._detect_original_reverse,
            )
            if known != original:
                self._file_for_hash[hash_] = original

            res.path = duplicate
            res.vars[self.filter_config.name] = {"original": original}
            return True

        return False

@dataclass(config=ConfigDict(extra="forbid"))
class Empty:

    """Finds empty dirs and files"""

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="empty",
        files=True,
        dirs=True,
    )

    def pipeline(self, res: Resource, output: Output) -> bool:
        return res.is_empty()

class Exif(BaseModel):
    """Filter by image EXIF data

    The `exif` filter can be used as a filter as well as a way to get exif information
    into your actions.

    By default this library uses the `exifread` library. If your image format is not
    supported you can install `exiftool` (exiftool.org) and set the environment variable:

    ```
    ORGANIZE_EXIFTOOL_PATH="exiftool"
    ```

    organize will then use `exiftool` to extract the EXIF data.

    Exif fields which contain "datetime", "date" or "offsettime" in their fieldname
    will have their value converted to 'datetime.datetime', 'datetime.date' and
    'datetime.timedelta' respectivly.
    - `datetime.datetime` : exif.image.datetime, exif.exif.datetimeoriginal, ...
    - `datetime.date` : exif.gps.date, ...
    - `datetime.timedelta` : exif.exif.offsettimeoriginal, exif.exif.offsettimedigitized, ...

    Attributes:
        lowercase_keys (bool): Whether to lowercase all EXIF keys (Default: true)

    :returns:
        ``{exif}`` -- a dict of all the collected exif inforamtion available in the
        file. Typically it consists of the following tags (if present in the file):

        - ``{exif.image}`` -- information related to the main image
        - ``{exif.exif}`` -- Exif information
        - ``{exif.gps}`` -- GPS information
        - ``{exif.interoperability}`` -- Interoperability information
    """

    filter_tags: Dict
    lowercase_keys: bool = True

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="exif",
        files=True,
        dirs=False,
    )

    def __init__(
        self,
        *args,
        filter_tags: Optional[Dict] = None,
        lowercase_keys: bool = True,
        **kwargs,
    ):
        # exif filter is used differently from other filters. The **kwargs are not
        # filter parameters but all belong into the filter_tags dictionary to filter
        # for specific exif tags.
        params = filter_tags or dict()
        params.update(kwargs)
        # *args are tags filtered without a value, like ["gps", "image.model"].
        for arg in args:
            params[arg] = None
        super().__init__(filter_tags=params, lowercase_keys=lowercase_keys)

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None, "Does not support standalone mode"

        # gather the exif data in a dict
        if exiftool_available():
            data = exiftool_read(path=res.path)
        else:
            data = exifread_read(path=res.path)

        # lowercase keys if wanted
        if self.lowercase_keys:
            data = lowercase_keys_recursive(data)

        # convert strings to datetime objects where possible
        parsed = convert_recursive(data)

        res.vars[self.filter_config.name] = parsed
        return matches_tags(self.filter_tags, data)

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class Extension:
    """Filter by file extension

    Attributes:
        *extensions (list(str) or str):
            The file extensions to match (does not need to start with a colon).

    **Returns:**

    - `{extension}`: the original file extension (without colon)
    """

    extensions: Set[str] = Field(default_factory=set)

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="extension",
        files=True,
        dirs=False,
    )

    @field_validator("extensions", mode="before")
    def normalize_extensions(cls, v):
        as_list = convert_to_list(v)
        return set(map(normalize_extension, flatten(list(as_list))))

    def suffix_match(self, path: Path) -> Tuple[str, bool]:
        suffix = path.suffix.lstrip(".")
        if not self.extensions:
            return (suffix, True)
        if not suffix:
            return (suffix, False)
        return (suffix, normalize_extension(suffix) in self.extensions)

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None, "Does not support standalone mode"
        if res.is_dir():
            raise ValueError("Dirs not supported")

        suffix, match = self.suffix_match(path=res.path)
        res.vars[self.filter_config.name] = suffix
        return match

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class FileContent:
    """Matches file content with the given regular expression.

    Supports .md, .txt, .log, .pdf and .docx files.

    For PDF content extraction poppler should be installed for the `pdftotext` command.
    If this is not available `filecontent` will fall back to the `pdfminer` library.

    Attributes:
        expr (str): The regular expression to be matched.

    Any named groups (`(?P<groupname>.*)`) in your regular expression will
    be returned like this:

    **Returns:**

    - `{filecontent.groupname}`: The text matched with the named group
      `(?P<groupname>)`

    You can test the filter on any file by running:

    ```sh
    python -m organize.filters.filecontent "/path/to/file.pdf"
    ```
    """

    expr: str = r"(?P<all>.*)"

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="filecontent",
        files=True,
        dirs=False,
    )

    def __post_init__(self):
        self._expr = re.compile(self.expr, re.MULTILINE | re.DOTALL)

    def matches(self, path: Path) -> Union[re.Match, None]:
        try:
            content = textract(path)
            match = self._expr.search(content)
            return match
        except Exception:
            return None

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None, "Does not support standalone mode"
        match = self.matches(path=res.path)

        if match:
            res.deep_merge(self.filter_config.name, match.groupdict())
        return bool(match)

@dataclass(config=ConfigDict(extra="forbid"))
class Hash:

    """Calculates the hash of a file.

    Attributes:
        algorithm (str): Any hashing algorithm available to python's `hashlib`.
            `md5` by default.

    Algorithms guaranteed to be available are
    `shake_256`, `sha3_256`, `sha1`, `sha3_224`, `sha384`, `sha512`, `blake2b`,
    `blake2s`, `sha256`, `sha224`, `shake_128`, `sha3_512`, `sha3_384` and `md5`.

    Depending on your python installation and installed libs there may be additional
    hash algorithms to chose from.

    To list the available algorithms on your installation run this in a python
    interpreter:

    ```py
    >>> import hashlib
    >>> hashlib.algorithms_available
    {'shake_256', 'whirlpool', 'mdc2', 'blake2s', 'sha224', 'shake_128', 'sha3_512',
    'sha3_224', 'sha384', 'md5', 'sha1', 'sha512_256', 'blake2b', 'sha256',
    'sha512_224', 'ripemd160', 'sha3_384', 'md4', 'sm3', 'sha3_256', 'md5-sha1',
    'sha512'}
    ```

    **Returns:**

    - `{hash}`:  The hash of the file.
    """

    algorithm: str = "md5"

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="hash",
        files=True,
        dirs=False,
    )

    def __post_init__(self):
        self._algorithm = Template.from_string(self.algorithm)

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None
        algo = render(self._algorithm, res.dict()).lower()
        result = hash(path=res.path, algo=algo)
        res.vars[self.filter_config.name] = result
        return True

class LastModified(TimeFilter):

    """Matches files by last modified date

    Attributes:
        years (int): specify number of years
        months (int): specify number of months
        weeks (float): specify number of weeks
        days (float): specify number of days
        hours (float): specify number of hours
        minutes (float): specify number of minutes
        seconds (float): specify number of seconds
        mode (str):
            either 'older' or 'newer'. 'older' matches files / folders last modified before
            the given time, 'newer' matches files / folders last modified within the given
            time. (default = 'older')

    Returns:
        {lastmodified}: The datetime the files / folders was lastmodified.
    """

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="lastmodified",
        files=True,
        dirs=True,
    )

    def get_datetime(self, path: Path) -> datetime:
        return read_lastmodified(path)

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class MacOSTags:
    """Filter by macOS tags

    Attributes:
        tags (list(str) or str):
            The tags to filter by
    """

    tags: List[str] = Field(default_factory=list)

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="macos_tags",
        files=True,
        dirs=True,
    )

    def __post_init__(self):
        if sys.platform != "darwin":
            raise EnvironmentError("The macos_tags filter is only available on macOS")

    @field_validator("tags", mode="before")
    def ensure_list(cls, v):
        if isinstance(v, str):
            return [v]
        return v

    def pipeline(self, res: Resource, output: Output) -> bool:
        file_tags = list_tags(res.path)
        res.vars[self.filter_config.name] = file_tags
        return matches_tags(filter_tags=self.tags, file_tags=file_tags)

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class MimeType:

    """Filter by MIME type associated with the file extension.

    Supports a single string or list of MIME type strings as argument.
    The types don't need to be fully specified, for example "audio" matches everything
    from "audio/midi" to "audio/quicktime".

    You can see a list of known MIME types on your system by running this oneliner:

    ```sh
    python3 -m organize.filters.mimetype
    ```

    Attributes:
        *mimetypes (list(str) or str): The MIME types to filter for.

    **Returns:**

    - `{mimetype}`: The MIME type of the file.
    """

    mimetypes: FlatList[str] = Field(default_factory=list)

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="mimetype",
        files=True,
        dirs=False,
    )

    def matches(self, mimetype) -> bool:
        if mimetype is None:
            return False
        if not self.mimetypes:
            return True
        return any(mimetype.startswith(x) for x in self.mimetypes)

    def pipeline(self, res: Resource, output: Output) -> bool:
        mimetype = guess_mimetype(res.path)
        res.vars[self.filter_config.name] = mimetype
        return self.matches(mimetype)

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class Name:
    """Match files and folders by name

    Attributes:
        match (str):
            A matching string in [simplematch-syntax](https://github.com/tfeldmann/simplematch)

        startswith (str):
            The filename must begin with the given string

        contains (str):
            The filename must contain the given string

        endswith (str):
            The filename (without extension) must end with the given string

        case_sensitive (bool):
            By default, the matching is case sensitive. Change this to False to use
            case insensitive matching.
    """

    match: str = "*"
    startswith: Union[str, List[str]] = ""
    contains: Union[str, List[str]] = ""
    endswith: Union[str, List[str]] = ""
    case_sensitive: bool = True

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="name",
        files=True,
        dirs=True,
    )

    def __post_init__(self, *args, **kwargs):
        self._matcher = simplematch.Matcher(
            self.match,
            case_sensitive=self.case_sensitive,
        )
        self.startswith = self.create_list(self.startswith, self.case_sensitive)
        self.contains = self.create_list(self.contains, self.case_sensitive)
        self.endswith = self.create_list(self.endswith, self.case_sensitive)

    def matches(self, name: str) -> bool:
        if not self.case_sensitive:
            name = name.lower()

        is_match = (
            self._matcher.test(name)
            and any(x in name for x in self.contains)
            and any(name.startswith(x) for x in self.startswith)
            and any(name.endswith(x) for x in self.endswith)
        )
        return is_match

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None, "Does not support standalone mode"
        if res.is_dir():
            name = res.path.stem
        else:
            name, ext = res.path.stem, res.path.suffix
            if not name:
                name = ext
        result = self.matches(normalize_unicode(name))
        m = self._matcher.match(normalize_unicode(name))
        if not m:
            m = name

        res.vars[self.filter_config.name] = m
        return result

    @staticmethod
    def create_list(x: Union[int, str, List[Any]], case_sensitive: bool) -> List[str]:
        if isinstance(x, (int, float)):
            x = str(x)
        if isinstance(x, str):
            x = [x]
        x = [str(x) for x in x]
        if not case_sensitive:
            x = [x.lower() for x in x]
        return x

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class Python:

    """Use python code to filter files.

    Attributes:
        code (str):
            The python code to execute. The code must contain a `return` statement.


    **Returns:**

    - If your code returns `False` or `None` the file is filtered out,
      otherwise the file is passed on to the next filters.
    - `{python}` contains the returned value. If you return a dictionary (for
      example `return {"some_key": some_value, "nested": {"k": 2}}`) it will be
      accessible via dot syntax actions: `{python.some_key}`, `{python.nested.k}`.
    - Variables of previous filters are available, but you have to use the normal python
      dictionary syntax `x = regex["my_group"]`.
    """

    code: str

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="python",
        files=True,
        dirs=True,
    )

    @field_validator("code", mode="after")
    @classmethod
    def must_have_return_statement(cls, value):
        if "return" not in value:
            raise ValueError("No return statement found in your code!")
        return value

    def __post_init__(self):
        self.code = textwrap.dedent(self.code)

    def __usercode__(self, print, **kwargs) -> Optional[Dict]:
        raise NotImplementedError()

    def pipeline(self, res: Resource, output: Output) -> bool:
        def _output_msg(*values, sep: str = " ", end: str = ""):
            """
            the print function for the use code needs to print via the current output
            """
            output.msg(
                res=res,
                msg=f"{sep.join(str(x) for x in values)}{end}",
                sender="python",
            )

        # codegen the user function with arguments as available in the resource
        kwargs = ", ".join(res.dict().keys())
        func = f"def __userfunc(print, {kwargs}):\n"
        func += textwrap.indent(self.code, "    ")
        func += "\n\nself.__usercode__ = __userfunc"

        exec(func, globals().copy(), locals().copy())
        result = self.__usercode__(print=_output_msg, **res.dict())

        if isinstance(result, dict):
            res.deep_merge(key=self.filter_config.name, data=result)
        else:
            res.vars[self.filter_config.name] = result
        return result not in (False, None)

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class Regex:

    """Matches filenames with the given regular expression

    Attributes:
        expr (str): The regular expression to be matched.

    **Returns:**

    Any named groups in your regular expression will be returned like this:

    - `{regex.groupname}`: The text matched with the named
      group `(?P<groupname>.*)`

    """

    expr: str

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="regex",
        files=True,
        dirs=True,
    )

    def __post_init__(self):
        self._expr = re.compile(self.expr, flags=re.UNICODE)

    def matches(self, path: str):
        return self._expr.search(path)

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None, "Does not support standalone mode"
        match = self.matches(normalize_unicode(res.path.name))
        if match:
            res.deep_merge(key=self.filter_config.name, data=match.groupdict())
            return True
        return False

@dataclass(config=ConfigDict(coerce_numbers_to_str=True, extra="forbid"))
class Size:
    """Matches files and folders by size

    Attributes:
        *conditions (list(str) or str):
            The size constraints.

    Accepts file size conditions, e.g: `">= 500 MB"`, `"< 20k"`, `">0"`,
    `"= 10 KiB"`.

    It is possible to define both lower and upper conditions like this:
    `">20k, < 1 TB"`, `">= 20 Mb, <25 Mb"`. The filter will match if all given
    conditions are satisfied.

    - Accepts all units from KB to YB.
    - If no unit is given, kilobytes are assumend.
    - If binary prefix is given (KiB, GiB) the size is calculated using base 1024.

    **Returns:**

    - `{size.bytes}`: (int) Size in bytes
    - `{size.traditional}`: (str) Size with unit (powers of 1024, JDEC prefixes)
    - `{size.binary}`: (str) Size with unit (powers of 1024, IEC prefixes)
    - `{size.decimal}`: (str) Size with unit (powers of 1000, SI prefixes)
    """

    conditions: FlatList[str] = Field(default_factory=list)

    filter_config: ClassVar[FilterConfig] = FilterConfig(
        name="size", files=True, dirs=True
    )

    def __post_init__(self):
        self._constraints = set()
        for x in self.conditions:
            for constraint in create_constraints(x):
                self._constraints.add(constraint)

    def matches(self, filesize: int) -> bool:
        if not self._constraints:
            return True
        return all(op(filesize, c_size) for op, c_size in self._constraints)

    def pipeline(self, res: Resource, output: Output) -> bool:
        assert res.path is not None
        bytes = read_resource_size(res=res)
        res.vars[self.filter_config.name] = {
            "bytes": bytes,
            "traditional": traditional(bytes),
            "binary": binary(bytes),
            "decimal": decimal(bytes),
        }
        return self.matches(bytes)