# 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.

"""
Common helper module for CI/CD Steps
"""
import asyncio
import logging
import os
import urllib.parse
from pathlib import PurePath
from typing import List, Optional, Dict
from api.v1 import api_pb2

import pydantic
import yaml

from api.v1.api_pb2 import (
    SDK_UNSPECIFIED,
    STATUS_UNSPECIFIED,
    Sdk,
    STATUS_VALIDATING,
    STATUS_PREPARING,
    STATUS_COMPILING,
    STATUS_EXECUTING,
    PRECOMPILED_OBJECT_TYPE_UNIT_TEST,
    PRECOMPILED_OBJECT_TYPE_KATA,
    PRECOMPILED_OBJECT_TYPE_UNSPECIFIED,
    PRECOMPILED_OBJECT_TYPE_EXAMPLE,
    PrecompiledObjectType,
)
from config import Config, TagFields, PrecompiledExampleType
from grpc_client import GRPCClient
from constants import BEAM_ROOT_DIR_ENV_VAR_KEY

from models import Example, Tag, SdkEnum, Dataset


def _check_no_nested(subdirs: List[str]):
    """
    Check there're no nested subdirs

    Sort alphabetically and compare the pairs of adjacent items
    using pathlib.PurePath: we don't want fs calls in this check
    """
    sorted_subdirs = sorted(PurePath(s) for s in subdirs)
    for dir1, dir2 in zip(sorted_subdirs, sorted_subdirs[1:]):
        if dir1 in [dir2, *dir2.parents]:
            raise ValueError(f"{dir2} is a subdirectory of {dir1}")


def find_examples(root_dir: str, subdirs: List[str], sdk: SdkEnum) -> List[Example]:
    """
    Find and return beam examples.

    Search throws all child files of work_dir directory files with beam tag:
    Beam-playground:
        name: NameOfExample
        description: Description of NameOfExample.
        multifile: false
        default_example: false
        context_line: 10
        categories:
            - category-1
            - category-2
        pipeline_options: --inputFile your_file --outputFile your_output_file
        complexity: MEDIUM
        tags:
            - example
    If some example contains beam tag with incorrect format raise an error.

    Args:
        root_dir: project root dir
        subdirs: sub-directories where to search examples.
        sdk: sdk that using to find examples for the specific sdk.

    Returns:
        List of Examples.
    """
    has_errors = False
    examples = []
    _check_no_nested(subdirs)
    for subdir in subdirs:
        subdir = os.path.join(root_dir, subdir)
        logging.info("subdir: %s", subdir)
        for root, _, files in os.walk(subdir):
            for filename in files:
                filepath = os.path.join(root, filename)
                try:
                    try:
                        example = _load_example(
                            filename=filename, filepath=filepath, sdk=sdk
                        )
                        if example is not None:
                            examples.append(example)
                    except pydantic.ValidationError as err:
                        if len(err.errors()) > 1:
                            raise
                        if err.errors()[0]["msg"] == "multifile is True but no files defined":
                            logging.warning("incomplete multifile example ignored %s", filepath)
                            continue
                        raise
                except Exception:
                    logging.exception("error loading example at %s", filepath)
                    has_errors = True
    if has_errors:
        raise ValueError(
            "Some of the beam examples contain beam playground tag with "
            "an incorrect format"
        )
    return examples


def get_tag(filepath: PurePath) -> Optional[Tag]:
    """
    Parse file by filepath and find beam tag

    Args:
        filepath: path of the file

    Returns:
        If file contains tag, returns Tag object
        If file doesn't contain tag, returns None
    """
    with open(filepath, encoding="utf-8") as parsed_file:
        lines = parsed_file.readlines()

    line_start: Optional[int] = None
    line_finish: Optional[int] = None
    tag_prefix: Optional[str] = ""
    for idx, line in enumerate(lines):
        if line_start is None and line.endswith(Config.BEAM_PLAYGROUND_TITLE):
            line_start = idx
            prefix_len = len(line) - len(Config.BEAM_PLAYGROUND_TITLE)
            tag_prefix = line[:prefix_len]
        elif line_start and not line.startswith(tag_prefix):
            line_finish = idx
            break

    if not line_start or not line_finish:
        return None

    embdedded_yaml_content = "".join(
        line[len(tag_prefix) :] for line in lines[line_start:line_finish]
    )
    yml = yaml.load(embdedded_yaml_content, Loader=yaml.SafeLoader)

    try:
        return Tag(
            filepath=str(filepath),
            line_start=line_start,
            line_finish=line_finish,
            **yml[Config.BEAM_PLAYGROUND],
            )
    except pydantic.ValidationError as err:
        if len(err.errors()) == 1 and err.errors()[0]["msg"] == "multifile is True but no files defined":
            logging.warning("incomplete multifile example ignored %s", filepath)
            return None
        raise

