#!/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)