# fileset.py - file set queries for mercurial

#

# Copyright 2010 Olivia Mackall <olivia@selenic.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

import errno

import re

from .i18n import _

from .pycompat import getattr

from . import (

    error,

    filesetlang,

    match as matchmod,

    mergestate as mergestatemod,

    pycompat,

    registrar,

    scmutil,

    util,

)

from .utils import stringutil

# common weight constants

_WEIGHT_CHECK_FILENAME = filesetlang.WEIGHT_CHECK_FILENAME

_WEIGHT_READ_CONTENTS = filesetlang.WEIGHT_READ_CONTENTS

_WEIGHT_STATUS = filesetlang.WEIGHT_STATUS

_WEIGHT_STATUS_THOROUGH = filesetlang.WEIGHT_STATUS_THOROUGH

# helpers for processing parsed tree

getsymbol = filesetlang.getsymbol

getstring = filesetlang.getstring

_getkindpat = filesetlang.getkindpat

getpattern = filesetlang.getpattern

getargs = filesetlang.getargs

def getmatch(mctx, x):

    if not x:

        raise error.ParseError(_(b"missing argument"))

    return methods[x[0]](mctx, *x[1:])

def getmatchwithstatus(mctx, x, hint):

    keys = set(getstring(hint, b'status hint must be a string').split())

    return getmatch(mctx.withstatus(keys), x)

def stringmatch(mctx, x):

    return mctx.matcher([x])

def kindpatmatch(mctx, x, y):

    return stringmatch(

        mctx,

        _getkindpat(

            x, y, matchmod.allpatternkinds, _(b"pattern must be a string")

        ),

    )

def patternsmatch(mctx, *xs):

    allkinds = matchmod.allpatternkinds

    patterns = [

        getpattern(x, allkinds, _(b"pattern must be a string")) for x in xs

    ]

    return mctx.matcher(patterns)

def andmatch(mctx, x, y):

    xm = getmatch(mctx, x)

    ym = getmatch(mctx.narrowed(xm), y)

    return matchmod.intersectmatchers(xm, ym)

def ormatch(mctx, *xs):

    ms = [getmatch(mctx, x) for x in xs]

    return matchmod.unionmatcher(ms)

def notmatch(mctx, x):

    m = getmatch(mctx, x)

    return mctx.predicate(lambda f: not m(f), predrepr=(b'<not %r>', m))

def minusmatch(mctx, x, y):

    xm = getmatch(mctx, x)

    ym = getmatch(mctx.narrowed(xm), y)

    return matchmod.differencematcher(xm, ym)

def listmatch(mctx, *xs):

    raise error.ParseError(

        _(b"can't use a list in this context"),

        hint=_(b'see \'hg help "filesets.x or y"\''),

    )

def func(mctx, a, b):

    funcname = getsymbol(a)

    if funcname in symbols:

        return symbols[funcname](mctx, b)

    keep = lambda fn: getattr(fn, '__doc__', None) is not None

    syms = [s for (s, fn) in symbols.items() if keep(fn)]

    raise error.UnknownIdentifier(funcname, syms)

# symbols are callable like:

#  fun(mctx, x)

# with:

#  mctx - current matchctx instance

#  x - argument in tree form

symbols = filesetlang.symbols

predicate = registrar.filesetpredicate(symbols)

@predicate(b'modified()', callstatus=True, weight=_WEIGHT_STATUS)

def modified(mctx, x):

    """File that is modified according to :hg:`status`."""

    # i18n: "modified" is a keyword

    getargs(x, 0, 0, _(b"modified takes no arguments"))

    s = set(mctx.status().modified)

    return mctx.predicate(s.__contains__, predrepr=b'modified')

@predicate(b'added()', callstatus=True, weight=_WEIGHT_STATUS)

def added(mctx, x):

    """File that is added according to :hg:`status`."""

    # i18n: "added" is a keyword

    getargs(x, 0, 0, _(b"added takes no arguments"))

    s = set(mctx.status().added)

    return mctx.predicate(s.__contains__, predrepr=b'added')

@predicate(b'removed()', callstatus=True, weight=_WEIGHT_STATUS)

