10229078296

Committed 03 Aug 2024 03:27PM UTC coverage: 63.165% (+0.003%) from 63.162%

Build # 10229078296

Build Type

Pull #480

github

Committed by

web-flow

Commit Message

Merge 2467fc010 into b18ac6a3b

Pull Request Pull Request #480: chore: convert json files to yaml

Run Details

2006 of 3018 branches covered (66.47%)

Branch coverage included in aggregate %.

39 of 50 new or added lines in 2 files covered. (78.0%)

96 existing lines in 3 files now uncovered.

7626 of 12231 relevant lines covered (62.35%)

0.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

8.7

/fuji_server/harvester/github_harvester.py

# SPDX-FileCopyrightText: 2020 PANGAEA (https://www.pangaea.de/)
#
# SPDX-License-Identifier: MIT

import json
import os
import re
import time
from configparser import ConfigParser

from github import Auth, Github
from github.GithubException import UnknownObjectException


class GithubHarvester:
    def __init__(self, id, logger, host="https://github.com", verbose=True):
        # Read Github API access token from config file.
        config = ConfigParser()
        config.read(os.path.join(os.path.abspath(os.path.dirname(__file__)), "../config/github.ini"))
        self.logger = logger
        self.id = id
        self.host = host
        self.verbose = verbose
        self.authenticate(config)
        self.data = {}  # dictionary with all info
        fuji_server_dir = os.path.dirname(os.path.dirname(__file__))  # project_root
        software_file_path = os.path.join(fuji_server_dir, "data", "software_file.json")
        with open(software_file_path) as f:
            self.files_map = json.load(f)

    def authenticate(self, config):
        """Runs every time a new harvesting request comes in, as the harvester is re-initialised every time. Picks a token (if available) and initialises the pyGithub handle.

        Args:
            config (dict): parsed configuration dictionary
        """
        token_file = config["ACCESS"]["token_file"]
        token_to_use = None
        if token_file != "":
            with open(token_file) as f:
                token_list = f.read().splitlines()
            # find a token with enough remaining requests, or the one with most remaining if none available
            fallback_max_token = None
            fallback_max_rate_limit = 0
            for token in token_list:
                try:
                    rate_limit = Github(auth=Auth.Token(token)).get_rate_limit()
                    if rate_limit.core.remaining >= 1000 and rate_limit.search.remaining >= 2:
                        token_to_use = token
                        break
                    elif rate_limit.core.remaining > fallback_max_rate_limit:
                        fallback_max_rate_limit = rate_limit.core.remaining
                        fallback_max_token = token
                except:  # ignore expired or invalid tokens
                    pass
            if token_to_use is None:
                token_to_use = fallback_max_token
        else:
            token = config["ACCESS"]["token"]
            if token != "":
                token_to_use = token
        if token_to_use is not None:  # found a token, one way or another
            auth = Auth.Token(token)
            if self.verbose:
                rate_limit = Github(auth=Auth.Token(token)).get_rate_limit()
                print(
                    f"Authenticate using GitHub token ending on '{token[-4:]}'.\n    Remaining core requests: {rate_limit.core.remaining}\n    Remaining search requests: {rate_limit.search.remaining}"
                )
        else:  # empty token, so no authentication possible (rate limit will be much lower)
            auth = None
            self.logger.warning(
                "FRSM-09-A1 : Running in unauthenticated mode. Capabilities are limited."
            )  # TODO: would be better if it were a general warning!
        if self.host != "https://github.com":
            base_url = f"{self.host}/api/v3"
            self.handle = Github(auth=auth, base_url=base_url)
        else:
            self.handle = Github(auth=auth)

    def harvest(self):
        tic = time.perf_counter()
        # check if it's a URL or repo ID
        # NOTE: this should probably be handled by IdentifierHelper, but I don't understand that module yet.
        if self.id.count("/") > 1:  # URL
            _, self.username, self.repo_name = self.id.rsplit("/", 2)
        else:  # repo ID
            self.username, self.repo_name = self.id.split("/")
        self.repo_id = "/".join([self.username, self.repo_name])

        # access repo via GitHub API
        try:
            repo = self.handle.get_repo(self.repo_id)
        except UnknownObjectException:
            print("Could not find repo.")
            self.logger.warning(
                "FRSM-09-A1 : Could not find repository on GitHub."
            )  # TODO: would be better if it were a general warning!
            return

        # harvesting
        try:  # LICENSE in metadata
            license_metadata = repo.get_license()
            self.data["license_path"] = license_metadata.path
            self.data["license"] = license_metadata.license.name
        except UnknownObjectException:
            pass

        # identify source code (sample files in the main language used in the repo)
        repo_languages = repo.get_languages()
        if repo_languages != {}:
            self.data["languages"] = repo_languages
        main_source_code_language = repo.language
        if main_source_code_language is not None:
            self.data["main_language"] = main_source_code_language
            query = f' repo:{self.repo_id} language:"{main_source_code_language}"'  # needs the space in front as every query needs a string to match on
            source_code_files = self.handle.search_code(query)
            # extract code of up to n=5 files
            n = min(5, source_code_files.totalCount)
            source_code_samples = []
            for i in range(n):
                source_code_samples.append(
                    {
                        "path": source_code_files[i].path,
                        "language": main_source_code_language,
                        "content": source_code_files[i].decoded_content,
                    }
                )
            if len(source_code_samples) > 0:
                self.data["source_code_samples"] = source_code_samples

        self.retrieve_all(repo)
        toc = time.perf_counter()
        if self.verbose:
            print(f"Harvesting took {toc - tic:.4f} seconds.")

    def retrieve_all(self, repo):
        file_pattern = r"|".join([rf"(?P<{k}>{'|'.join(v['pattern'])})" for k, v in self.files_map.items()])
        repo_contents = repo.get_contents("")
        while repo_contents:
            content_file = repo_contents.pop(0)
            if content_file.type == "dir":
                repo_contents.extend(repo.get_contents(content_file.path))
            else:
                m = re.fullmatch(file_pattern, content_file.path)
                if m is not None and any(m.groupdict().values()):
                    for k, v in m.groupdict().items():
                        if v is not None:
                            if self.files_map[k]["parse"] == "full":
                                file_entry = {
                                    "name": content_file.name,
                                    "path": content_file.path,
                                    "content": content_file.decoded_content,
                                }
                            elif self.files_map[k]["parse"] == "file_name":
                                file_entry = {"name": content_file.name, "path": content_file.path}
                            else:
                                self.logger.warning(
                                    f"FRSM-09-A1 : Parsing strategy {self.files_map[k]['parse']} is currently not implemented. Choose one of 'full' or 'file_name' for files {k}. Defaulting to parsing strategy 'file_name'."
                                )
                                file_entry = {"name": content_file.name, "path": content_file.path}
                            try:
                                self.data[k].append(file_entry)
                            except KeyError:
                                self.data[k] = [file_entry]

