Skip to content

Commit

Permalink
Minor refactoring
Browse files Browse the repository at this point in the history
  • Loading branch information
jasursadikov committed Sep 12, 2024
1 parent 297ce58 commit bb97c78
Show file tree
Hide file tree
Showing 4 changed files with 57 additions and 50 deletions.
2 changes: 1 addition & 1 deletion app.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ def init(self, args) -> None:
continue
self.config.add_label(directory, getattr(args, 'label', ''))
index += 1
table.add_row([f'{DIM}{directory}{RESET}', f'{GREEN}{utils.GLYPHS["added"]}{RESET}'])
table.add_row([f'{DIM}{directory}{RESET}', f'{GREEN}{glyphs("added")}{RESET}'])
if index == 0:
utils.print_error('No git repositories were found in this directory.')
return
Expand Down
4 changes: 3 additions & 1 deletion main.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,15 @@
#!/usr/bin/env python3

import sys
import utils
import settings
from app import App

if __name__ == '__main__':
try:
utils.settings = settings.Settings(utils.SETTINGS_FILE_NAME)
utils.setup()
if utils.settings.config['mud'].getboolean('ask_updates') and utils.update():
sys.exit()
mud = App()
mud.run()
except KeyboardInterrupt:
Expand Down
83 changes: 46 additions & 37 deletions runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import asyncio
import subprocess

from utils import glyphs
from styles import *
from typing import List, Dict
from collections import Counter
Expand All @@ -26,7 +27,7 @@ def info(self, repos: Dict[str, List[str]]) -> None:
formatted_path = self._get_formatted_path(path)
branch = self._get_branch_status(path)
status = self._get_status_string(files)
colored_labels = self._get_formatted_labels(labels, utils.GLYPHS["label"])
colored_labels = self._get_formatted_labels(labels, glyphs("label"))

# Sync with origin status
ahead_behind_cmd = subprocess.run('git rev-list --left-right --count HEAD...@{upstream}', shell=True, text=True, cwd=path, capture_output=True)
Expand All @@ -35,14 +36,14 @@ def info(self, repos: Dict[str, List[str]]) -> None:
if len(stdout) >= 2:
ahead, behind = stdout[0], stdout[1]
if ahead and ahead != '0':
origin_sync += f'{BRIGHT_GREEN}{utils.GLYPHS["ahead"]} {ahead}{RESET}'
origin_sync += f'{BRIGHT_GREEN}{glyphs("ahead")} {ahead}{RESET}'
if behind and behind != '0':
if origin_sync:
origin_sync += ' '
origin_sync += f'{BRIGHT_BLUE}{utils.GLYPHS["behind"]} {behind}{RESET}'
origin_sync += f'{BRIGHT_BLUE}{glyphs("behind")} {behind}{RESET}'

if not origin_sync.strip():
origin_sync = f'{BLUE}{utils.GLYPHS["synced"]}{RESET}'
origin_sync = f'{BLUE}{glyphs("synced")}{RESET}'

table.add_row([formatted_path, branch, origin_sync, status, colored_labels])

Expand Down Expand Up @@ -91,7 +92,7 @@ def labels(self, repos: Dict[str, List[str]]):
table = utils.get_table()
for path, labels in repos.items():
formatted_path = self._get_formatted_path(path)
colored_labels = self._get_formatted_labels(labels, utils.GLYPHS['label'])
colored_labels = self._get_formatted_labels(labels, glyphs('label'))
table.add_row([formatted_path, colored_labels])

utils.print_table(table)
Expand Down Expand Up @@ -151,7 +152,7 @@ def tags(self, repos: Dict[str, List[str]]):
for path, labels in repos.items():
formatted_path = self._get_formatted_path(path)
tags = [line.strip() for line in subprocess.check_output('git tag', shell=True, text=True, cwd=path).splitlines() if line.strip()]
tags = [f'{utils.GLYPHS["tag"]}{utils.GLYPHS["space"]}{tag}' for tag in tags]
tags = [f'{glyphs("tag")}{glyphs("space")}{tag}' for tag in tags]
tags = ' '.join(tags)
table.add_row([formatted_path, tags])

Expand Down Expand Up @@ -198,7 +199,7 @@ async def task(repo: str) -> None:

