#!/usr/bin/env python3
# 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.

import logging
import os
import re
import subprocess
import textwrap
from collections import defaultdict
from typing import Any, Dict, List, NamedTuple, Optional, Set, Union

import click
from github import Github, Issue, PullRequest, UnknownObjectException
from rich.console import Console
from rich.progress import Progress

logger = logging.getLogger(__name__)

console = Console(width=400, color_system="standard")

PullRequestOrIssue = Union[PullRequest.PullRequest, Issue.Issue]

MY_DIR_PATH = os.path.dirname(__file__)
SOURCE_DIR_PATH = os.path.abspath(os.path.join(MY_DIR_PATH, os.pardir))
PR_PATTERN = re.compile(r".*\(#([0-9]+)\)")
ISSUE_MATCH_IN_BODY = re.compile(r" #([0-9]+)[^0-9]")


@click.group(context_settings={'help_option_names': ['-h', '--help'], 'max_content_width': 500})
def cli():
    ...


option_verbose = click.option(
    "--verbose",
    is_flag=True,
    help="Print verbose information about performed steps",
)

option_previous_release = click.option(
    "--previous-release",
    type=str,
    required=True,
    help="commit reference (for example hash or tag) of the previous release.",
)

option_current_release = click.option(
    "--current-release",
    type=str,
    required=True,
    help="commit reference (for example hash or tag) of the current release.",
)

option_github_token = click.option(
    "--github-token",
    type=str,
    required=True,
    help=textwrap.dedent(
        """
        Github token used to authenticate.
        You can set omit it if you have GITHUB_TOKEN env variable set
        Can be generated with:
        https://github.com/settings/tokens/new?description=Read%20sssues&scopes=repo:status"""
    ),
    envvar='GITHUB_TOKEN',
)

option_excluded_pr_list = click.option(
    "--excluded-pr-list", type=str, default='', help="Coma-separated list of PRs to exclude from the issue."
)

option_limit_pr_count = click.option(
    "--limit-pr-count",
    type=int,
    default=None,
    help="Limit PR count processes (useful for testing small subset of PRs).",
)


def get_git_log_command(
    verbose: bool, from_commit: Optional[str] = None, to_commit: Optional[str] = None
) -> List[str]:
    """
    Get git command to run for the current repo from the current folder (which is the package folder).
    :param verbose: whether to print verbose info while getting the command
    :param from_commit: if present - base commit from which to start the log from
    :param to_commit: if present - final commit which should be the start of the log
    :return: git command to run
    """
    git_cmd = [
        "git",
        "log",
        "--pretty=format:%H %h %cd %s",
        "--date=short",
    ]
    if from_commit and to_commit:
        git_cmd.append(f"{from_commit}...{to_commit}")
    elif from_commit:
        git_cmd.append(from_commit)
    git_cmd.extend(['--', '.'])
    if verbose:
        console.print(f"Command to run: '{' '.join(git_cmd)}'")
    return git_cmd


class Change(NamedTuple):
    """Stores details about commits"""

    full_hash: str
    short_hash: str
    date: str
    message: str
    message_without_backticks: str
    pr: Optional[int]


def get_change_from_line(line: str):
    split_line = line.split(" ", maxsplit=3)
    message = split_line[3]
    pr = None
    pr_match = PR_PATTERN.match(message)
    if pr_match:
        pr = pr_match.group(1)
    return Change(
        full_hash=split_line[0],
        short_hash=split_line[1],
        date=split_line[2],
        message=message,
        message_without_backticks=message.replace("`", "'").replace("&#39;", "'").replace('&amp;', "&"),
        pr=int(pr) if pr else None,
    )


def get_changes(verbose: bool, previous_release: str, current_release: str) -> List[Change]:
    change_strings = subprocess.check_output(
        get_git_log_command(verbose, from_commit=previous_release, to_commit=current_release),
        cwd=SOURCE_DIR_PATH,
        universal_newlines=True,
    )
    return [get_change_from_line(line) for line in change_strings.split("\n")]


def render_template(
    template_name: str,
    context: Dict[str, Any],
    autoescape: bool = True,
    keep_trailing_newline: bool = False,
) -> str:
    """
    Renders template based on it's name. Reads the template from <name>_TEMPLATE.md.jinja2 in current dir.
    :param template_name: name of the template to use
    :param context: Jinja2 context
    :param autoescape: Whether to autoescape HTML
    :param keep_trailing_newline: Whether to keep the newline in rendered output
    :return: rendered template
    """
    import jinja2

    template_loader = jinja2.FileSystemLoader(searchpath=MY_DIR_PATH)
    template_env = jinja2.Environment(
        loader=template_loader,
        undefined=jinja2.StrictUndefined,
        autoescape=autoescape,
        keep_trailing_newline=keep_trailing_newline,
    )
    template = template_env.get_template(f"{template_name}_TEMPLATE.md.jinja2")
    content: str = template.render(context)
    return content