def _load_example(filename, filepath, sdk: SdkEnum) -> Optional[Example]:
    """
    Check file by filepath for matching to beam example. If file is beam example,

    Args:
        filename: name of the file.
        filepath: path to the file.
        sdk: sdk that using to find examples for the specific sdk.

    Returns:
        If the file is an example, return Example object
        If it's not, return None
        In case of error, raise Exception
    """
    logging.debug("inspecting file %s", filepath)
    extension = filepath.split(os.extsep)[-1]
    if extension == Config.SDK_TO_EXTENSION[sdk]:
        logging.debug("sdk %s matched extension %s", api_pb2.Sdk.Name(sdk), extension)
        tag = get_tag(filepath)
        if tag is not None:
            logging.debug("playground-beam tag found")
            return _get_example(filepath, filename, tag, sdk)
    return None


# Make load_supported_categories called only once
# to make testing easier
_load_supported_categories = False


def load_supported_categories(categories_path: str):
    """
    Load the list of supported categories from categories_path file
    into Tag model config

    Args:
        categories_path: path to the file with categories.
    """
    global _load_supported_categories
    if _load_supported_categories:
        return
    with open(categories_path, encoding="utf-8") as supported_categories:
        yaml_object = yaml.load(supported_categories.read(), Loader=yaml.SafeLoader)

    Tag.Config.supported_categories = yaml_object[TagFields.categories]
    _load_supported_categories = True


def _get_content(filepath: str, tag_start_line: int, tag_finish_line) -> str:
    with open(filepath, encoding="utf-8") as parsed_file:
        lines = parsed_file.readlines()
        lines = lines[:tag_start_line] + lines[tag_finish_line:]
    return "".join(lines)


def _get_url_vcs(filepath: str) -> str:
    """
    Construct VCS URL from example's filepath
    """
    root_dir = os.getenv(BEAM_ROOT_DIR_ENV_VAR_KEY, "../..")
    rel_path = os.path.relpath(filepath, root_dir)
    url_vcs = "{}/{}".format(Config.URL_VCS_PREFIX, urllib.parse.quote(rel_path))
    return url_vcs


def _get_example(filepath: str, filename: str, tag: Tag, sdk: int) -> Example:
    """
    Return an Example by filepath and filename.

    Args:
         filepath: path of the example's file.
         filename: name of the example's file.
         tag: tag of the example.

    Returns:
        Parsed Example object.
    """

    # Calculate context line with tag removed. Note: context_line is 1-based, line_start and line_finish are 0-based.
    context_line = tag.context_line if tag.context_line <= tag.line_start else tag.context_line - (tag.line_finish - tag.line_start)
    
    return Example(
        sdk=SdkEnum(sdk),
        tag=tag,
        filepath=filepath,
        status=STATUS_UNSPECIFIED,
        type=_get_object_type(filename, filepath),
        code=_get_content(filepath, tag.line_start, tag.line_finish),
        url_vcs=_get_url_vcs(filepath),  # type: ignore
        context_line=context_line,
    )