async def _run_process(self, repo_path: str, table: Dict[str, List[str]], command: List[str]) -> None:
process = await asyncio.create_subprocess_exec(*command, cwd=repo_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
table[repo_path] = ['', f'{YELLOW}{utils.GLYPHS["running"]}{RESET}']
table[repo_path] = ['', f'{YELLOW}{glyphs("running")}{RESET}']

while True:
line = await process.stdout.readline()
Expand All @@ -208,15 +209,15 @@ async def _run_process(self, repo_path: str, table: Dict[str, List[str]], comman
break
line = line.decode().strip()
line = table[repo_path][0] if not line.strip() else line
table[repo_path] = [line, f'{YELLOW}{utils.GLYPHS["running"]}{RESET}']
table[repo_path] = [line, f'{YELLOW}{glyphs("running")}{RESET}']
self._print_process(table)

return_code = await process.wait()

if return_code == 0:
status = f'{GREEN}{utils.GLYPHS["finished"]}{RESET}'
status = f'{GREEN}{glyphs("finished")}{RESET}'
else:
status = f'{RED}{utils.GLYPHS["failed"]} Code: {return_code}{RESET}'
status = f'{RED}{glyphs("failed")} Code: {return_code}{RESET}'

table[repo_path] = [table[repo_path][0], status]
self._print_process(table)
Expand Down Expand Up @@ -253,52 +254,52 @@ def _get_status_string(files: List[str]):
moved += 1
status = ''
if added:
status += f'{BRIGHT_GREEN}{added} {utils.GLYPHS["added"]}{RESET} '
status += f'{BRIGHT_GREEN}{added} {glyphs("added")}{RESET} '
if modified:
status += f'{YELLOW}{modified} {utils.GLYPHS["modified"]}{RESET} '
status += f'{YELLOW}{modified} {glyphs("modified")}{RESET} '
if moved:
status += f'{BLUE}{moved} {utils.GLYPHS["moved"]}{RESET} '
status += f'{BLUE}{moved} {glyphs("moved")}{RESET} '
if removed:
status += f'{RED}{removed} {utils.GLYPHS["removed"]}{RESET} '
status += f'{RED}{removed} {glyphs("removed")}{RESET} '
if not files:
status = f'{GREEN}{utils.GLYPHS["clear"]}{RESET}'
status = f'{GREEN}{glyphs("clear")}{RESET}'
return status

@staticmethod
def _get_branch_status(path: str) -> str:
branch_cmd = subprocess.run('git rev-parse --abbrev-ref HEAD', shell=True, text=True, cwd=path, capture_output=True)
branch_stdout = branch_cmd.stdout.strip()
if branch_stdout == 'master' or branch_stdout == 'main':
return f'{YELLOW}{utils.GLYPHS["master"]}{RESET}{utils.GLYPHS["space"]}{branch_stdout}'
return f'{YELLOW}{glyphs("master")}{RESET}{glyphs("space")}{branch_stdout}'
elif branch_stdout == 'develop':
return f'{GREEN}{utils.GLYPHS["feature"]}{RESET}{utils.GLYPHS["space"]}{branch_stdout}'
return f'{GREEN}{glyphs("feature")}{RESET}{glyphs("space")}{branch_stdout}'
elif '/' in branch_stdout:
branch_path = branch_stdout.split('/')
icon = Runner._get_branch_icon(branch_path[0])
branch_color = Runner._get_branch_color(branch_path[0])
return f'{branch_color}{icon}{RESET}{utils.GLYPHS["space"]}{branch_path[0]}{RESET}/{BOLD}{("/".join(branch_path[1:]))}'
return f'{branch_color}{icon}{RESET}{glyphs("space")}{branch_path[0]}{RESET}/{BOLD}{("/".join(branch_path[1:]))}'
elif branch_stdout == 'HEAD':
# check if we are on tag
glyph = utils.GLYPHS['tag']
glyph = glyphs('tag')
color = BRIGHT_MAGENTA
info_cmd = subprocess.run('git describe --tags --exact-match', shell=True, text=True, cwd=path, capture_output=True)
info_cmd = info_cmd.stdout.strip()

if not info_cmd.strip():
glyph = utils.GLYPHS["branch"]
glyph = glyphs("branch")
color = CYAN
info_cmd = subprocess.run('git rev-parse --short HEAD', shell=True, text=True, cwd=path, capture_output=True)
info_cmd = info_cmd.stdout.strip()

return f'{color}{glyph}{RESET}{utils.GLYPHS["space"]}{DIM}{branch_stdout}{RESET}:{info_cmd}'
return f'{color}{glyph}{RESET}{glyphs("space")}{DIM}{branch_stdout}{RESET}:{info_cmd}'
else:
return f'{CYAN}{utils.GLYPHS["branch"]}{RESET}{utils.GLYPHS["space"]}{branch_stdout}'
return f'{CYAN}{glyphs("branch")}{RESET}{glyphs("space")}{branch_stdout}'

@staticmethod
def _print_process_header(path: str, command: str, failed: bool, code: int):
path = f'{BKG_BLACK}{Runner._get_formatted_path(path)}{RESET}'
command = f'{BKG_WHITE}{BLACK}{utils.GLYPHS[")"]}{utils.GLYPHS["space"]}{utils.GLYPHS["terminal"]}{utils.GLYPHS["space"]}{BOLD}{command} {RESET}{WHITE}{RESET}'
code = f'{WHITE}{BKG_RED if failed else BKG_GREEN}{utils.GLYPHS[")"]}{BRIGHT_WHITE}{utils.GLYPHS["space"]}{utils.GLYPHS["failed"] if failed else utils.GLYPHS["finished"]} {f"Code: {BOLD}{code}" if failed else ""}{utils.GLYPHS["space"]}{RESET}{RED if failed else GREEN}{utils.GLYPHS[")"]}{RESET}'
command = f'{BKG_WHITE}{BLACK}{glyphs(")")}{glyphs("space")}{glyphs("terminal")}{glyphs("space")}{BOLD}{command} {RESET}{WHITE}{RESET}'
code = f'{WHITE}{BKG_RED if failed else BKG_GREEN}{glyphs(")")}{BRIGHT_WHITE}{glyphs("space")}{glyphs("failed") if failed else glyphs("finished")} {f"Code: {BOLD}{code}" if failed else ""}{glyphs("space")}{RESET}{RED if failed else GREEN}{glyphs(")")}{RESET}'
print(f'{path} {command}{code}')

@staticmethod
Expand Down Expand Up @@ -338,7 +339,7 @@ def _get_formatted_labels(labels: List[str], glyph: str) -> str:
colored_label = ''
for label in labels:
color_index = Runner._get_color_index(label) % len(TEXT)
colored_label += f'{TEXT[color_index + 3]}{glyph}{utils.GLYPHS["space"]}{label}{RESET} '
colored_label += f'{TEXT[color_index + 3]}{glyph}{glyphs("space")}{label}{RESET} '
return colored_label

@staticmethod
Expand All @@ -356,13 +357,13 @@ def _get_formatted_branches(branches: List[str], current_branch: str) -> str:
current_prefix = current_prefix + DIM if is_origin else current_prefix
origin_prefix = f'{MAGENTA}{DIM}o/' if is_origin else ''
color = WHITE
icon = utils.GLYPHS['branch']
icon = glyphs('branch')
if branch == 'master' or branch == 'main':
color = YELLOW
icon = f'{utils.GLYPHS["master"]}'
icon = glyphs("master")
elif branch == 'develop':
color = GREEN
icon = f'{utils.GLYPHS["feature"]}'
icon = glyphs("feature")
elif '/' in branch:
parts = branch.split('/')
end_dim = '' if is_origin else END_DIM
Expand All @@ -372,22 +373,30 @@ def _get_formatted_branches(branches: List[str], current_branch: str) -> str:
branch = f'{DIM}{branch}'
color = Runner._get_branch_color(parts[0])
icon = Runner._get_branch_icon(parts[0])
output += f'{current_prefix}{color}{icon}{utils.GLYPHS["space"]}{origin_prefix}{color}{branch}{RESET} '
output += f'{current_prefix}{color}{icon}{glyphs("space")}{origin_prefix}{color}{branch}{RESET} '
return output

@staticmethod
def _get_branch_icon(branch_prefix: str) -> str:
return f'{utils.GLYPHS["bugfix"]}' if branch_prefix in ['bugfix', 'bug', 'hotfix'] else \
f'{utils.GLYPHS["release"]}' if branch_prefix == 'release' else \
f'{utils.GLYPHS["feature"]}' if branch_prefix in ['feature', 'feat', 'develop'] else \
f'{utils.GLYPHS["branch"]}'
if branch_prefix in ['bugfix', 'bug', 'hotfix']:
return glyphs("bugfix")
elif branch_prefix == 'release':
return glyphs("release")
elif branch_prefix in ['feature', 'feat', 'develop']:
return glyphs("feature")
else:
return glyphs("branch")

@staticmethod
def _get_branch_color(branch_name: str) -> str:
return RED if branch_name in ['bugfix', 'bug', 'hotfix'] else \
BLUE if branch_name == 'release' else \
GREEN if branch_name in ['feature', 'feat', 'develop'] else \
GREEN
if branch_name in ['bugfix', 'bug', 'hotfix']:
return RED
elif branch_name == 'release':
return BLUE
elif branch_name in ['feature', 'feat', 'develop']:
return GREEN
else:
return GREEN

@staticmethod
def _get_color_index(label: str) -> (str, str):
Expand Down
18 changes: 7 additions & 11 deletions utils.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
import sys
import shutil
import random
import shutil
import subprocess
import sys

from styles import *
from settings import *
from prettytable import PrettyTable, PLAIN_COLUMNS

from settings import *
from styles import *

SETTINGS_FILE_NAME = '.mudsettings'
CONFIG_FILE_NAME = '.mudconfig'
GLYPHS = {}
ICON_GLYPHS = {
'ahead': '\uf062',
'behind': '\uf063',
Expand Down Expand Up @@ -62,12 +62,8 @@
settings: Settings


def setup():
global GLYPHS
GLYPHS = ICON_GLYPHS if settings.mud_settings['nerd_fonts'] else TEXT_GLYPHS

if settings.config['mud'].getboolean('ask_updates') and update():
sys.exit()
def glyphs(key: str) -> str:
return (ICON_GLYPHS if settings.mud_settings['nerd_fonts'] else TEXT_GLYPHS)[key]


def version() -> None:
Expand Down

0 comments on commit bb97c78

Please sign in to comment.