repo-cloner/repo_cloner/lib/detector.py

from repo_cloner.lib import DiskStoredList, DiskStoredRefs, RepoTool
from pathlib import Path
import logging
import json
from typing import Callable

log = logging.getLogger("rc.detector")


class DetectedCommit:
	_commit: str = None
	_abbrev: str = None
	_author: str = None
	_date: int = 0
	_is_tag: bool = False
	_tags: list = []
	_is_branch: bool = False
	_branches: list = []
	_log: str = ""
	_dict = {}

	def __init__(self, env: dict):
		for key in env.keys():
			self.__setattr__(f"_{key}", env[key])
		self._dict = env

	@property
	def commit(self) -> str:
		return self._commit

	@property
	def abbrev(self) -> str:
		return self._abbrev

	@property
	def author(self) -> str:
		return self._author

	@property
	def date(self) -> int:
		return self._date

	@property
	def is_tag(self) -> bool:
		return self._is_tag

	@property
	def tags(self) -> str:
		return ", ".join(self._tags)

	@property
	def is_branch(self) -> bool:
		return self._is_branch

	@property
	def branches(self) -> str:
		return ", ".join(self._branches)

	@property
	def log(self) -> str:
		return self._log

	@property
	def dict(self) -> dict:
		return self._dict


class Detector:
	_repo: RepoTool = None
	_repo_path: Path = None
	_detector_dir: Path = None
	_executed: DiskStoredList = None
	_branches: DiskStoredRefs = None
	_tags: DiskStoredRefs = None

	def __init__(self, repo_path: Path, cache_dir: Path, project: str):
		log.debug(f"Initializing detector...")
		log.debug(f"Repo: {repo_path}")
		self._repo_path = Path(repo_path)
		self._repo = RepoTool(repo_path)
		self._detector_dir = Path(cache_dir).joinpath("detector")
		if not self._detector_dir.exists():
			log.debug(f"Creating detector dir")
			self._detector_dir.mkdir()

		log.debug(f"Detector cache: {self._detector_dir}")
		self._executed = DiskStoredList(self._detector_dir.joinpath("detectorExecuted").as_posix())
		log.debug(f"Detector executed: {len(self._executed)} commits")

		# modify branches and tags to new standards
		self.check_legacy_config()
		# parse json files afterwards
		self._branches = DiskStoredRefs(self._detector_dir.joinpath("branches"))
		self._tags = DiskStoredRefs(self._detector_dir.joinpath("tags"))
		log.info(f"Loaded {self._branches.count()} branches and {self._tags.count()} tags")

	@classmethod
	def _ref_dir_to_json(cls, dir: Path):
		def rmdir(directory):
			directory = Path(directory)
			for item in directory.iterdir():
				if item.is_dir():
					rmdir(item)
				else:
					item.unlink()
			directory.rmdir()

		ref_dict = {}
		for item in dir.iterdir():
			if item.is_file():
				content = item.read_text().strip()
				name = item.name
				log.debug(f"Found reference {name} -> {content}")
				ref_dict[name] = content
		rmdir(dir)
		dir.touch()
		dir.write_text(json.dumps(sorted(ref_dict.items())))

	def check_legacy_config(self):
		branch_dir = self._detector_dir.joinpath("branches")
		tag_dir = self._detector_dir.joinpath("tags")

		if branch_dir.exists() and branch_dir.is_dir():
			log.info(f"Found legacy branch dir: {branch_dir} - converting now")
			Detector._ref_dir_to_json(branch_dir)

		if tag_dir.exists() and tag_dir.is_dir():
			log.info(f"Found legacy tag dir: {tag_dir} - converting now")
			Detector._ref_dir_to_json(tag_dir)

	def initialize_caches(self):
		# initialize caches
		log.info(f"Initializing detector cache")
		for commit in self._repo.list_commits():
			if commit not in self._executed:
				self._executed.append(commit.hexsha)
		# cleanup old branches
		for branch in self._branches.keys():
			self._branches.remove(branch)
		for tag in self._tags.keys():
			self._tags.remove(tag)

		# persist new ones
		for branch, commit in self._repo.list_branches().items():
			self._branches.update(branch, commit)
		for tag, commit in self._repo.list_tags().items():
			self._tags.update(tag, commit)

	def run(self, callback: Callable[[DetectedCommit], None]) -> int:
		log.info(f"Running commit detector")
		new_branches = self._repo.list_branches()
		new_tags = self._repo.list_tags()

		# remove removed
		old_keys = self._branches.keys()
		for branch in old_keys:
			if branch not in new_branches.keys():
				log.info(f"Branch {branch} removed in source, removing from detector")
				self._branches.remove(branch)
		old_keys = self._tags.keys()
		for tag in old_keys:
			if tag not in new_tags.keys():
				log.info(f"Tag {tag} removed in source, removing from detector")
				self._tags.remove(tag)

		keep_in_mind_commits: list = []
		for commit in list(new_tags.values()) + list(new_branches.values()):
			if commit not in keep_in_mind_commits:
				keep_in_mind_commits.append(commit)

		# list commits
		executed_count: int = 0
		for commit in self._repo.list_commits():
			if commit.hexsha in self._executed and commit.hexsha not in keep_in_mind_commits:
				continue
			special_commit: bool = False
			if commit.hexsha in keep_in_mind_commits:
				log.debug(f"Found keep-in-mind commit {commit.hexsha}")
				special_commit = True

			special_branch: list = []
			special_tag: list = []

			if special_commit:
				for branch, commit_candidate in new_branches.items():
					if commit.hexsha == commit_candidate:
						if not self._branches.get(branch) == commit.hexsha:
							log.debug(f"Found branch {branch} for commit {commit_candidate}")
							special_branch.append(branch)
				for tag, commit_candidate in new_tags.items():
					if commit.hexsha == commit_candidate:
						if not self._tags.get(tag) == commit.hexsha:
							log.debug(f"Found tag {tag} for commit {commit_candidate}")
							special_tag.append(tag)

			env = {
				'commit':    commit.hexsha,
				'abbrev':    commit.hexsha[0:7],
				'author':    str(commit.author),
				'date':      commit.authored_date,
				'is_tag':    len(special_tag) > 0,
				'tags':      special_tag,
				'is_branch': len(special_branch) > 0,
				'branches':  special_branch,
				'log':       commit.message,
			}
			env = DetectedCommit(env)
			executed_count += 1
			callback(env)
		return executed_count