async def update_example_status(example: Example, client: GRPCClient):
    """
    Receive status for examples and update example.status and pipeline_id

    Use client to send requests to the backend:
    1. Start code processing.
    2. Ping the backend while status is STATUS_VALIDATING/
      STATUS_PREPARING/STATUS_COMPILING/STATUS_EXECUTING
    Update example.status with resulting status.

    Args:
        example: beam example for processing and updating status and pipeline_id.
        client: client to send requests to the server.
    """
    datasets: List[api_pb2.Dataset] = []
    for emulator in example.tag.emulators:
        dataset: Dataset = example.tag.datasets[emulator.topic.source_dataset]

        datasets.append(
            api_pb2.Dataset(
                type=api_pb2.EmulatorType.Value(
                    f"EMULATOR_TYPE_{emulator.type.upper()}"
                ),
                options={"topic": emulator.topic.id},
                dataset_path=dataset.file_name,
            )
        )
    files: List[api_pb2.SnippetFile] = [
        api_pb2.SnippetFile(name=example.filepath, content=example.code, is_main=True)
    ]
    for file in example.tag.files:
        files.append(
            api_pb2.SnippetFile(name=file.name, content=file.content, is_main=False)
        )

    pipeline_id = await client.run_code(
        example.code, example.sdk, example.tag.pipeline_options, datasets, files=files,
    )
    example.pipeline_id = pipeline_id
    status = await client.check_status(pipeline_id)
    while status in [
        STATUS_VALIDATING,
        STATUS_PREPARING,
        STATUS_COMPILING,
        STATUS_EXECUTING,
    ]:
        await asyncio.sleep(Config.PAUSE_DELAY)
        status = await client.check_status(pipeline_id)
    example.status = status


def _get_object_type(filename, filepath):
    """
    Get type of an object based on it filename/filepath

    Args:
        filename: object's filename
        filepath: object's filepath

    Returns: type of the object (example, kata, unit-test)
    """
    filename_no_ext = (os.path.splitext(filename)[0]).lower()
    if filename_no_ext.endswith(PrecompiledExampleType.test_ends):
        object_type = PRECOMPILED_OBJECT_TYPE_UNIT_TEST
    elif PrecompiledExampleType.katas in filepath.split(os.sep):
        object_type = PRECOMPILED_OBJECT_TYPE_KATA
    elif PrecompiledExampleType.examples in filepath.split(os.sep):
        object_type = PRECOMPILED_OBJECT_TYPE_EXAMPLE
    else:
        object_type = PRECOMPILED_OBJECT_TYPE_UNSPECIFIED
    return object_type


class DuplicatesError(Exception):
    pass


class ConflictingDatasetsError(Exception):
    pass


def validate_examples_for_duplicates_by_name(examples: List[Example]):
    """
    Validate examples for duplicates by example name to avoid duplicates in the Cloud Datastore
    :param examples: examples from the repository for saving to the Cloud Datastore
    """
    duplicates: Dict[str, Example] = {}
    for example in examples:
        if example.tag.name not in duplicates.keys():
            duplicates[example.tag.name] = example
        else:
            err_msg = f"Examples have duplicate names.\nDuplicates: \n - path #1: {duplicates[example.tag.name].filepath} \n - path #2: {example.filepath}"
            logging.error(err_msg)
            raise DuplicatesError(err_msg)


def validate_examples_for_conflicting_datasets(examples: List[Example]):
    """
    Validate examples for conflicting datasets to avoid conflicts in the Cloud Datastore
    :param examples: examples from the repository for saving to the Cloud Datastore
    """
    datasets: Dict[str, Dataset] = {}
    for example in examples:
        for k, v in example.tag.datasets.items():
            if k not in datasets:
                datasets[k] = v
            elif datasets[k].file_name != v.file_name or \
                    datasets[k].format != v.format or \
                    datasets[k].location != v.location:
                err_msg = f"Examples have conflicting datasets.\n" \
                          f"Conflicts: \n" \
                          f" - file_name #1: {datasets[k].file_name} \n" \
                          f" - format #1: {datasets[k].format} \n" \
                          f"  - location #1: {datasets[k].location} \n" \
                          f" - file_name #2: {v.file_name}\n" \
                          f" - format #2: {v.format}\n" \
                          f" - location #2: {v.location}\n" \
                          f"Dataset name: {k}"
                logging.error(err_msg)
                raise ConflictingDatasetsError(err_msg)