def check_type_tag(tag: str) -> None:
def check_min_version(version: str) -> None:
def _make_argparser(filenames_help: str) -> argparse.ArgumentParser:
def validate_manifest_main(argv: Sequence[str] | None = None) -> int:
    def check(self, dct: dict[str, Any]) -> None:
    def check(self, dct: dict[str, Any]) -> None:
    def check(self, dct: dict[str, Any]) -> None:
    def _cond(key: str) -> cfgv.Conditional:
    def check(self, dct: dict[str, Any]) -> None:
    def apply_default(self, dct: dict[str, Any]) -> None:
def _entry(modname: str) -> str:
def warn_unknown_keys_root(
        extra: Sequence[str],
        orig_keys: Sequence[str],
        dct: dict[str, str],
) -> None:
def warn_unknown_keys_repo(
        extra: Sequence[str],
        orig_keys: Sequence[str],
        dct: dict[str, str],
) -> None:
    def check(self, dct: dict[str, Any]) -> None:
def ordered_load_normalize_legacy_config(contents: str) -> dict[str, Any]:
def validate_config_main(argv: Sequence[str] | None = None) -> int:
MANIFEST_HOOK_DICT = cfgv.Map(
    'Hook', 'id',

    cfgv.Required('id', cfgv.check_string),
    cfgv.Required('name', cfgv.check_string),
    cfgv.Required('entry', cfgv.check_string),
    cfgv.Required('language', cfgv.check_one_of(all_languages)),
    cfgv.Optional('alias', cfgv.check_string, ''),

    cfgv.Optional('files', check_string_regex, ''),
    cfgv.Optional('exclude', check_string_regex, '^$'),
    cfgv.Optional('types', cfgv.check_array(check_type_tag), ['file']),
    cfgv.Optional('types_or', cfgv.check_array(check_type_tag), []),
    cfgv.Optional('exclude_types', cfgv.check_array(check_type_tag), []),

    cfgv.Optional(
        'additional_dependencies', cfgv.check_array(cfgv.check_string), [],
    ),
    cfgv.Optional('args', cfgv.check_array(cfgv.check_string), []),
    cfgv.Optional('always_run', cfgv.check_bool, False),
    cfgv.Optional('fail_fast', cfgv.check_bool, False),
    cfgv.Optional('pass_filenames', cfgv.check_bool, True),
    cfgv.Optional('description', cfgv.check_string, ''),
    cfgv.Optional('language_version', cfgv.check_string, C.DEFAULT),
    cfgv.Optional('log_file', cfgv.check_string, ''),
    cfgv.Optional('minimum_pre_commit_version', cfgv.check_string, '0'),
    cfgv.Optional('require_serial', cfgv.check_bool, False),
    cfgv.Optional('stages', cfgv.check_array(cfgv.check_one_of(C.STAGES)), []),
    cfgv.Optional('verbose', cfgv.check_bool, False),
)
MANIFEST_SCHEMA = cfgv.Array(MANIFEST_HOOK_DICT)
LOCAL = 'local'
META = 'meta'
META_HOOK_DICT = cfgv.Map(
    'Hook', 'id',
    cfgv.Required('id', cfgv.check_string),
    cfgv.Required('id', cfgv.check_one_of(tuple(k for k, _ in _meta))),
    # language must be system
    cfgv.Optional('language', cfgv.check_one_of({'system'}), 'system'),
    # entry cannot be overridden
    NotAllowed('entry', cfgv.check_any),
    *(
        # default to the hook definition for the meta hooks
        cfgv.ConditionalOptional(key, cfgv.check_any, value, 'id', hook_id)
        for hook_id, values in _meta
        for key, value in values
    ),
    *(
        # default to the "manifest" parsing
        cfgv.OptionalNoDefault(item.key, item.check_fn)
        # these will always be defaulted above
        if item.key in {'name', 'language', 'entry'} else
        item
        for item in MANIFEST_HOOK_DICT.items
    ),
)
CONFIG_HOOK_DICT = cfgv.Map(
    'Hook', 'id',

    cfgv.Required('id', cfgv.check_string),

    # All keys in manifest hook dict are valid in a config hook dict, but
    # are optional.
    # No defaults are provided here as the config is merged on top of the
    # manifest.
    *(
        cfgv.OptionalNoDefault(item.key, item.check_fn)
        for item in MANIFEST_HOOK_DICT.items
        if item.key != 'id'
    ),
    OptionalSensibleRegexAtHook('files', cfgv.check_string),
    OptionalSensibleRegexAtHook('exclude', cfgv.check_string),
)
CONFIG_REPO_DICT = cfgv.Map(
    'Repository', 'repo',

    cfgv.Required('repo', cfgv.check_string),

    cfgv.ConditionalRecurse(
        'hooks', cfgv.Array(CONFIG_HOOK_DICT),
        'repo', cfgv.NotIn(LOCAL, META),
    ),
    cfgv.ConditionalRecurse(
        'hooks', cfgv.Array(MANIFEST_HOOK_DICT),
        'repo', LOCAL,
    ),
    cfgv.ConditionalRecurse(
        'hooks', cfgv.Array(META_HOOK_DICT),
        'repo', META,
    ),

    MigrateShaToRev(),
    WarnMutableRev(
        'rev',
        cfgv.check_string,
        '',
        'repo',
        cfgv.NotIn(LOCAL, META),
        True,
    ),
    cfgv.WarnAdditionalKeys(('repo', 'rev', 'hooks'), warn_unknown_keys_repo),
)
DEFAULT_LANGUAGE_VERSION = cfgv.Map(
    'DefaultLanguageVersion', None,
    cfgv.NoAdditionalKeys(all_languages),
    *(cfgv.Optional(x, cfgv.check_string, C.DEFAULT) for x in all_languages),
)
CONFIG_SCHEMA = cfgv.Map(
    'Config', None,

    cfgv.RequiredRecurse('repos', cfgv.Array(CONFIG_REPO_DICT)),
    cfgv.OptionalRecurse(
        'default_language_version', DEFAULT_LANGUAGE_VERSION, {},
    ),
    cfgv.Optional(
        'default_stages',
        cfgv.check_array(cfgv.check_one_of(C.STAGES)),
        C.STAGES,
    ),
    cfgv.Optional('files', check_string_regex, ''),
    cfgv.Optional('exclude', check_string_regex, '^$'),
    cfgv.Optional('fail_fast', cfgv.check_bool, False),
    cfgv.Optional(
        'minimum_pre_commit_version',
        cfgv.check_and(cfgv.check_string, check_min_version),
        '0',
    ),
    cfgv.WarnAdditionalKeys(
        (
            'repos',
            'default_language_version',
            'default_stages',
            'files',
            'exclude',
            'fail_fast',
            'minimum_pre_commit_version',
            'ci',
        ),
        warn_unknown_keys_root,
    ),
    OptionalSensibleRegexAtTop('files', cfgv.check_string),
    OptionalSensibleRegexAtTop('exclude', cfgv.check_string),

    # do not warn about configuration for pre-commit.ci
    cfgv.OptionalNoDefault('ci', cfgv.check_type(dict)),
)
class InvalidManifestError(FatalError):
class WarnMutableRev(cfgv.ConditionalOptional):
class OptionalSensibleRegexAtHook(cfgv.OptionalNoDefault):
class OptionalSensibleRegexAtTop(cfgv.OptionalNoDefault):
class MigrateShaToRev:
class NotAllowed(cfgv.OptionalNoDefault):
class InvalidConfigError(FatalError): LOCAL = 'local' META = 'meta' class Hook(NamedTuple):
    src: str
    prefix: Prefix
    id: str
    name: str
    entry: str
    language: str
    alias: str
    files: str
    exclude: str
    types: Sequence[str]
    types_or: Sequence[str]
    exclude_types: Sequence[str]
    additional_dependencies: Sequence[str]
    args: Sequence[str]
    always_run: bool
    fail_fast: bool
    pass_filenames: bool
    description: str
    language_version: str
    log_file: str
    minimum_pre_commit_version: str
    require_serial: bool
    stages: Sequence[str]
    verbose: bool

    @property
    def cmd(self) -> tuple[str, ...]:
        return (*shlex.split(self.entry), *self.args)

    @property
    def install_key(self) -> tuple[Prefix, str, str, tuple[str, ...]]:
        return (
            self.prefix,
            self.language,
            self.language_version,
            tuple(self.additional_dependencies),
        )

    @classmethod
    def create(cls, src: str, prefix: Prefix, dct: dict[str, Any]) -> Hook:
        # TODO: have cfgv do this (?)
        extra_keys = set(dct) - _KEYS
        if extra_keys:
            logger.warning(
                f'Unexpected key(s) present on {src} => {dct["id"]}: '
                f'{", ".join(sorted(extra_keys))}',
            )
        return cls(src=src, prefix=prefix, **{k: dct[k] for k in _KEYS}) class Language(NamedTuple):
    ENVIRONMENT_DIR: str | None @overload
