<commit_before>"""Merge Request Nag Tool "Mr. Nag".

MIT License -- Copyright (c) 2019 hg-jt
"""
import logging
from datetime import datetime
from dataclasses import dataclass, field
from concurrent.futures import ThreadPoolExecutor
from os import environ
from typing import Callable, Iterable, List, Optional
import pendulum
from requests import Response, Session
from yaml import SafeLoader, load as yml_load


LOG = logging.getLogger(__name__)


@dataclass
class MergeRequestApprovals:
    """Represents the approvals for a merge/pull request"""
    total: int = 0
    required: int = 0


@dataclass
class MergeRequest:
    """Represents metadata about a merge/pull request."""
    title: str
    author: str
    created_at: datetime
    updated_at: datetime
    approvals: MergeRequestApprovals = field(default_factory=MergeRequestApprovals, init=False)
    labels: List[str] = field(default_factory=list)
    wip: bool = False
    comment_count: int = 0
    merge_request_iid: int = None
    url: str = None


@dataclass
class Project:
    """Represents a project in a source code management system (a.k.a a "forge")."""
    project_id: int
    forge: str
    name: str
    merge_requests: List[MergeRequest] = field(default_factory=list, init=False)
    url: str = None


@dataclass
class Forge:
    """Represents a "forge" (i.e. a web-based SCM repo manager).

    This class attempts to automatically load a GitLab API token from an
    environment variable that uses the forge type and forge id. For example,
    if the forge type is "gitlab" and the id is "abc, the API token environment
    variable would be ABC_GITLAB_TOKEN.
    """
    id: str
    type: str
    api_url: str
    token: str
    projects: List[Project] = field(default_factory=list)

    def __post_init__(self):
        """Post-initialization for ``token`` and ``projects``."""
        if not self.token:  # get token from environment variable
            self.token = environ.get(f'{self.id}_{self.type}_TOKEN'.upper())

        if self.projects:  # find "raw" dicts and convert them to Project objects
            projects: List[Project] = []

            for project in self.projects:
                if isinstance(project, dict):
                    projects.append(Project(**project, forge=self.id))
                elif isinstance(project, Project):
                    projects.append(project)
                else:
                    LOG.warning('Unable to parse project metadata, unexpected type: %s', type(project))

            self.projects = projects

    def fetch_project(self, project: Project) -> Project:
        """Fetch the project details.

        Sub-classes should override this method.

        :param project: The project to fetch the details for.
        :return: A hydrated project.
        """
        raise NotImplementedError()


@dataclass
class Gitlab(Forge):
    """Provides a minimal GitLab client for fetching project and merge request details."""
    session: Session = field(default_factory=Session, init=False, repr=False)

    def __post_init__(self):
        super().__post_init__()
        self.session.headers['PRIVATE-TOKEN'] = self.token

    def fetch_project(self, project: Project) -> Project:
        """Fetch the project details.

        :param project: The project to fetch the details for.
        :return: A hydrated project.
        """
        project: Project = self.get_merge_request_details(self.get_project_details(project))

        return project

    def get_merge_request_details(self, project: Project) -> Project:
        """Get merge request information from GitLab.

        :param project: The project to enrich.
        :return: The given project, enriched with merge request data.
        """
        resp: Response = self.session.get(f'{self.api_url}/projects/{project.project_id}/merge_requests?state=opened')

        if not resp.ok:
            # TODO: mark the project to indicate something went wrong
            # TODO: smoother handling of error state
            raise Exception('Unable to merge request details')

        for mr in resp.json():
            merge_request: MergeRequest = MergeRequest(
                mr['title'],
                mr.get('author', {}).get('name'),
                self.timestamp_to_datetime(mr['created_at']),
                self.timestamp_to_datetime(mr['updated_at']),
                mr['labels'],
                comment_count=mr['user_notes_count'],
                merge_request_iid=mr['iid'],
                url=mr['web_url']
            )

            resp = self.session.get(
                f'{self.api_url}/projects/{project.project_id}/merge_requests/{mr["iid"]}/approvals'
            )

            if not resp.ok:
                raise Exception('Unable to get approval details')  # TODO: smoother handling of error state

            approvals: dict = resp.json()
            merge_request.approvals.total = len(approvals['approved_by'])
            merge_request.approvals.required = approvals.get('approvals_required', 0)
            merge_request.wip = mr.get('work_in_progress', False)

            project.merge_requests.append(merge_request)

        return project

    def get_project_details(self, project: Project) -> Project:
        """Get project metadata from GitLab.

        :param project: The project to enrich.
        :return: The given project, enriched with additional metadata.
        """
        resp: Response = self.session.get(f'{self.api_url}/projects/{project.project_id}')

        if not resp.ok:
            raise Exception('Unable to fetch project details')  # TODO: smoother handling of error state

        gl_project = resp.json()

        if not project.name:
            project.name = gl_project.get('name', gl_project.get('id'))

        project.url = gl_project['web_url']

        return project

    @staticmethod
    def timestamp_to_datetime(timestamp) -> Optional[datetime]:
        """Convert a timestamp string to a datetime object with timezone info.

        GitLab's API uses timestamp strings in the format: %Y-%m-%dT%H:%M:%S.%fZ.

        :param timestamp: timestamp string.
        :return: A datetime object with timezone info or None.
        """
        if not timestamp:
            return None

        try:
            return pendulum.parse(timestamp)
        except ValueError:
            return None<commit_msg>Fixed typos and moved Gitlab#timestamp_to_datetime.<commit_after>"""Merge Request Nag Tool "Mr. Nag".