def print_issue_content(
    current_release: str,
    pull_requests: Dict[int, PullRequestOrIssue],
    linked_issues: Dict[int, List[Issue.Issue]],
    users: Dict[int, Set[str]],
):
    pr_list = list(pull_requests.keys())
    pr_list.sort()
    user_logins: Dict[int, str] = {pr: "@" + " @".join(users[pr]) for pr in users}
    all_users: Set[str] = set()
    for user_list in users.values():
        all_users.update(user_list)
    all_user_logins = "@" + " @".join(all_users)
    content = render_template(
        template_name='ISSUE',
        context={
            'version': current_release,
            'pr_list': pr_list,
            'pull_requests': pull_requests,
            'linked_issues': linked_issues,
            'users': users,
            'user_logins': user_logins,
            'all_user_logins': all_user_logins,
        },
        autoescape=False,
        keep_trailing_newline=True,
    )
    print(content)


@cli.command()
@option_github_token
@option_previous_release
@option_current_release
@option_excluded_pr_list
@option_verbose
@option_limit_pr_count
def generate_issue_content(
    github_token: str,
    previous_release: str,
    current_release: str,
    excluded_pr_list: str,
    verbose: bool,
    limit_pr_count: Optional[int],
):
    if excluded_pr_list:
        excluded_prs = [int(pr) for pr in excluded_pr_list.split(",")]
    else:
        excluded_prs = []
    changes = get_changes(verbose, previous_release, current_release)
    change_prs = [change.pr for change in changes]
    prs = [pr for pr in change_prs if pr is not None and pr not in excluded_prs]

    g = Github(github_token)
    repo = g.get_repo("apache/airflow")
    pull_requests: Dict[int, PullRequestOrIssue] = {}
    linked_issues: Dict[int, List[Issue.Issue]] = defaultdict(lambda: [])
    users: Dict[int, Set[str]] = defaultdict(lambda: set())
    count_prs = len(prs)
    if limit_pr_count:
        count_prs = limit_pr_count
    with Progress(console=console) as progress:
        task = progress.add_task(f"Retrieving {count_prs} PRs ", total=count_prs)
        for i in range(count_prs):
            pr_number = prs[i]
            progress.console.print(
                f"Retrieving PR#{pr_number}: " f"https://github.com/apache/airflow/pull/{pr_number}"
            )

            pr: PullRequestOrIssue
            try:
                pr = repo.get_pull(pr_number)
            except UnknownObjectException:
                # Fallback to issue if PR not found
                try:
                    pr = repo.get_issue(pr_number)  # (same fields as PR)
                except UnknownObjectException:
                    console.print(f"[red]The PR #{pr_number} could not be found[/]")
                    continue

            # Ignore doc-only and skipped PRs
            label_names = [label.name for label in pr.labels]
            if "type:doc-only" in label_names or "changelog:skip" in label_names:
                continue

            pull_requests[pr_number] = pr
            # GitHub does not have linked issues in PR - but we quite rigorously add Fixes/Closes
            # Relate so we can find those from the body
            if pr.body:
                body = pr.body.replace("\n", " ").replace("\r", " ")
                linked_issue_numbers = {
                    int(issue_match.group(1)) for issue_match in ISSUE_MATCH_IN_BODY.finditer(body)
                }
                for linked_issue_number in linked_issue_numbers:
                    progress.console.print(
                        f"Retrieving Linked issue PR#{linked_issue_number}: "
                        f"https://github.com/apache/airflow/issue/{linked_issue_number}"
                    )
                    try:
                        linked_issues[pr_number].append(repo.get_issue(linked_issue_number))
                    except UnknownObjectException:
                        progress.console.print(
                            f"Failed to retrieve linked issue #{linked_issue_number}: Unknown Issue"
                        )
            users[pr_number].add(pr.user.login)
            for linked_issue in linked_issues[pr_number]:
                users[pr_number].add(linked_issue.user.login)
            progress.advance(task)
    print_issue_content(current_release, pull_requests, linked_issues, users)


if __name__ == "__main__":
    cli()