def environment_dir(d: None, language_version: str) -> None: ... class Prefix(NamedTuple):
    prefix_dir: str

    def path(self, *parts: str) -> str:
        return os.path.normpath(os.path.join(self.prefix_dir, *parts))

    def exists(self, *parts: str) -> bool:
        return os.path.exists(self.path(*parts))

    def star(self, end: str) -> tuple[str, ...]:
        paths = os.listdir(self.prefix_dir)
        return tuple(path for path in paths if path.endswith(end)) class Store:
    get_default_directory = staticmethod(_get_default_directory)

    def __init__(self, directory: str | None = None) -> None:
        self.directory = directory or Store.get_default_directory()
        self.db_path = os.path.join(self.directory, 'db.db')
        self.readonly = (
            os.path.exists(self.directory) and
            not os.access(self.directory, os.W_OK)
        )

        if not os.path.exists(self.directory):
            os.makedirs(self.directory, exist_ok=True)
            with open(os.path.join(self.directory, 'README'), 'w') as f:
                f.write(
                    'This directory is maintained by the pre-commit project.\n'
                    'Learn more: https://github.com/pre-commit/pre-commit\n',
                )

        if os.path.exists(self.db_path):
            return
        with self.exclusive_lock():
            # Another process may have already completed this work
            if os.path.exists(self.db_path):  # pragma: no cover (race)
                return
            # To avoid a race where someone ^Cs between db creation and
            # execution of the CREATE TABLE statement
            fd, tmpfile = tempfile.mkstemp(dir=self.directory)
            # We'll be managing this file ourselves
            os.close(fd)
            with self.connect(db_path=tmpfile) as db:
                db.executescript(
                    'CREATE TABLE repos ('
                    '    repo TEXT NOT NULL,'
                    '    ref TEXT NOT NULL,'
                    '    path TEXT NOT NULL,'
                    '    PRIMARY KEY (repo, ref)'
                    ');',
                )
                self._create_config_table(db)

            # Atomic file move
            os.replace(tmpfile, self.db_path)

    @contextlib.contextmanager
    def exclusive_lock(self) -> Generator[None, None, None]:
        def blocked_cb() -> None:  # pragma: no cover (tests are in-process)
            logger.info('Locking pre-commit directory')

        with file_lock.lock(os.path.join(self.directory, '.lock'), blocked_cb):
            yield

    @contextlib.contextmanager
    def connect(
            self,
            db_path: str | None = None,
    ) -> Generator[sqlite3.Connection, None, None]:
        db_path = db_path or self.db_path
        # sqlite doesn't close its fd with its contextmanager >.<
        # contextlib.closing fixes this.
        # See: https://stackoverflow.com/a/28032829/812183
        with contextlib.closing(sqlite3.connect(db_path)) as db:
            # this creates a transaction
            with db:
                yield db

    @classmethod
    def db_repo_name(cls, repo: str, deps: Sequence[str]) -> str:
        if deps:
            return f'{repo}:{",".join(sorted(deps))}'
        else:
            return repo

    def _new_repo(
            self,
            repo: str,
            ref: str,
            deps: Sequence[str],
            make_strategy: Callable[[str], None],
    ) -> str:
        repo = self.db_repo_name(repo, deps)

        def _get_result() -> str | None:
            # Check if we already exist
            with self.connect() as db:
                result = db.execute(
                    'SELECT path FROM repos WHERE repo = ? AND ref = ?',
                    (repo, ref),
                ).fetchone()
                return result[0] if result else None

        result = _get_result()
        if result:
            return result
        with self.exclusive_lock():
            # Another process may have already completed this work
            result = _get_result()
            if result:  # pragma: no cover (race)
                return result

            logger.info(f'Initializing environment for {repo}.')

            directory = tempfile.mkdtemp(prefix='repo', dir=self.directory)
            with clean_path_on_failure(directory):
                make_strategy(directory)

            # Update our db with the created repo
            with self.connect() as db:
                db.execute(
                    'INSERT INTO repos (repo, ref, path) VALUES (?, ?, ?)',
                    [repo, ref, directory],
                )
        return directory

    def _complete_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:
        """Perform a complete clone of a repository and its submodules """

        git_cmd('fetch', 'origin', '--tags')
        git_cmd('checkout', ref)
        git_cmd('submodule', 'update', '--init', '--recursive')

    def _shallow_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:
        """Perform a shallow clone of a repository and its submodules """

        git_config = 'protocol.version=2'
        git_cmd('-c', git_config, 'fetch', 'origin', ref, '--depth=1')
        git_cmd('checkout', 'FETCH_HEAD')
        git_cmd(
            '-c', git_config, 'submodule', 'update', '--init', '--recursive',
            '--depth=1',
        )

    def clone(self, repo: str, ref: str, deps: Sequence[str] = ()) -> str:
        """Clone the given url and checkout the specific ref."""

        def clone_strategy(directory: str) -> None:
            git.init_repo(directory, repo)
            env = git.no_git_env()

            def _git_cmd(*args: str) -> None:
                cmd_output_b('git', *args, cwd=directory, env=env)

            try:
                self._shallow_clone(ref, _git_cmd)
            except CalledProcessError:
                self._complete_clone(ref, _git_cmd)

        return self._new_repo(repo, ref, deps, clone_strategy)

    LOCAL_RESOURCES = (
        'Cargo.toml', 'main.go', 'go.mod', 'main.rs', '.npmignore',
        'package.json', 'pre-commit-package-dev-1.rockspec',
        'pre_commit_placeholder_package.gemspec', 'setup.py',
        'environment.yml', 'Makefile.PL', 'pubspec.yaml',
        'renv.lock', 'renv/activate.R', 'renv/LICENSE.renv',
    )

    def make_local(self, deps: Sequence[str]) -> str:
        def make_local_strategy(directory: str) -> None:
            for resource in self.LOCAL_RESOURCES:
                resource_dirname, resource_basename = os.path.split(resource)
                contents = resource_text(f'empty_template_{resource_basename}')
                target_dir = os.path.join(directory, resource_dirname)
                target_file = os.path.join(target_dir, resource_basename)
                os.makedirs(target_dir, exist_ok=True)
                with open(target_file, 'w') as f:
                    f.write(contents)

            env = git.no_git_env()

            # initialize the git repository so it looks more like cloned repos
            def _git_cmd(*args: str) -> None:
                cmd_output_b('git', *args, cwd=directory, env=env)

            git.init_repo(directory, '<<unknown>>')
            _git_cmd('add', '.')
            git.commit(repo=directory)

        return self._new_repo(
            'local', C.LOCAL_REPO_VERSION, deps, make_local_strategy,
        )

    def _create_config_table(self, db: sqlite3.Connection) -> None:
        db.executescript(
            'CREATE TABLE IF NOT EXISTS configs ('
            '   path TEXT NOT NULL,'
            '   PRIMARY KEY (path)'
            ');',
        )

    def mark_config_used(self, path: str) -> None:
        if self.readonly:  # pragma: win32 no cover
            return
        path = os.path.realpath(path)
        # don't insert config files that do not exist
        if not os.path.exists(path):
            return
        with self.connect() as db:
            # TODO: eventually remove this and only create in _create
            self._create_config_table(db)
            db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))

    def select_all_configs(self) -> list[str]:
        with self.connect() as db:
            self._create_config_table(db)
            rows = db.execute('SELECT path FROM configs').fetchall()
            return [path for path, in rows]

    def delete_configs(self, configs: list[str]) -> None:
        with self.connect() as db:
            rows = [(path,) for path in configs]
            db.executemany('DELETE FROM configs WHERE path = ?', rows)

    def select_all_repos(self) -> list[tuple[str, str, str]]:
        with self.connect() as db:
            return db.execute('SELECT repo, ref, path from repos').fetchall()

    def delete_repo(self, db_repo_name: str, ref: str, path: str) -> None:
        with self.connect() as db:
            db.execute(
                'DELETE FROM repos WHERE repo = ? and ref = ?',
                (db_repo_name, ref),
            )
        rmtree(path) def parse_version(s: str) -> tuple[int, ...]:
    """poor man's version comparison"""
    return tuple(int(p) for p in s.split('.')) def rmtree(path: str) -> None:
    """On windows, rmtree fails for readonly dirs."""
    def handle_remove_readonly(
            func: Callable[..., Any],
            path: str,
            exc: tuple[type[OSError], OSError, TracebackType],
    ) -> None:
        excvalue = exc[1]
        if (
                func in (os.rmdir, os.remove, os.unlink) and
                excvalue.errno in {errno.EACCES, errno.EPERM}
        ):
            for p in (path, os.path.dirname(path)):
                os.chmod(p, os.stat(p).st_mode | stat.S_IWUSR)
            func(path)
        else:
            raise
    shutil.rmtree(path, ignore_errors=False, onerror=handle_remove_readonly)