MIT License -- Copyright (c) 2019 hg-jt
"""
import logging
from datetime import datetime
from dataclasses import dataclass, field
from concurrent.futures import ThreadPoolExecutor
from os import environ
from typing import Callable, Iterable, List, Optional
import pendulum
from requests import Response, Session
from yaml import SafeLoader, load as yml_load


LOG = logging.getLogger(__name__)


@dataclass
class MergeRequestApprovals:
    """Represents the approvals for a merge/pull request"""
    total: int = 0
    required: int = 0


@dataclass
class MergeRequest:
    """Represents metadata about a merge/pull request."""
    title: str
    author: str
    created_at: datetime
    updated_at: datetime
    approvals: MergeRequestApprovals = field(default_factory=MergeRequestApprovals, init=False)
    labels: List[str] = field(default_factory=list)
    wip: bool = False
    comment_count: int = 0
    merge_request_id: int = None
    url: str = None


@dataclass
class Project:
    """Represents a project in a source code management system (i.e. a "forge")."""
    project_id: int
    forge: str
    name: str
    merge_requests: List[MergeRequest] = field(default_factory=list, init=False)
    url: str = None


@dataclass
class Forge:
    """Represents a "forge" (i.e. a web-based SCM repo manager).

    This class attempts to automatically load a GitLab API token from an
    environment variable that uses the forge type and forge id. For example,
    if the forge type is "gitlab" and the id is "abc, the API token environment
    variable would be ABC_GITLAB_TOKEN.

    *NOTE*: Dashes will be removed from the forge id.
    """
    id: str
    type: str
    api_url: str
    token: str
    projects: List[Project] = field(default_factory=list)

    def __post_init__(self):
        """Post-initialization for ``token`` and ``projects``."""
        if not self.token:  # get token from environment variable
            self.token = environ.get(f"{self.id.replace('-', '')}_{self.type}_TOKEN".upper())

        if self.projects:  # find "raw" dicts and convert them to Project objects
            projects: List[Project] = []

            for project in self.projects:
                if isinstance(project, dict):
                    projects.append(Project(**project, forge=self.id))
                elif isinstance(project, Project):
                    projects.append(project)
                else:
                    LOG.warning('Unable to parse project metadata, unexpected type: %s', type(project))

            self.projects = projects

    def fetch_project(self, project: Project) -> Project:
        """Fetch the project details.

        Sub-classes should override this method.

        :param project: The project to fetch the details for.
        :return: A hydrated project.
        """
        raise NotImplementedError()


@dataclass
class Gitlab(Forge):
    """Provides a minimal GitLab client for fetching project and merge request details."""
    session: Session = field(default_factory=Session, init=False, repr=False)

    def __post_init__(self):
        super().__post_init__()
        self.session.headers['PRIVATE-TOKEN'] = self.token

    def fetch_project(self, project: Project) -> Project:
        """Fetch the project details.

        :param project: The project to fetch the details for.
        :return: A hydrated project.
        """
        project: Project = self.get_merge_request_details(self.get_project_details(project))

        return project

    def get_merge_request_details(self, project: Project) -> Project:
        """Get merge request information from GitLab.

        :param project: The project to enrich.
        :return: The given project, enriched with merge request data.
        """
        resp: Response = self.session.get(f'{self.api_url}/projects/{project.project_id}/merge_requests?state=opened')

        if not resp.ok:
            # TODO: mark the project to indicate something went wrong
            # TODO: smoother handling of error state
            raise Exception('Unable to get list of merge request.')

        for mr in resp.json():
            merge_request: MergeRequest = MergeRequest(
                mr['title'],
                mr.get('author', {}).get('name'),
                timestamp_to_datetime(mr['created_at']),
                timestamp_to_datetime(mr['updated_at']),
                mr['labels'],
                comment_count=mr['user_notes_count'],
                merge_request_id=mr['iid'],
                url=mr['web_url']
            )

            resp = self.session.get(
                f'{self.api_url}/projects/{project.project_id}/merge_requests/{mr["iid"]}/approvals'
            )

            if not resp.ok:
                raise Exception('Unable to get approval details')  # TODO: smoother handling of error state

            approvals: dict = resp.json()
            merge_request.approvals.total = len(approvals['approved_by'])
            merge_request.approvals.required = approvals.get('approvals_required', 0)
            merge_request.wip = mr.get('work_in_progress', False)

            project.merge_requests.append(merge_request)

        return project

    def get_project_details(self, project: Project) -> Project:
        """Get project metadata from GitLab.

        :param project: The project to enrich.
        :return: The given project, enriched with additional metadata.
        """
        resp: Response = self.session.get(f'{self.api_url}/projects/{project.project_id}')

        if not resp.ok:
            raise Exception('Unable to fetch project details')  # TODO: smoother handling of error state

        gl_project: dict = resp.json()

        if not project.name:
            project.name = gl_project.get('name', gl_project.get('id'))

        project.url = gl_project['web_url']

        return project


def timestamp_to_datetime(timestamp) -> Optional[datetime]:
    """Convert a timestamp string to a datetime object with timezone info.

    :param timestamp: timestamp string.
    :return: A datetime object with timezone info or None.
    """
    if not timestamp:
        return None

    try:
        return pendulum.parse(timestamp)
    except ValueError:
        return None
{'fraction_non_alphanumeric': 0.09569791176008792, 'fraction_numerical': 0.0012560841576385617, 'mean_word_length': 2.594525959367946, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7613626', 'n_tokens_mistral': 3536, 'n_tokens_neox': 3353, 'n_words': 1086}