import sys import argparse from pathlib import Path from ..configuration import Config from ..chat import ChatDB from ..message import MessageFilter, Message msg_suffix = Message.file_suffix_write # currently '.msg' def convert_messages(args: argparse.Namespace, config: Config) -> None: """ Convert messages to a new format. Also used to change old suffixes ('.txt', '.yaml') to the latest default message file suffix ('.msg'). """ chat = ChatDB.from_dir(Path(config.cache), Path(config.db)) # read all known message files msgs = chat.msg_gather(loc='disk', glob='*.*') # make a set of all message IDs msg_ids = set([m.msg_id() for m in msgs]) # set requested format and write all messages chat.set_msg_format(args.convert) # delete the current suffix # -> a new one will automatically be created for m in msgs: if m.file_path: m.file_path = m.file_path.with_suffix('') chat.msg_write(msgs) # read all messages with the current default suffix msgs = chat.msg_gather(loc='disk', glob=f'*{msg_suffix}') # make sure we converted all of the original messages for mid in msg_ids: if not any(mid == m.msg_id() for m in msgs): print(f"Message '{mid}' has not been found after conversion. Aborting.") sys.exit(1) # delete messages with old suffixes msgs = chat.msg_gather(loc='disk', glob='*.*') for m in msgs: if m.file_path and m.file_path.suffix != msg_suffix: m.rm_file() print(f"Successfully converted {len(msg_ids)} messages.") def print_chat(args: argparse.Namespace, config: Config) -> None: """ Print the DB chat history. """ mfilter = MessageFilter(tags_or=args.or_tags, tags_and=args.and_tags, tags_not=args.exclude_tags, question_contains=args.question, answer_contains=args.answer) chat = ChatDB.from_dir(Path(config.cache), Path(config.db), mfilter=mfilter) chat.print(args.source_code_only, args.with_metadata) def hist_cmd(args: argparse.Namespace, config: Config) -> None: """ Handler for the 'hist' command. """ if args.print: print_chat(args, config) elif args.convert: convert_messages(args, config)