def removed(mctx, x):

    """File that is removed according to :hg:`status`."""

    # i18n: "removed" is a keyword

    getargs(x, 0, 0, _(b"removed takes no arguments"))

    s = set(mctx.status().removed)

    return mctx.predicate(s.__contains__, predrepr=b'removed')

@predicate(b'deleted()', callstatus=True, weight=_WEIGHT_STATUS)

def deleted(mctx, x):

    """Alias for ``missing()``."""

    # i18n: "deleted" is a keyword

    getargs(x, 0, 0, _(b"deleted takes no arguments"))

    s = set(mctx.status().deleted)

    return mctx.predicate(s.__contains__, predrepr=b'deleted')

@predicate(b'missing()', callstatus=True, weight=_WEIGHT_STATUS)

def missing(mctx, x):

    """File that is missing according to :hg:`status`."""

    # i18n: "missing" is a keyword

    getargs(x, 0, 0, _(b"missing takes no arguments"))

    s = set(mctx.status().deleted)

    return mctx.predicate(s.__contains__, predrepr=b'deleted')

@predicate(b'unknown()', callstatus=True, weight=_WEIGHT_STATUS_THOROUGH)

def unknown(mctx, x):

    """File that is unknown according to :hg:`status`."""

    # i18n: "unknown" is a keyword

    getargs(x, 0, 0, _(b"unknown takes no arguments"))

    s = set(mctx.status().unknown)

    return mctx.predicate(s.__contains__, predrepr=b'unknown')

@predicate(b'ignored()', callstatus=True, weight=_WEIGHT_STATUS_THOROUGH)

def ignored(mctx, x):

    """File that is ignored according to :hg:`status`."""

    # i18n: "ignored" is a keyword

    getargs(x, 0, 0, _(b"ignored takes no arguments"))

    s = set(mctx.status().ignored)

    return mctx.predicate(s.__contains__, predrepr=b'ignored')

@predicate(b'clean()', callstatus=True, weight=_WEIGHT_STATUS)

def clean(mctx, x):

    """File that is clean according to :hg:`status`."""

    # i18n: "clean" is a keyword

    getargs(x, 0, 0, _(b"clean takes no arguments"))

    s = set(mctx.status().clean)

    return mctx.predicate(s.__contains__, predrepr=b'clean')

@predicate(b'tracked()')

def tracked(mctx, x):

    """File that is under Mercurial control."""

    # i18n: "tracked" is a keyword

    getargs(x, 0, 0, _(b"tracked takes no arguments"))

    return mctx.predicate(mctx.ctx.__contains__, predrepr=b'tracked')

@predicate(b'binary()', weight=_WEIGHT_READ_CONTENTS)

def binary(mctx, x):

    """File that appears to be binary (contains NUL bytes)."""

    # i18n: "binary" is a keyword

    getargs(x, 0, 0, _(b"binary takes no arguments"))

    return mctx.fpredicate(

        lambda fctx: fctx.isbinary(), predrepr=b'binary', cache=True

    )

@predicate(b'exec()')

def exec_(mctx, x):

    """File that is marked as executable."""

    # i18n: "exec" is a keyword

    getargs(x, 0, 0, _(b"exec takes no arguments"))

    ctx = mctx.ctx

    return mctx.predicate(lambda f: ctx.flags(f) == b'x', predrepr=b'exec')

@predicate(b'symlink()')

def symlink(mctx, x):

    """File that is marked as a symlink."""

    # i18n: "symlink" is a keyword

    getargs(x, 0, 0, _(b"symlink takes no arguments"))

    ctx = mctx.ctx

    return mctx.predicate(lambda f: ctx.flags(f) == b'l', predrepr=b'symlink')

@predicate(b'resolved()', weight=_WEIGHT_STATUS)

def resolved(mctx, x):

    """File that is marked resolved according to :hg:`resolve -l`."""

    # i18n: "resolved" is a keyword

    getargs(x, 0, 0, _(b"resolved takes no arguments"))

    if mctx.ctx.rev() is not None:

        return mctx.never()

    ms = mergestatemod.mergestate.read(mctx.ctx.repo())

    return mctx.predicate(

        lambda f: f in ms and ms[f] == b'r', predrepr=b'resolved'

    )

@predicate(b'unresolved()', weight=_WEIGHT_STATUS)

