<commit_before>#SPDX-License-Identifier: MIT
from multiprocessing import Process, Queue
from urllib.parse import urlparse
import pandas as pd
import sqlalchemy as s
import requests
import time
import logging
import json
import os
import math
from datetime import datetime
from workers.worker_base import Worker

class GitHubWorker(Worker):
    """ Worker that collects data from the Github API and stores it in our database
    task: most recent task the broker added to the worker's queue
    child: current process of the queue being ran
    queue: queue of tasks to be fulfilled
    config: holds info like api keys, descriptions, and database connection strings
    """
    def __init__(self, config={}):

        worker_type = 'github_worker'

        given = [['github_url']]
        models = ['issues']

        data_tables = [
            'contributors', 'issues', 'issue_labels', 'message',
            'issue_message_ref', 'issue_events','issue_assignees','contributors_aliases',
            'pull_request_assignees', 'pull_request_events', 'pull_request_reviewers',
            'pull_request_meta', 'pull_request_repo'
        ]
        operations_tables = ['worker_history', 'worker_job']

        # These 3 are included in every tuple the worker inserts (data collection info)
        self.tool_source = 'GitHub API Worker'
        self.tool_version = '1.0.0'
        self.data_source = 'GitHub API'

        # if we are finishing a previous task, pagination works differenty (deprecated)
        self.finishing_task = True

        self.platform_id = 25150  # GitHub
        self.process_count = 1
        self.deep_collection = True

        # Run the general worker initialization
        super().__init__(worker_type, config, given, models, data_tables, operations_tables)

    def _get_pk_source_issues(self):

        issues_url = (
            f"https://api.github.com/repos/{self.owner}/{self.repo}"
            "/issues?per_page=100&state=all&page={}"
        )

        action_map = {
            'insert': {
                'source': ['id'],
                'augur': ['gh_issue_id']
            },
            'update': {
                'source': ['comments', 'state'],
                'augur': ['comment_count', 'issue_state']
            }
        }

        source_issues = self.new_paginate_endpoint(
            issues_url, action_map=action_map,
            table=self.issues_table, where_clause=self.issues_table.c.repo_id == self.repo_id
        )

        if len(source_issues['all']) == 0:
            self.logger.info("There are no issues for this repository.\n")
            self.register_task_completion(self.task_info, self.repo_id, 'issues')
            return False

        def is_valid_pr_block(issue):
            return (
                'pull_request' in issue and issue['pull_request']
                and isinstance(issue['pull_request'], dict) and 'url' in issue['pull_request']
            )

        source_issues['insert'] = self.enrich_cntrb_id(
            source_issues['insert'], 'user.login', action_map_additions={
                'insert': {
                    'source': ['user.node_id'],
                    'augur': ['gh_node_id']
                }
            }, prefix='user.'
        )

        issues_insert = [
            {
                'repo_id': self.repo_id,
                'reporter_id': issue['cntrb_id'],
                'pull_request': (
                    issue['pull_request']['url'].split('/')[-1]
                    if is_valid_pr_block(issue) else None
                ),
                'pull_request_id': (
                    issue['pull_request']['url'].split('/')[-1]
                    if is_valid_pr_block(issue) else None
                ),
                'created_at': issue['created_at'],
                'issue_title': issue['title'],
                'issue_body': issue['body'].replace('0x00', '____') if issue['body'] else None,
                'comment_count': issue['comments'],
                'updated_at': issue['updated_at'],
                'closed_at': issue['closed_at'],
                'repository_url': issue['repository_url'],
                'issue_url': issue['url'],
                'labels_url': issue['labels_url'],
                'comments_url': issue['comments_url'],
                'events_url': issue['events_url'],
                'html_url': issue['html_url'],
                'issue_state': issue['state'],
                'issue_node_id': issue['node_id'],
                'gh_issue_id': issue['id'],
                'gh_issue_number': issue['number'],
                'gh_user_id': issue['user']['id'],
                'tool_source': self.tool_source,
                'tool_version': self.tool_version,
                'data_source': self.data_source
            } for issue in source_issues['insert']
        ]

        if len(source_issues['insert']) > 0 or len(source_issues['update']) > 0:

            issues_insert_result, issues_update_result = self.bulk_insert(
                self.issues_table, update=source_issues['update'],
                unique_columns=action_map['insert']['augur'],
                insert=issues_insert, update_columns=action_map['update']['augur']
            )

            source_data = source_issues['insert'] + source_issues['update']

        elif not self.deep_collection:
            self.logger.info(
                "There are not issues to update, insert, or collect nested information for.\n"
            )
            self.register_task_completion(entry_info, self.repo_id, 'issues')
            return

        if self.deep_collection:
            source_data = source_issues['all']

        # Merge source data to inserted data to have access to inserted primary keys

        gh_merge_fields = ['id']
        augur_merge_fields = ['gh_issue_id']

        pk_source_issues = self.enrich_data_primary_keys(
            source_data, self.issues_table, gh_merge_fields, augur_merge_fields
        )

        return pk_source_issues

    def issues_model(self, entry_info, repo_id):
        """ Data collection function
        Query the GitHub API for issues
        """

        github_url = entry_info['given']['github_url']

        # Contributors are part of this model, and finding all for the repo saves us
        #   from having to add them as we discover committers in the issue process
        # self.query_github_contributors(entry_info, self.repo_id)

        pk_source_issues = self._get_pk_source_issues()
        if pk_source_issues:
            # self.issue_comments_model(pk_source_issues)
            issue_events_all = self.issue_events_model(pk_source_issues)
            self.issue_nested_data_model(pk_source_issues, issue_events_all)

        # Register this task as completed
        self.register_task_completion(entry_info, self.repo_id, 'issues')

    def issue_comments_model(self, pk_source_issues):

        comments_url = (
            f"https://api.github.com/repos/{self.owner}/{self.repo}"
            "/issues/comments?per_page=100&page={}"
        )

        # Get contributors that we already have stored
        #   Set our duplicate and update column map keys (something other than PK) to
        #   check dupicates/needed column updates with
        comment_action_map = {
            'insert': {
                'source': ['created_at', 'body'],<commit_msg>uncomment comment model<commit_after>#SPDX-License-Identifier: MIT
from multiprocessing import Process, Queue
from urllib.parse import urlparse
import pandas as pd
import sqlalchemy as s
import requests
import time
import logging
import json
import os
import math
from datetime import datetime
from workers.worker_base import Worker

class GitHubWorker(Worker):
    """ Worker that collects data from the Github API and stores it in our database
    task: most recent task the broker added to the worker's queue
    child: current process of the queue being ran
    queue: queue of tasks to be fulfilled
    config: holds info like api keys, descriptions, and database connection strings
    """
    def __init__(self, config={}):

        worker_type = 'github_worker'

        given = [['github_url']]
        models = ['issues']

        data_tables = [
            'contributors', 'issues', 'issue_labels', 'message',
            'issue_message_ref', 'issue_events','issue_assignees','contributors_aliases',
            'pull_request_assignees', 'pull_request_events', 'pull_request_reviewers',
            'pull_request_meta', 'pull_request_repo'
        ]
        operations_tables = ['worker_history', 'worker_job']

        # These 3 are included in every tuple the worker inserts (data collection info)
        self.tool_source = 'GitHub API Worker'
        self.tool_version = '1.0.0'
        self.data_source = 'GitHub API'

        # if we are finishing a previous task, pagination works differenty (deprecated)
        self.finishing_task = True

        self.platform_id = 25150  # GitHub
        self.process_count = 1
        self.deep_collection = True

        # Run the general worker initialization
        super().__init__(worker_type, config, given, models, data_tables, operations_tables)

    def _get_pk_source_issues(self):

        issues_url = (
            f"https://api.github.com/repos/{self.owner}/{self.repo}"
            "/issues?per_page=100&state=all&page={}"
        )

        action_map = {
            'insert': {
                'source': ['id'],
                'augur': ['gh_issue_id']
            },
            'update': {
                'source': ['comments', 'state'],
                'augur': ['comment_count', 'issue_state']
            }
        }

        source_issues = self.new_paginate_endpoint(
            issues_url, action_map=action_map,
            table=self.issues_table, where_clause=self.issues_table.c.repo_id == self.repo_id
        )

        if len(source_issues['all']) == 0:
            self.logger.info("There are no issues for this repository.\n")
            self.register_task_completion(self.task_info, self.repo_id, 'issues')
            return False

        def is_valid_pr_block(issue):
            return (
                'pull_request' in issue and issue['pull_request']
                and isinstance(issue['pull_request'], dict) and 'url' in issue['pull_request']
            )

        source_issues['insert'] = self.enrich_cntrb_id(
            source_issues['insert'], 'user.login', action_map_additions={
                'insert': {
                    'source': ['user.node_id'],
                    'augur': ['gh_node_id']
                }
            }, prefix='user.'
        )

        issues_insert = [
            {
                'repo_id': self.repo_id,
                'reporter_id': issue['cntrb_id'],
                'pull_request': (
                    issue['pull_request']['url'].split('/')[-1]
                    if is_valid_pr_block(issue) else None
                ),
                'pull_request_id': (
                    issue['pull_request']['url'].split('/')[-1]
                    if is_valid_pr_block(issue) else None
                ),
                'created_at': issue['created_at'],
                'issue_title': issue['title'],
                'issue_body': issue['body'].replace('0x00', '____') if issue['body'] else None,
                'comment_count': issue['comments'],
                'updated_at': issue['updated_at'],
                'closed_at': issue['closed_at'],
                'repository_url': issue['repository_url'],
                'issue_url': issue['url'],
                'labels_url': issue['labels_url'],
                'comments_url': issue['comments_url'],
                'events_url': issue['events_url'],
                'html_url': issue['html_url'],
                'issue_state': issue['state'],
                'issue_node_id': issue['node_id'],
                'gh_issue_id': issue['id'],
                'gh_issue_number': issue['number'],
                'gh_user_id': issue['user']['id'],
                'tool_source': self.tool_source,
                'tool_version': self.tool_version,
                'data_source': self.data_source
            } for issue in source_issues['insert']
        ]

        if len(source_issues['insert']) > 0 or len(source_issues['update']) > 0:

            issues_insert_result, issues_update_result = self.bulk_insert(
                self.issues_table, update=source_issues['update'],
                unique_columns=action_map['insert']['augur'],
                insert=issues_insert, update_columns=action_map['update']['augur']
            )

            source_data = source_issues['insert'] + source_issues['update']

        elif not self.deep_collection:
            self.logger.info(
                "There are not issues to update, insert, or collect nested information for.\n"
            )
            self.register_task_completion(entry_info, self.repo_id, 'issues')
            return

        if self.deep_collection:
            source_data = source_issues['all']

        # Merge source data to inserted data to have access to inserted primary keys

        gh_merge_fields = ['id']
        augur_merge_fields = ['gh_issue_id']

        pk_source_issues = self.enrich_data_primary_keys(
            source_data, self.issues_table, gh_merge_fields, augur_merge_fields
        )

        return pk_source_issues

    def issues_model(self, entry_info, repo_id):
        """ Data collection function
        Query the GitHub API for issues
        """

        github_url = entry_info['given']['github_url']

        # Contributors are part of this model, and finding all for the repo saves us
        #   from having to add them as we discover committers in the issue process
        # self.query_github_contributors(entry_info, self.repo_id)

        pk_source_issues = self._get_pk_source_issues()
        if pk_source_issues:
            self.issue_comments_model(pk_source_issues)
            issue_events_all = self.issue_events_model(pk_source_issues)
            self.issue_nested_data_model(pk_source_issues, issue_events_all)

        # Register this task as completed
        self.register_task_completion(entry_info, self.repo_id, 'issues')

    def issue_comments_model(self, pk_source_issues):

        comments_url = (
            f"https://api.github.com/repos/{self.owner}/{self.repo}"
            "/issues/comments?per_page=100&page={}"
        )

        # Get contributors that we already have stored
        #   Set our duplicate and update column map keys (something other than PK) to
        #   check dupicates/needed column updates with
        comment_action_map = {
            'insert': {
                'source': ['created_at', 'body'],
{'fraction_non_alphanumeric': 0.10673624288425047, 'fraction_numerical': 0.003252914068853348, 'mean_word_length': 2.1721840068787617, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 4, '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': '511317', 'n_tokens_mistral': 4169, 'n_tokens_neox': 3932, 'n_words': 927}