import json
import logging
import os
import pre_commit.constants as C
from typing import Any
from typing import Sequence
from pre_commit.clientlib import load_manifest
from pre_commit.clientlib import LOCAL
from pre_commit.clientlib import META
from pre_commit.hook import Hook
from pre_commit.languages.all import languages
from pre_commit.languages.helpers import environment_dir
from pre_commit.prefix import Prefix
from pre_commit.store import Store
from pre_commit.util import parse_version
from pre_commit.util import rmtree
from __future__ import annotations




logger = logging.getLogger('pre_commit')


def _state(additional_deps: Sequence[str]) -> object:
    return {'additional_dependencies': sorted(additional_deps)}


def _state_filename(prefix: Prefix, venv: str) -> str:
    return prefix.path(venv, f'.install_state_v{C.INSTALLED_STATE_VERSION}')


def _read_state(prefix: Prefix, venv: str) -> object | None:
    filename = _state_filename(prefix, venv)
    if not os.path.exists(filename):
        return None
    else:
        with open(filename) as f:
            return json.load(f)


def _write_state(prefix: Prefix, venv: str, state: object) -> None:
    state_filename = _state_filename(prefix, venv)
    staging = f'{state_filename}staging'
    with open(staging, 'w') as state_file:
        state_file.write(json.dumps(state))
    # Move the file into place atomically to indicate we've installed
    os.replace(staging, state_filename)



def _hook_installed(hook: Hook) -> bool: