2024-06-26 18:26:56 +02:00
#!/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')
2024-06-27 10:39:17 +02:00
parser.add_argument('--repos-location', type=SshLocation, default='graffy@store.ipr.univ-rennes.fr:/mnt/store.ipr/InstallProgs/ipr/tickets-details', help='the tickets details repository location')
2024-06-26 18:26:56 +02:00
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)
