<commit_before># Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Useful tools for various Paths used inside Airflow Sources.
"""
from __future__ import annotations

import hashlib
import os
import subprocess
import sys
import tempfile
from functools import lru_cache
from pathlib import Path

from airflow_breeze import NAME
from airflow_breeze.utils.confirm import set_forced_answer
from airflow_breeze.utils.console import get_console
from airflow_breeze.utils.reinstall import reinstall_breeze, warn_dependencies_changed, warn_non_editable

AIRFLOW_CFG_FILE = "setup.cfg"


def search_upwards_for_airflow_sources_root(start_from: Path) -> Path | None:
    root = Path(start_from.root)
    d = start_from
    while d != root:
        attempt = d / AIRFLOW_CFG_FILE
        if attempt.exists() and "name = apache-airflow\n" in attempt.read_text():
            return attempt.parent
        d = d.parent
    return None


def in_autocomplete() -> bool:
    return os.environ.get(f"_{NAME.upper()}_COMPLETE") is not None


def in_self_upgrade() -> bool:
    return "self-upgrade" in sys.argv


def in_help() -> bool:
    return "--help" in sys.argv


def skip_upgrade_check():
    return in_self_upgrade() or in_autocomplete() or in_help() or hasattr(sys, '_called_from_test')


def skip_group_output():
    return in_autocomplete() or in_help() or os.environ.get('SKIP_GROUP_OUTPUT') is not None


def get_package_setup_metadata_hash() -> str:
    """
    Retrieves hash of setup files from the source of installation of Breeze.

    This is used in order to determine if we need to upgrade Breeze, because some
    setup files changed. Blake2b algorithm will not be flagged by security checkers
    as insecure algorithm (in Python 3.9 and above we can use `usedforsecurity=False`
    to disable it, but for now it's better to use more secure algorithms.
    """
    # local imported to make sure that autocomplete works
    try:
        from importlib.metadata import distribution  # type: ignore[attr-defined]
    except ImportError:
        from importlib_metadata import distribution  # type: ignore[no-redef]

    prefix = "Package config hash: "

    for line in distribution('apache-airflow-breeze').metadata.as_string().splitlines(keepends=False):
        if line.startswith(prefix):
            return line[len(prefix) :]
    return "NOT FOUND"


def get_sources_setup_metadata_hash(sources: Path) -> str:
    try:
        the_hash = hashlib.new("blake2b")
        the_hash.update((sources / "dev" / "breeze" / "setup.py").read_bytes())
        the_hash.update((sources / "dev" / "breeze" / "setup.cfg").read_bytes())
        the_hash.update((sources / "dev" / "breeze" / "pyproject.toml").read_bytes())
        return the_hash.hexdigest()
    except FileNotFoundError as e:
        return f"Missing file {e.filename}"


def get_installation_sources_config_metadata_hash() -> str:
    """
    Retrieves hash of setup.py and setup.cfg files from the source of installation of Breeze.

    This is used in order to determine if we need to upgrade Breeze, because some
    setup files changed. Blake2b algorithm will not be flagged by security checkers
    as insecure algorithm (in Python 3.9 and above we can use `usedforsecurity=False`
    to disable it, but for now it's better to use more secure algorithms.
    """
    installation_sources = get_installation_airflow_sources()
    if installation_sources is None:
        return "NOT FOUND"
    return get_sources_setup_metadata_hash(installation_sources)


def get_used_sources_setup_metadata_hash() -> str:
    """
    Retrieves hash of setup files from the currently used sources.
    """
    return get_sources_setup_metadata_hash(get_used_airflow_sources())


def set_forced_answer_for_upgrade_check():
    """When we run upgrade check --answer is not parsed yet, so we need to guess it."""
    if "--answer n" in " ".join(sys.argv).lower() or os.environ.get('ANSWER', '').lower().startswith("n"):
        set_forced_answer("no")
    if "--answer y" in " ".join(sys.argv).lower() or os.environ.get('ANSWER', '').lower().startswith("y"):
        set_forced_answer("yes")
    if "--answer q" in " ".join(sys.argv).lower() or os.environ.get('ANSWER', '').lower().startswith("q"):
        set_forced_answer("quit")


def reinstall_if_setup_changed() -> bool:
    """
    Prints warning if detected airflow sources are not the ones that Breeze was installed with.
    :return: True if warning was printed.
    """
    try:
        package_hash = get_package_setup_metadata_hash()
    except ModuleNotFoundError as e:
        if "importlib_metadata" in e.msg:
            return False
    sources_hash = get_installation_sources_config_metadata_hash()
    if sources_hash != package_hash:
        installation_sources = get_installation_airflow_sources()
        if installation_sources is not None:
            breeze_sources = installation_sources / "dev" / "breeze"
            warn_dependencies_changed()
            set_forced_answer_for_upgrade_check()
            reinstall_breeze(breeze_sources)
            set_forced_answer(None)<commit_msg>Fix error on running pre-commit without installing breeze (#26985)<commit_after># Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Useful tools for various Paths used inside Airflow Sources.
"""
from __future__ import annotations

import hashlib
import os
import subprocess
import sys
import tempfile
from functools import lru_cache
from pathlib import Path

from airflow_breeze import NAME
from airflow_breeze.utils.confirm import set_forced_answer
from airflow_breeze.utils.console import get_console
from airflow_breeze.utils.reinstall import reinstall_breeze, warn_dependencies_changed, warn_non_editable

AIRFLOW_CFG_FILE = "setup.cfg"


def search_upwards_for_airflow_sources_root(start_from: Path) -> Path | None:
    root = Path(start_from.root)
    d = start_from
    while d != root:
        attempt = d / AIRFLOW_CFG_FILE
        if attempt.exists() and "name = apache-airflow\n" in attempt.read_text():
            return attempt.parent
        d = d.parent
    return None


def in_autocomplete() -> bool:
    return os.environ.get(f"_{NAME.upper()}_COMPLETE") is not None


def in_self_upgrade() -> bool:
    return "self-upgrade" in sys.argv


def in_help() -> bool:
    return "--help" in sys.argv


def skip_upgrade_check():
    return in_self_upgrade() or in_autocomplete() or in_help() or hasattr(sys, '_called_from_test')


def skip_group_output():
    return in_autocomplete() or in_help() or os.environ.get('SKIP_GROUP_OUTPUT') is not None


def get_package_setup_metadata_hash() -> str:
    """
    Retrieves hash of setup files from the source of installation of Breeze.

    This is used in order to determine if we need to upgrade Breeze, because some
    setup files changed. Blake2b algorithm will not be flagged by security checkers
    as insecure algorithm (in Python 3.9 and above we can use `usedforsecurity=False`
    to disable it, but for now it's better to use more secure algorithms.
    """
    # local imported to make sure that autocomplete works
    try:
        from importlib.metadata import distribution  # type: ignore[attr-defined]
    except ImportError:
        from importlib_metadata import distribution  # type: ignore[no-redef]

    prefix = "Package config hash: "

    for line in distribution('apache-airflow-breeze').metadata.as_string().splitlines(keepends=False):
        if line.startswith(prefix):
            return line[len(prefix) :]
    return "NOT FOUND"


def get_sources_setup_metadata_hash(sources: Path) -> str:
    try:
        the_hash = hashlib.new("blake2b")
        the_hash.update((sources / "dev" / "breeze" / "setup.py").read_bytes())
        the_hash.update((sources / "dev" / "breeze" / "setup.cfg").read_bytes())
        the_hash.update((sources / "dev" / "breeze" / "pyproject.toml").read_bytes())
        return the_hash.hexdigest()
    except FileNotFoundError as e:
        return f"Missing file {e.filename}"


def get_installation_sources_config_metadata_hash() -> str:
    """
    Retrieves hash of setup.py and setup.cfg files from the source of installation of Breeze.

    This is used in order to determine if we need to upgrade Breeze, because some
    setup files changed. Blake2b algorithm will not be flagged by security checkers
    as insecure algorithm (in Python 3.9 and above we can use `usedforsecurity=False`
    to disable it, but for now it's better to use more secure algorithms.
    """
    installation_sources = get_installation_airflow_sources()
    if installation_sources is None:
        return "NOT FOUND"
    return get_sources_setup_metadata_hash(installation_sources)


def get_used_sources_setup_metadata_hash() -> str:
    """
    Retrieves hash of setup files from the currently used sources.
    """
    return get_sources_setup_metadata_hash(get_used_airflow_sources())


def set_forced_answer_for_upgrade_check():
    """When we run upgrade check --answer is not parsed yet, so we need to guess it."""
    if "--answer n" in " ".join(sys.argv).lower() or os.environ.get('ANSWER', '').lower().startswith("n"):
        set_forced_answer("no")
    if "--answer y" in " ".join(sys.argv).lower() or os.environ.get('ANSWER', '').lower().startswith("y"):
        set_forced_answer("yes")
    if "--answer q" in " ".join(sys.argv).lower() or os.environ.get('ANSWER', '').lower().startswith("q"):
        set_forced_answer("quit")


def reinstall_if_setup_changed() -> bool:
    """
    Prints warning if detected airflow sources are not the ones that Breeze was installed with.
    :return: True if warning was printed.
    """
    try:
        package_hash = get_package_setup_metadata_hash()
    except ModuleNotFoundError as e:
        if "importlib_metadata" in e.msg:
            return False
        if "apache-airflow-breeze" in e.msg:
            print(
                """Missing Package `apache-airflow-breeze`.
                   Use `pipx install -e ./dev/breeze` to install the package."""
            )
            return False
    sources_hash = get_installation_sources_config_metadata_hash()
    if sources_hash != package_hash:
        installation_sources = get_installation_airflow_sources()
        if installation_sources is not None:
            breeze_sources = installation_sources / "dev" / "breeze"
            warn_dependencies_changed()
            set_forced_answer_for_upgrade_check()
            reinstall_breeze(breeze_sources)
            set_forced_answer(None)
{'fraction_non_alphanumeric': 0.08865927839346721, 'fraction_numerical': 0.002249812515623698, 'mean_word_length': 3.560030395136778, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7313957', 'n_tokens_mistral': 3517, 'n_tokens_neox': 3448, 'n_words': 1160}