def unresolved(mctx, x):

    """File that is marked unresolved according to :hg:`resolve -l`."""

    # i18n: "unresolved" is a keyword

    getargs(x, 0, 0, _(b"unresolved takes no arguments"))

    if mctx.ctx.rev() is not None:

        return mctx.never()

    ms = mergestatemod.mergestate.read(mctx.ctx.repo())

    return mctx.predicate(

        lambda f: f in ms and ms[f] == b'u', predrepr=b'unresolved'

    )

@predicate(b'hgignore()', weight=_WEIGHT_STATUS)

def hgignore(mctx, x):

    """File that matches the active .hgignore pattern."""

    # i18n: "hgignore" is a keyword

    getargs(x, 0, 0, _(b"hgignore takes no arguments"))

    return mctx.ctx.repo().dirstate._ignore

@predicate(b'portable()', weight=_WEIGHT_CHECK_FILENAME)

def portable(mctx, x):

    """File that has a portable name. (This doesn't include filenames with case

    collisions.)

    """

    # i18n: "portable" is a keyword

    getargs(x, 0, 0, _(b"portable takes no arguments"))

    return mctx.predicate(

        lambda f: util.checkwinfilename(f) is None, predrepr=b'portable'

    )

@predicate(b'grep(regex)', weight=_WEIGHT_READ_CONTENTS)

def grep(mctx, x):

    """File contains the given regular expression."""

    try:

        # i18n: "grep" is a keyword

        r = re.compile(getstring(x, _(b"grep requires a pattern")))

    except re.error as e:

        raise error.ParseError(

            _(b'invalid match pattern: %s') % stringutil.forcebytestr(e)

        )

    return mctx.fpredicate(

        lambda fctx: r.search(fctx.data()),

        predrepr=(b'grep(%r)', r.pattern),

        cache=True,

    )

def _sizetomax(s):

    try:

        s = s.strip().lower()

        for k, v in util._sizeunits:

            if s.endswith(k):

                # max(4k) = 5k - 1, max(4.5k) = 4.6k - 1

                n = s[: -len(k)]

                inc = 1.0

                if b"." in n:

                    inc /= 10 ** len(n.split(b".")[1])

                return int((float(n) + inc) * v) - 1

        # no extension, this is a precise value

        return int(s)

    except ValueError:

        raise error.ParseError(_(b"couldn't parse size: %s") % s)

def sizematcher(expr):

    """Return a function(size) -> bool from the ``size()`` expression"""

    expr = expr.strip()

    if b'-' in expr:  # do we have a range?

        a, b = expr.split(b'-', 1)

        a = util.sizetoint(a)

        b = util.sizetoint(b)

        return lambda x: x >= a and x <= b

    elif expr.startswith(b"<="):

        a = util.sizetoint(expr[2:])

        return lambda x: x <= a

    elif expr.startswith(b"<"):

        a = util.sizetoint(expr[1:])

        return lambda x: x < a

    elif expr.startswith(b">="):

        a = util.sizetoint(expr[2:])

        return lambda x: x >= a

    elif expr.startswith(b">"):

        a = util.sizetoint(expr[1:])

        return lambda x: x > a

    else:

        a = util.sizetoint(expr)

        b = _sizetomax(expr)

        return lambda x: x >= a and x <= b

@predicate(b'size(expression)', weight=_WEIGHT_STATUS)

def size(mctx, x):

    """File size matches the given expression. Examples:

    - size('1k') - files from 1024 to 2047 bytes

    - size('< 20k') - files less than 20480 bytes

    - size('>= .5MB') - files at least 524288 bytes

    - size('4k - 1MB') - files from 4096 bytes to 1048576 bytes

    """

    # i18n: "size" is a keyword

    expr = getstring(x, _(b"size requires an expression"))

    m = sizematcher(expr)

    return mctx.fpredicate(

        lambda fctx: m(fctx.size()), predrepr=(b'size(%r)', expr), cache=True

    )

@predicate(b'encoding(name)', weight=_WEIGHT_READ_CONTENTS)

def encoding(mctx, x):

    """File can be successfully decoded with the given character

    encoding. May not be useful for encodings other than ASCII and

    UTF-8.