diff --git a/home/bin/ticman b/home/bin/ticman new file mode 100755 index 0000000..a695c5f --- /dev/null +++ b/home/bin/ticman @@ -0,0 +1,173 @@ +#!/usr/bin/env python3 +from typing import List +import logging +import argparse +import re +import os +from textwrap import dedent +from enum import Enum +from pathlib import Path +# from shutil import rmtree +from subprocess import run +from datetime import datetime +import getpass +import socket + +__version__ = '1.0.0' + +TicketId = str # eg bug3879 +HostFqdn = str # eg store.ipr.univ-rennes.fr +UserId = str # eg 'graffy' +SshLocation = str # eg graffy@store.ipr.univ-rennes.fr:/ + + +class UserError(Exception): + pass + + +class Action(Enum): + CHECK_IN = 'checkin' + CHECK_OUT = 'checkout' + PUSH = 'push' + PULL = 'pull' + LIST = 'list' + + +class TicketDbLocation(): + """the location of the ticket details database""" + ssh_user: UserId + ssh_host_fqdn: HostFqdn + db_root_path: Path + + def __init__(self, ssh_user: UserId, ssh_host_fqdn: HostFqdn, db_root_path: Path): + self.ssh_user = ssh_user + self.ssh_host_fqdn = ssh_host_fqdn + self.db_root_path = db_root_path + + def as_ssh_location(self) -> SshLocation: + return f'{self.ssh_user}@{self.ssh_host_fqdn}:{self.db_root_path}' + + def as_ssh_target(self) -> str: + return f'{self.ssh_user}@{self.ssh_host_fqdn}' + + +def delete_tree(root_dir: Path): + _completed_process = run(['trash', str(root_dir)], check=True) # noqa: F841 + # rmtree(root_dir) + + +class TicketManager(): + db_location: TicketDbLocation + tickets_local_path: Path + + def __init__(self, db_location: TicketDbLocation, tickets_local_path: Path): + self.db_location = db_location + self.tickets_local_path = tickets_local_path + + def get_repos_tickets_ids(self) -> List[TicketId]: + _completed_process = run(['ssh', self.db_location.as_ssh_target(), f'ls -d {self.db_location.db_root_path}/*/'], check=True, capture_output=True) # noqa: F841 + dirs = re.sub('\n$', '', _completed_process.stdout.decode('utf8')).split('\n') + ticket_ids = [TicketId(Path(re.sub('/$', '', dir)).name) for dir in dirs] + return ticket_ids + + def list_repos_tickets(self): + repos_ticket_ids = self.get_repos_tickets_ids() + for ticket_id in repos_ticket_ids: + print(ticket_id) + + def push(self, ticket_id: TicketId): + _completed_process = run(['rsync', '-va', self.get_ticket_local_path(ticket_id), self.db_location.as_ssh_location()], check=True, capture_output=True) # noqa: F841 + + def checkin(self, ticket_id: TicketId): + ticket_local_dir = self.get_ticket_local_path(ticket_id) + if not ticket_local_dir.exists(): + raise UserError(f'impossible to to check-in {ticket_local_dir} as this directory doesn\'t exist') + # append the checkin time to a history file + time_stamp_file_path = ticket_local_dir / 'ticman-history.tsv' + if not time_stamp_file_path.exists(): + with open(time_stamp_file_path, 'wt', encoding='utf8') as time_stamp_file: + time_stamp_file.write('#date\tcheckin-origin\n') + with open(time_stamp_file_path, 'at', encoding='utf8') as time_stamp_file: + checkin_origin = f'{getpass.getuser()}@{socket.getfqdn()}:{ticket_local_dir}' + time_stamp_file.write(f'{datetime.now().astimezone().isoformat()}\t{checkin_origin}\n') + self.push(ticket_id) + delete_tree(ticket_local_dir) + print(f'ticket {ticket_id} successfully checked in to {self.db_location.as_ssh_location()}, and {ticket_local_dir} has been deleted') + + def checkout(self, ticket_id: TicketId): + repos_ticket_ids = self.get_repos_tickets_ids() + if ticket_id in repos_ticket_ids: + ticket_local_dir = self.get_ticket_local_path(ticket_id) + if ticket_local_dir.exists(): + raise UserError(f'impossible to to checkout {ticket_local_dir} as it has already been checked out ({ticket_local_dir} already exists). Please make sure you delete this directory (possibly backing it up first) before checking it out.') + os.makedirs(ticket_local_dir, exist_ok=True) + _completed_process = run(['rsync', '-va', f'{self.db_location.as_ssh_location()}/{ticket_id}/', f'{ticket_local_dir}/'], check=True, capture_output=True) # noqa: F841 + print(f'ticket {ticket_id} successfully checked out to {ticket_local_dir}') + else: + raise UserError(f'impossible to checkout ticket {ticket_id} as it doesn\'t exist in the ticket details repository') + + def get_ticket_local_path(self, ticket_id: TicketId) -> Path: + return self.tickets_local_path / ticket_id + + +def main(): + logging.basicConfig(level=logging.DEBUG) # , format='%(asctime)s - %(levelname)s - %(message)s') + description = dedent('''\ + ipr ticket manager - a tool to help centralizing ipr tickets details (additional files that are not appropriate as attachment to ticketing system) + + The repository of tickets details is stored on a shared disk, and the user can checkout a ticket detail when he wants to get a local copy in a workspace located on his workstation. Once the user has finished working on a ticket, he can checkin this ticket to the repository to save space on his workstation. + + Ideally we would use a git repository as a ticket detail repository but the data would be too bog for a git repository, so ticman uses a simple file hierarchy. + + ''') + # parser = argparse.ArgumentParser(description='ipr ticket manager - a tool to help centralizing ipr tickets details (additional files that are not appropriate as attachment to ticketing system)', formatter_class=argparse.RawDescriptionHelpFormatter) + parser = argparse.ArgumentParser(description=description, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument('--workspace-dir', type=Path, default=Path('/home/graffy/work/tickets'), help='the local directory used as a workspace') + parser.add_argument('--repos-location', type=SshLocation, default='graffy@store.ipr.univ-rennes.fr:/mnt/store.ipr/InstallProgs/ipr/tickets', help='the tickets details repository location') + + subparsers = parser.add_subparsers(dest='cmd', required=True, description='action to perform') + # parser.add_argument('--ticket-id', type=str, required=True, help='the identifier of the ticket on which the action is performed (eg bug3879)') + + parser.add_argument('--version', action='version', help=f'shows {parser.prog}\'s version', version=f'{parser.prog} version {__version__}') + + # parser.add_argument('action', type=Action, help=f'synchronizes the global tickets detail database from the local workspace') + epilog = dedent(f'''\ + examples: + {parser.prog} checkout bug3879 + + ''') + parser.epilog = epilog + + _list_parser = subparsers.add_parser('list', description='lists the tickets in the tickets detail repository') # noqa: F841 + + _checkout_parser = subparsers.add_parser('checkout', description='create a local copy of the given ticket from the database') # noqa: F841 + _checkout_parser.add_argument('ticketid', type=TicketId, help='the identifier of the ticket on which the action is performed (eg bug3879)') + + _checkin_parser = subparsers.add_parser('checkin', description='updates the global tickets detail database from the local workspace') # noqa: F841 + _checkin_parser.add_argument('ticketid', type=TicketId, help='the identifier of the ticket on which the action is performed (eg bug3879)') + + _pull_parser = subparsers.add_parser('pull', description='updates the global tickets detail database from the local workspace ticket') # noqa: F841 + + _push_parser = subparsers.add_parser('push', description='updates the local workspace from the global tickets detail database') # noqa: F841 + _push_parser.add_argument('ticketid', type=TicketId, help='the identifier of the ticket on which the action is performed (eg bug3879)') + + args = parser.parse_args() + + db_location = TicketDbLocation(ssh_user='graffy', ssh_host_fqdn='store.ipr.univ-rennes.fr', db_root_path=Path('/mnt/store.ipr/InstallProgs/ipr/tickets')) + tic_man = TicketManager(db_location, tickets_local_path=args.workspace_dir) + + if args.cmd == 'list': + tic_man.list_repos_tickets() + if args.cmd == 'checkin': + tic_man.checkin(args.ticketid) + elif args.cmd == 'checkout': + tic_man.checkout(args.ticketid) + elif args.cmd == 'push': + tic_man.push(args.ticketid) + + +try: + main() +except UserError as err: + print(err) + exit(1)