1	# SPDX-FileCopyrightText: 2020 PANGAEA (https://www.pangaea.de/)
2	#
3	# SPDX-License-Identifier: MIT
4
5	import json	1✔
6	import os	1✔
7	import re	1✔
8	import time	1✔
9	from configparser import ConfigParser	1✔
10
11	from github import Auth, Github	1✔
12	from github.GithubException import UnknownObjectException	1✔
13
14
15	class GithubHarvester:	1✔
16	def __init__(self, id, logger, host="https://github.com", verbose=True):	1✔
17	# Read Github API access token from config file.
UNCOV 18	config = ConfigParser()	×
UNCOV 19	config.read(os.path.join(os.path.abspath(os.path.dirname(__file__)), "../config/github.ini"))	×
20	self.logger = logger	×
21	self.id = id	×
22	self.host = host	×
23	self.verbose = verbose	×
24	self.authenticate(config)	×
25	self.data = {} # dictionary with all info	×
26	fuji_server_dir = os.path.dirname(os.path.dirname(__file__)) # project_root	×
27	software_file_path = os.path.join(fuji_server_dir, "data", "software_file.json")	×
28	with open(software_file_path) as f:	×
29	self.files_map = json.load(f)	×
30
31	def authenticate(self, config):	1✔
32	"""Runs every time a new harvesting request comes in, as the harvester is re-initialised every time. Picks a token (if available) and initialises the pyGithub handle.
33
34	Args:
35	config (dict): parsed configuration dictionary
36	"""
UNCOV 37	token_file = config["ACCESS"]["token_file"]	×
UNCOV 38	token_to_use = None	×
39	if token_file != "":	×
40	with open(token_file) as f:	×
41	token_list = f.read().splitlines()	×
42	# find a token with enough remaining requests, or the one with most remaining if none available
43	fallback_max_token = None	×
UNCOV 44	fallback_max_rate_limit = 0	×
45	for token in token_list:	×
46	try:	×
47	rate_limit = Github(auth=Auth.Token(token)).get_rate_limit()	×
48	if rate_limit.core.remaining >= 1000 and rate_limit.search.remaining >= 2:	×
49	token_to_use = token	×
50	break	×
51	elif rate_limit.core.remaining > fallback_max_rate_limit:	×
52	fallback_max_rate_limit = rate_limit.core.remaining	×
53	fallback_max_token = token	×
54	except: # ignore expired or invalid tokens	×
55	pass	×
56	if token_to_use is None:	×
57	token_to_use = fallback_max_token	×
58	else:
59	token = config["ACCESS"]["token"]	×
UNCOV 60	if token != "":	×
61	token_to_use = token	×
62	if token_to_use is not None: # found a token, one way or another	×
63	auth = Auth.Token(token)	×
64	if self.verbose:	×
65	rate_limit = Github(auth=Auth.Token(token)).get_rate_limit()	×
66	print(	×
67	f"Authenticate using GitHub token ending on '{token[-4:]}'.\n Remaining core requests: {rate_limit.core.remaining}\n Remaining search requests: {rate_limit.search.remaining}"
68	)
69	else: # empty token, so no authentication possible (rate limit will be much lower)
UNCOV 70	auth = None	×
UNCOV 71	self.logger.warning(	×
72	"FRSM-09-A1 : Running in unauthenticated mode. Capabilities are limited."
73	) # TODO: would be better if it were a general warning!
UNCOV 74	if self.host != "https://github.com":	×
UNCOV 75	base_url = f"{self.host}/api/v3"	×
76	self.handle = Github(auth=auth, base_url=base_url)	×
77	else:
78	self.handle = Github(auth=auth)	×
79
80	def harvest(self):	1✔
UNCOV 81	tic = time.perf_counter()	×
82	# check if it's a URL or repo ID
83	# NOTE: this should probably be handled by IdentifierHelper, but I don't understand that module yet.
UNCOV 84	if self.id.count("/") > 1: # URL	×
UNCOV 85	_, self.username, self.repo_name = self.id.rsplit("/", 2)	×
86	else: # repo ID
87	self.username, self.repo_name = self.id.split("/")	×
UNCOV 88	self.repo_id = "/".join([self.username, self.repo_name])	×
89
90	# access repo via GitHub API
UNCOV 91	try:	×
UNCOV 92	repo = self.handle.get_repo(self.repo_id)	×
93	except UnknownObjectException:	×
94	print("Could not find repo.")	×
95	self.logger.warning(	×
96	"FRSM-09-A1 : Could not find repository on GitHub."
97	) # TODO: would be better if it were a general warning!
UNCOV 98	return	×
99
100	# harvesting
UNCOV 101	try: # LICENSE in metadata	×
UNCOV 102	license_metadata = repo.get_license()	×
103	self.data["license_path"] = license_metadata.path	×
104	self.data["license"] = license_metadata.license.name	×
105	except UnknownObjectException:	×
106	pass	×
107
108	# identify source code (sample files in the main language used in the repo)
UNCOV 109	repo_languages = repo.get_languages()	×
UNCOV 110	if repo_languages != {}:	×
111	self.data["languages"] = repo_languages	×
112	main_source_code_language = repo.language	×
113	if main_source_code_language is not None:	×
114	self.data["main_language"] = main_source_code_language	×
115	query = f' repo:{self.repo_id} language:"{main_source_code_language}"' # needs the space in front as every query needs a string to match on	×
116	source_code_files = self.handle.search_code(query)	×
117	# extract code of up to n=5 files
118	n = min(5, source_code_files.totalCount)	×
UNCOV 119	source_code_samples = []	×
120	for i in range(n):	×
121	source_code_samples.append(	×
122	{
123	"path": source_code_files[i].path,
124	"language": main_source_code_language,
125	"content": source_code_files[i].decoded_content,
126	}
127	)
UNCOV 128	if len(source_code_samples) > 0:	×
UNCOV 129	self.data["source_code_samples"] = source_code_samples	×
130
131	self.retrieve_all(repo)	×
UNCOV 132	toc = time.perf_counter()	×
133	if self.verbose:	×
134	print(f"Harvesting took {toc - tic:.4f} seconds.")	×
135
136	def retrieve_all(self, repo):	1✔
UNCOV 137	file_pattern = r"\|".join([rf"(?P<{k}>{'\|'.join(v['pattern'])})" for k, v in self.files_map.items()])	×
UNCOV 138	repo_contents = repo.get_contents("")	×
139	while repo_contents:	×
140	content_file = repo_contents.pop(0)	×
141	if content_file.type == "dir":	×
142	repo_contents.extend(repo.get_contents(content_file.path))	×
143	else:
144	m = re.fullmatch(file_pattern, content_file.path)	×
UNCOV 145	if m is not None and any(m.groupdict().values()):	×
146	for k, v in m.groupdict().items():	×
147	if v is not None:	×
148	if self.files_map[k]["parse"] == "full":	×
149	file_entry = {	×
150	"name": content_file.name,
151	"path": content_file.path,
152	"content": content_file.decoded_content,
153	}
UNCOV 154	elif self.files_map[k]["parse"] == "file_name":	×
UNCOV 155	file_entry = {"name": content_file.name, "path": content_file.path}	×
156	else:
157	self.logger.warning(	×
158	f"FRSM-09-A1 : Parsing strategy {self.files_map[k]['parse']} is currently not implemented. Choose one of 'full' or 'file_name' for files {k}. Defaulting to parsing strategy 'file_name'."
159	)
UNCOV 160	file_entry = {"name": content_file.name, "path": content_file.path}	×
UNCOV 161	try:	×
162	self.data[k].append(file_entry)	×
163	except KeyError:	×
164	self.data[k] = [file_entry]	×

pangaea-data-publisher / fuji / 10229078296

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous