Compare commits
8 Commits
a2ae52014b
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 9a957a89ac | |||
| 5d1bb1f9e4 | |||
| 75a123eb72 | |||
| 7c1c67f8ff | |||
| dbe72ff11c | |||
| bbc1ab5a0a | |||
| 2aee018708 | |||
| 17c6fa2453 |
@@ -47,12 +47,12 @@ class OpenAIAnswer:
|
|||||||
self.finished = True
|
self.finished = True
|
||||||
if not self.finished:
|
if not self.finished:
|
||||||
found_choice = False
|
found_choice = False
|
||||||
for choice in chunk['choices']:
|
for choice in chunk.choices:
|
||||||
if not choice['finish_reason']:
|
if not choice.finish_reason:
|
||||||
self.streams[choice['index']].data.append(choice['delta']['content'])
|
self.streams[choice.index].data.append(choice.delta.content)
|
||||||
self.tokens.completion += len(self.encoding.encode(choice['delta']['content']))
|
self.tokens.completion += len(self.encoding.encode(choice.delta.content))
|
||||||
self.tokens.total = self.tokens.prompt + self.tokens.completion
|
self.tokens.total = self.tokens.prompt + self.tokens.completion
|
||||||
if choice['index'] == self.idx:
|
if choice.index == self.idx:
|
||||||
found_choice = True
|
found_choice = True
|
||||||
if not found_choice:
|
if not found_choice:
|
||||||
return False
|
return False
|
||||||
@@ -68,7 +68,10 @@ class OpenAI(AI):
|
|||||||
self.ID = config.ID
|
self.ID = config.ID
|
||||||
self.name = config.name
|
self.name = config.name
|
||||||
self.config = config
|
self.config = config
|
||||||
openai.api_key = self.config.api_key
|
self.client = openai.OpenAI(api_key=self.config.api_key)
|
||||||
|
|
||||||
|
def _completions(self, *args, **kw): # type: ignore
|
||||||
|
return self.client.chat.completions.create(*args, **kw)
|
||||||
|
|
||||||
def request(self,
|
def request(self,
|
||||||
question: Message,
|
question: Message,
|
||||||
@@ -83,7 +86,7 @@ class OpenAI(AI):
|
|||||||
self.encoding = tiktoken.encoding_for_model(self.config.model)
|
self.encoding = tiktoken.encoding_for_model(self.config.model)
|
||||||
oai_chat, prompt_tokens = self.openai_chat(chat, self.config.system, question)
|
oai_chat, prompt_tokens = self.openai_chat(chat, self.config.system, question)
|
||||||
tokens: Tokens = Tokens(prompt_tokens, 0, prompt_tokens)
|
tokens: Tokens = Tokens(prompt_tokens, 0, prompt_tokens)
|
||||||
response = openai.ChatCompletion.create(
|
response = self._completions(
|
||||||
model=self.config.model,
|
model=self.config.model,
|
||||||
messages=oai_chat,
|
messages=oai_chat,
|
||||||
temperature=self.config.temperature,
|
temperature=self.config.temperature,
|
||||||
@@ -114,9 +117,8 @@ class OpenAI(AI):
|
|||||||
Return all models supported by this AI.
|
Return all models supported by this AI.
|
||||||
"""
|
"""
|
||||||
ret = []
|
ret = []
|
||||||
for engine in sorted(openai.Engine.list()['data'], key=lambda x: x['id']):
|
for engine in sorted(self.client.models.list().data, key=lambda x: x.id):
|
||||||
if engine['ready']:
|
ret.append(engine.id)
|
||||||
ret.append(engine['id'])
|
|
||||||
ret.sort()
|
ret.sort()
|
||||||
return ret
|
return ret
|
||||||
|
|
||||||
@@ -124,14 +126,8 @@ class OpenAI(AI):
|
|||||||
"""
|
"""
|
||||||
Print all models supported by the current AI.
|
Print all models supported by the current AI.
|
||||||
"""
|
"""
|
||||||
not_ready = []
|
for model in self.models():
|
||||||
for engine in sorted(openai.Engine.list()['data'], key=lambda x: x['id']):
|
print(model)
|
||||||
if engine['ready']:
|
|
||||||
print(engine['id'])
|
|
||||||
else:
|
|
||||||
not_ready.append(engine['id'])
|
|
||||||
if len(not_ready) > 0:
|
|
||||||
print('\nNot ready: ' + ', '.join(not_ready))
|
|
||||||
|
|
||||||
def openai_chat(self, chat: Chat, system: str,
|
def openai_chat(self, chat: Chat, system: str,
|
||||||
question: Optional[Message] = None) -> tuple[ChatType, int]:
|
question: Optional[Message] = None) -> tuple[ChatType, int]:
|
||||||
|
|||||||
@@ -1,69 +0,0 @@
|
|||||||
"""
|
|
||||||
Contains shared functions for the various CMM subcommands.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import argparse
|
|
||||||
from pathlib import Path
|
|
||||||
from ..message import Message, MessageError, source_code
|
|
||||||
|
|
||||||
|
|
||||||
def read_text_file(file: Path) -> str:
|
|
||||||
with open(file) as r:
|
|
||||||
content = r.read().strip()
|
|
||||||
return content
|
|
||||||
|
|
||||||
|
|
||||||
def add_file_as_text(question_parts: list[str], file: str) -> None:
|
|
||||||
"""
|
|
||||||
Add the given file as plain text to the question part list.
|
|
||||||
If the file is a Message, add the answer.
|
|
||||||
"""
|
|
||||||
file_path = Path(file)
|
|
||||||
content: str
|
|
||||||
try:
|
|
||||||
message = Message.from_file(file_path)
|
|
||||||
if message and message.answer:
|
|
||||||
content = message.answer
|
|
||||||
except MessageError:
|
|
||||||
content = read_text_file(Path(file))
|
|
||||||
if len(content) > 0:
|
|
||||||
question_parts.append(content)
|
|
||||||
|
|
||||||
|
|
||||||
def add_file_as_code(question_parts: list[str], file: str) -> None:
|
|
||||||
"""
|
|
||||||
Add all source code from the given file. If no code segments can be extracted,
|
|
||||||
the whole content is added as source code segment. If the file is a Message,
|
|
||||||
extract the source code from the answer.
|
|
||||||
"""
|
|
||||||
file_path = Path(file)
|
|
||||||
content: str
|
|
||||||
try:
|
|
||||||
message = Message.from_file(file_path)
|
|
||||||
if message and message.answer:
|
|
||||||
content = message.answer
|
|
||||||
except MessageError:
|
|
||||||
with open(file) as r:
|
|
||||||
content = r.read().strip()
|
|
||||||
# extract and add source code
|
|
||||||
code_parts = source_code(content, include_delims=True)
|
|
||||||
if len(code_parts) > 0:
|
|
||||||
question_parts += code_parts
|
|
||||||
else:
|
|
||||||
question_parts.append(f"```\n{content}\n```")
|
|
||||||
|
|
||||||
|
|
||||||
def invert_input_tag_args(args: argparse.Namespace) -> None:
|
|
||||||
"""
|
|
||||||
Changes the semantics of the INPUT tags for this command:
|
|
||||||
* not tags specified on the CLI -> no tags are selected
|
|
||||||
* empty tags specified on the CLI -> all tags are selected
|
|
||||||
"""
|
|
||||||
if args.or_tags is None:
|
|
||||||
args.or_tags = set()
|
|
||||||
elif len(args.or_tags) == 0:
|
|
||||||
args.or_tags = None
|
|
||||||
if args.and_tags is None:
|
|
||||||
args.and_tags = set()
|
|
||||||
elif len(args.and_tags) == 0:
|
|
||||||
args.and_tags = None
|
|
||||||
@@ -3,10 +3,9 @@ import argparse
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from itertools import zip_longest
|
from itertools import zip_longest
|
||||||
from copy import deepcopy
|
from copy import deepcopy
|
||||||
from .common import invert_input_tag_args, add_file_as_code, add_file_as_text
|
|
||||||
from ..configuration import Config
|
from ..configuration import Config
|
||||||
from ..chat import ChatDB, msg_location
|
from ..chat import ChatDB, msg_location
|
||||||
from ..message import Message, MessageFilter, Question
|
from ..message import Message, MessageFilter, MessageError, Question, source_code
|
||||||
from ..ai_factory import create_ai
|
from ..ai_factory import create_ai
|
||||||
from ..ai import AI, AIResponse
|
from ..ai import AI, AIResponse
|
||||||
|
|
||||||
@@ -15,6 +14,47 @@ class QuestionCmdError(Exception):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def add_file_as_text(question_parts: list[str], file: str) -> None:
|
||||||
|
"""
|
||||||
|
Add the given file as plain text to the question part list.
|
||||||
|
If the file is a Message, add the answer.
|
||||||
|
"""
|
||||||
|
file_path = Path(file)
|
||||||
|
content: str
|
||||||
|
try:
|
||||||
|
message = Message.from_file(file_path)
|
||||||
|
if message and message.answer:
|
||||||
|
content = message.answer
|
||||||
|
except MessageError:
|
||||||
|
with open(file) as r:
|
||||||
|
content = r.read().strip()
|
||||||
|
if len(content) > 0:
|
||||||
|
question_parts.append(content)
|
||||||
|
|
||||||
|
|
||||||
|
def add_file_as_code(question_parts: list[str], file: str) -> None:
|
||||||
|
"""
|
||||||
|
Add all source code from the given file. If no code segments can be extracted,
|
||||||
|
the whole content is added as source code segment. If the file is a Message,
|
||||||
|
extract the source code from the answer.
|
||||||
|
"""
|
||||||
|
file_path = Path(file)
|
||||||
|
content: str
|
||||||
|
try:
|
||||||
|
message = Message.from_file(file_path)
|
||||||
|
if message and message.answer:
|
||||||
|
content = message.answer
|
||||||
|
except MessageError:
|
||||||
|
with open(file) as r:
|
||||||
|
content = r.read().strip()
|
||||||
|
# extract and add source code
|
||||||
|
code_parts = source_code(content, include_delims=True)
|
||||||
|
if len(code_parts) > 0:
|
||||||
|
question_parts += code_parts
|
||||||
|
else:
|
||||||
|
question_parts.append(f"```\n{content}\n```")
|
||||||
|
|
||||||
|
|
||||||
def create_msg_args(msg: Message, args: argparse.Namespace) -> argparse.Namespace:
|
def create_msg_args(msg: Message, args: argparse.Namespace) -> argparse.Namespace:
|
||||||
"""
|
"""
|
||||||
Takes an existing message and CLI arguments, and returns modified args based
|
Takes an existing message and CLI arguments, and returns modified args based
|
||||||
@@ -123,6 +163,22 @@ def repeat_messages(messages: list[Message], chat: ChatDB, args: argparse.Namesp
|
|||||||
make_request(ai, chat, message, msg_args)
|
make_request(ai, chat, message, msg_args)
|
||||||
|
|
||||||
|
|
||||||
|
def invert_input_tag_args(args: argparse.Namespace) -> None:
|
||||||
|
"""
|
||||||
|
Changes the semantics of the INPUT tags for this command:
|
||||||
|
* not tags specified on the CLI -> no tags are selected
|
||||||
|
* empty tags specified on the CLI -> all tags are selected
|
||||||
|
"""
|
||||||
|
if args.or_tags is None:
|
||||||
|
args.or_tags = set()
|
||||||
|
elif len(args.or_tags) == 0:
|
||||||
|
args.or_tags = None
|
||||||
|
if args.and_tags is None:
|
||||||
|
args.and_tags = set()
|
||||||
|
elif len(args.and_tags) == 0:
|
||||||
|
args.and_tags = None
|
||||||
|
|
||||||
|
|
||||||
def question_cmd(args: argparse.Namespace, config: Config) -> None:
|
def question_cmd(args: argparse.Namespace, config: Config) -> None:
|
||||||
"""
|
"""
|
||||||
Handler for the 'question' command.
|
Handler for the 'question' command.
|
||||||
|
|||||||
@@ -1,105 +0,0 @@
|
|||||||
import argparse
|
|
||||||
import mimetypes
|
|
||||||
from pathlib import Path
|
|
||||||
from .common import invert_input_tag_args, read_text_file
|
|
||||||
from ..configuration import Config
|
|
||||||
from ..message import MessageFilter, Message, Question
|
|
||||||
from ..chat import ChatDB, msg_location
|
|
||||||
|
|
||||||
|
|
||||||
class TranslationCmdError(Exception):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
text_separator: str = 'TEXT:'
|
|
||||||
|
|
||||||
|
|
||||||
def assert_document_type_supported_openai(document_file: Path) -> None:
|
|
||||||
doctype = mimetypes.guess_type(document_file)
|
|
||||||
if doctype != 'text/plain':
|
|
||||||
raise TranslationCmdError("AI 'OpenAI' only supports document type 'text/plain''")
|
|
||||||
|
|
||||||
|
|
||||||
def translation_prompt_openai(source_lang: str, target_lang: str) -> str:
|
|
||||||
"""
|
|
||||||
Return the prompt for GPT that tells it to do the translation.
|
|
||||||
"""
|
|
||||||
return f"Translate the text below the line {text_separator} from {source_lang} to {target_lang}."
|
|
||||||
|
|
||||||
|
|
||||||
def create_message_openai(chat: ChatDB, args: argparse.Namespace) -> Message:
|
|
||||||
"""
|
|
||||||
Create a new message from the given arguments and write it to the cache directory.
|
|
||||||
|
|
||||||
Message format
|
|
||||||
1. Translation prompt (tells GPT to do a translation)
|
|
||||||
2. Glossary (if specified as an argument)
|
|
||||||
3. User provided prompt enhancements
|
|
||||||
4. Translation separator
|
|
||||||
5. User provided text to be translated
|
|
||||||
|
|
||||||
The text to be translated is determined as a follows:
|
|
||||||
- if a document is provided in the arguments, translate its content
|
|
||||||
- if no document is provided, translate the last text argument
|
|
||||||
|
|
||||||
The other text arguments will be put into the "header" and can be used
|
|
||||||
to improve the translation prompt.
|
|
||||||
"""
|
|
||||||
text_args: list[str] = []
|
|
||||||
if args.create is not None:
|
|
||||||
text_args = args.create
|
|
||||||
elif args.ask is not None:
|
|
||||||
text_args = args.ask
|
|
||||||
else:
|
|
||||||
raise TranslationCmdError("No input text found")
|
|
||||||
|
|
||||||
# extract user prompt and user text to be translated
|
|
||||||
user_text: str
|
|
||||||
user_prompt: str
|
|
||||||
if args.input_document is not None:
|
|
||||||
assert_document_type_supported_openai(Path(args.input_document))
|
|
||||||
user_text = read_text_file(Path(args.input_document))
|
|
||||||
user_prompt = '\n\n'.join([str(s) for s in text_args])
|
|
||||||
else:
|
|
||||||
user_text = text_args[-1]
|
|
||||||
user_prompt = '\n\n'.join([str(s) for s in text_args[:-1]])
|
|
||||||
|
|
||||||
# build full question string
|
|
||||||
# FIXME: add glossaries if given
|
|
||||||
question_text: str = '\n\n'.join([translation_prompt_openai(args.source_lang, args.target_lang),
|
|
||||||
user_prompt,
|
|
||||||
text_separator,
|
|
||||||
user_text])
|
|
||||||
# create and write the message
|
|
||||||
message = Message(question=Question(question_text),
|
|
||||||
tags=args.output_tags,
|
|
||||||
ai=args.AI,
|
|
||||||
model=args.model)
|
|
||||||
# only write the new message to the cache,
|
|
||||||
# don't add it to the internal list
|
|
||||||
chat.cache_write([message])
|
|
||||||
return message
|
|
||||||
|
|
||||||
|
|
||||||
def translation_cmd(args: argparse.Namespace, config: Config) -> None:
|
|
||||||
"""
|
|
||||||
Handler for the 'translation' command. Creates and executes translation
|
|
||||||
requests based on the input and selected AI. Depending on the AI, the
|
|
||||||
whole process may be significantly different (e.g. DeepL vs OpenAI).
|
|
||||||
"""
|
|
||||||
invert_input_tag_args(args)
|
|
||||||
mfilter = MessageFilter(tags_or=args.or_tags,
|
|
||||||
tags_and=args.and_tags,
|
|
||||||
tags_not=args.exclude_tags)
|
|
||||||
chat = ChatDB.from_dir(cache_path=Path(config.cache),
|
|
||||||
db_path=Path(config.db),
|
|
||||||
mfilter=mfilter,
|
|
||||||
glob=args.glob,
|
|
||||||
loc=msg_location(args.location))
|
|
||||||
# if it's a new translation, create and store it immediately
|
|
||||||
# FIXME: check AI type
|
|
||||||
if args.ask or args.create:
|
|
||||||
# message = create_message(chat, args)
|
|
||||||
create_message_openai(chat, args)
|
|
||||||
if args.create:
|
|
||||||
return
|
|
||||||
@@ -1,138 +0,0 @@
|
|||||||
"""
|
|
||||||
Module implementing glossaries for translations.
|
|
||||||
"""
|
|
||||||
import yaml
|
|
||||||
import tempfile
|
|
||||||
import shutil
|
|
||||||
import csv
|
|
||||||
from pathlib import Path
|
|
||||||
from dataclasses import dataclass, field
|
|
||||||
from typing import Type, TypeVar, ClassVar
|
|
||||||
|
|
||||||
GlossaryInst = TypeVar('GlossaryInst', bound='Glossary')
|
|
||||||
|
|
||||||
|
|
||||||
class GlossaryError(Exception):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def str_presenter(dumper: yaml.Dumper, data: str) -> yaml.ScalarNode:
|
|
||||||
"""
|
|
||||||
Changes the YAML dump style to multiline syntax for multiline strings.
|
|
||||||
"""
|
|
||||||
if len(data.splitlines()) > 1:
|
|
||||||
return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
|
|
||||||
return dumper.represent_scalar('tag:yaml.org,2002:str', data)
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Glossary:
|
|
||||||
"""
|
|
||||||
A glossary consists of the following parameters:
|
|
||||||
- Name (freely selectable)
|
|
||||||
- Path (full file path)
|
|
||||||
- Source language
|
|
||||||
- Target language
|
|
||||||
- Entries (pairs of source lang and target lang terms)
|
|
||||||
- ID (automatically generated / modified, required by DeepL)
|
|
||||||
"""
|
|
||||||
|
|
||||||
name: str
|
|
||||||
source_lang: str
|
|
||||||
target_lang: str
|
|
||||||
entries: dict[str, str] = field(default_factory=lambda: dict())
|
|
||||||
file_path: Path | None = None
|
|
||||||
ID: str | None = None
|
|
||||||
file_suffix: ClassVar[str] = '.glo'
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_file(cls: Type[GlossaryInst], file_path: Path) -> GlossaryInst:
|
|
||||||
"""
|
|
||||||
Create a glossary from the given file.
|
|
||||||
"""
|
|
||||||
if not file_path.exists():
|
|
||||||
raise GlossaryError(f"Glossary file '{file_path}' does not exist")
|
|
||||||
if file_path.suffix != cls.file_suffix:
|
|
||||||
raise GlossaryError(f"File type '{file_path.suffix}' is not supported")
|
|
||||||
with open(file_path, "r") as fd:
|
|
||||||
try:
|
|
||||||
data = yaml.load(fd, Loader=yaml.FullLoader)
|
|
||||||
# remove any quotes from the entries that YAML may have added while dumping
|
|
||||||
# (e. g. for special keywords like 'yes')
|
|
||||||
clean_entries = {key.strip('\"\' '): value for key, value in data['Entries'].items()}
|
|
||||||
return cls(name=data['Name'],
|
|
||||||
source_lang=data['SourceLang'],
|
|
||||||
target_lang=data['TargetLang'],
|
|
||||||
entries=clean_entries,
|
|
||||||
file_path=file_path,
|
|
||||||
ID=data['ID'] if data['ID'] != 'None' else None)
|
|
||||||
except Exception:
|
|
||||||
raise GlossaryError(f"'{file_path}' does not contain a valid glossary")
|
|
||||||
|
|
||||||
def to_file(self, file_path: Path | None = None) -> None:
|
|
||||||
"""
|
|
||||||
Write glossary to given file.
|
|
||||||
"""
|
|
||||||
if file_path:
|
|
||||||
self.file_path = file_path
|
|
||||||
if not self.file_path:
|
|
||||||
raise GlossaryError("Got no valid path to write glossary")
|
|
||||||
# check / add valid suffix
|
|
||||||
if not self.file_path.suffix:
|
|
||||||
self.file_path = self.file_path.with_suffix(self.file_suffix)
|
|
||||||
elif self.file_path.suffix != self.file_suffix:
|
|
||||||
raise GlossaryError(f"File suffix '{self.file_path.suffix}' is not supported")
|
|
||||||
# write YAML
|
|
||||||
with tempfile.NamedTemporaryFile(dir=self.file_path.parent, prefix=self.file_path.name, mode="w", delete=False) as temp_fd:
|
|
||||||
temp_file_path = Path(temp_fd.name)
|
|
||||||
data = {'Name': self.name,
|
|
||||||
'ID': str(self.ID),
|
|
||||||
'SourceLang': self.source_lang,
|
|
||||||
'TargetLang': self.target_lang,
|
|
||||||
'Entries': self.entries}
|
|
||||||
yaml.dump(data, temp_fd, sort_keys=False)
|
|
||||||
shutil.move(temp_file_path, self.file_path)
|
|
||||||
|
|
||||||
def export_csv(self, dictionary: dict[str, str], file_path: Path) -> None:
|
|
||||||
"""
|
|
||||||
Export the 'entries' of this glossary to a file in CSV format (compatible with DeepL).
|
|
||||||
"""
|
|
||||||
with open(file_path, 'w', newline='', encoding='utf-8') as csvfile:
|
|
||||||
writer = csv.writer(csvfile, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)
|
|
||||||
for source_entry, target_entry in self.entries.items():
|
|
||||||
writer.writerow([source_entry, target_entry])
|
|
||||||
|
|
||||||
def export_tsv(self, entries: dict[str, str], file_path: Path) -> None:
|
|
||||||
"""
|
|
||||||
Export the 'entries' of this glossary to a file in TSV format (compatible with DeepL).
|
|
||||||
"""
|
|
||||||
with open(file_path, 'w', encoding='utf-8') as file:
|
|
||||||
for source_entry, target_entry in self.entries.items():
|
|
||||||
file.write(f"{source_entry}\t{target_entry}\n")
|
|
||||||
|
|
||||||
def import_csv(self, file_path: Path) -> None:
|
|
||||||
"""
|
|
||||||
Import the entries from the given CSV file to those of the current glossary.
|
|
||||||
Existing entries are overwritten.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
with open(file_path, mode='r', encoding='utf-8') as csvfile:
|
|
||||||
reader = csv.reader(csvfile, delimiter=',', quotechar='"')
|
|
||||||
self.entries = {rows[0]: rows[1] for rows in reader if len(rows) >= 2}
|
|
||||||
except Exception as e:
|
|
||||||
raise GlossaryError(f"Error importing CSV: {e}")
|
|
||||||
|
|
||||||
def import_tsv(self, file_path: Path) -> None:
|
|
||||||
"""
|
|
||||||
Import the entries from the given CSV file to those of the current glossary.
|
|
||||||
Existing entries are overwritten.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
with open(file_path, mode='r', encoding='utf-8') as tsvfile:
|
|
||||||
self.entries = {}
|
|
||||||
for line in tsvfile:
|
|
||||||
parts = line.strip().split('\t')
|
|
||||||
if len(parts) == 2:
|
|
||||||
self.entries[parts[0]] = parts[1]
|
|
||||||
except Exception as e:
|
|
||||||
raise GlossaryError(f"Error importing TSV: {e}")
|
|
||||||
+1
-17
@@ -14,7 +14,6 @@ from .commands.tags import tags_cmd
|
|||||||
from .commands.config import config_cmd
|
from .commands.config import config_cmd
|
||||||
from .commands.hist import hist_cmd
|
from .commands.hist import hist_cmd
|
||||||
from .commands.print import print_cmd
|
from .commands.print import print_cmd
|
||||||
from .commands.translation import translation_cmd
|
|
||||||
from .chat import msg_location
|
from .chat import msg_location
|
||||||
|
|
||||||
|
|
||||||
@@ -103,7 +102,7 @@ def create_parser() -> argparse.ArgumentParser:
|
|||||||
# 'tags' command parser
|
# 'tags' command parser
|
||||||
tags_cmd_parser = cmdparser.add_parser('tags',
|
tags_cmd_parser = cmdparser.add_parser('tags',
|
||||||
help="Manage tags.",
|
help="Manage tags.",
|
||||||
aliases=['T'])
|
aliases=['t'])
|
||||||
tags_cmd_parser.set_defaults(func=tags_cmd)
|
tags_cmd_parser.set_defaults(func=tags_cmd)
|
||||||
tags_group = tags_cmd_parser.add_mutually_exclusive_group(required=True)
|
tags_group = tags_cmd_parser.add_mutually_exclusive_group(required=True)
|
||||||
tags_group.add_argument('-l', '--list', help="List all tags and their frequency",
|
tags_group.add_argument('-l', '--list', help="List all tags and their frequency",
|
||||||
@@ -137,21 +136,6 @@ def create_parser() -> argparse.ArgumentParser:
|
|||||||
print_cmd_modes.add_argument('-a', '--answer', help='Only print the answer', action='store_true')
|
print_cmd_modes.add_argument('-a', '--answer', help='Only print the answer', action='store_true')
|
||||||
print_cmd_modes.add_argument('-S', '--only-source-code', help='Only print embedded source code', action='store_true')
|
print_cmd_modes.add_argument('-S', '--only-source-code', help='Only print embedded source code', action='store_true')
|
||||||
|
|
||||||
# 'translation' command parser
|
|
||||||
translation_cmd_parser = cmdparser.add_parser('translation', parents=[ai_parser, tag_parser],
|
|
||||||
help="ask, create and repeat translations.",
|
|
||||||
aliases=['t'])
|
|
||||||
translation_cmd_parser.set_defaults(func=translation_cmd)
|
|
||||||
translation_group = translation_cmd_parser.add_mutually_exclusive_group(required=True)
|
|
||||||
translation_group.add_argument('-a', '--ask', nargs='+', help='Ask to translate the given text', metavar='TEXT')
|
|
||||||
translation_group.add_argument('-c', '--create', nargs='+', help='Create a translation', metavar='TEXT')
|
|
||||||
translation_group.add_argument('-r', '--repeat', nargs='*', help='Repeat a translation', metavar='MESSAGE')
|
|
||||||
translation_cmd_parser.add_argument('-S', '--source-lang', help="Source language", metavar="LANGUAGE", required=True)
|
|
||||||
translation_cmd_parser.add_argument('-T', '--target-lang', help="Target language", metavar="LANGUAGE", required=True)
|
|
||||||
translation_cmd_parser.add_argument('-G', '--glossaries', nargs='+', help="List of glossaries", metavar="GLOSSARY")
|
|
||||||
translation_cmd_parser.add_argument('-d', '--input-document', help="Document to translate", metavar="FILE")
|
|
||||||
translation_cmd_parser.add_argument('-D', '--output-document', help="Path for the translated document", metavar="FILE")
|
|
||||||
|
|
||||||
argcomplete.autocomplete(parser)
|
argcomplete.autocomplete(parser)
|
||||||
return parser
|
return parser
|
||||||
|
|
||||||
|
|||||||
+19
-31
@@ -9,43 +9,31 @@ from chatmastermind.configuration import OpenAIConfig
|
|||||||
|
|
||||||
class OpenAITest(unittest.TestCase):
|
class OpenAITest(unittest.TestCase):
|
||||||
|
|
||||||
@mock.patch('openai.ChatCompletion.create')
|
@mock.patch('chatmastermind.ais.openai.OpenAI._completions')
|
||||||
def test_request(self, mock_create: mock.MagicMock) -> None:
|
def test_request(self, mock_create: mock.MagicMock) -> None:
|
||||||
# Create a test instance of OpenAI
|
# Create a test instance of OpenAI
|
||||||
config = OpenAIConfig()
|
config = OpenAIConfig()
|
||||||
openai = OpenAI(config)
|
openai = OpenAI(config)
|
||||||
|
|
||||||
# Set up the mock response from openai.ChatCompletion.create
|
# Set up the mock response from openai.ChatCompletion.create
|
||||||
mock_chunk1 = {
|
class mock_obj:
|
||||||
'choices': [
|
pass
|
||||||
{
|
mock_chunk1 = mock_obj()
|
||||||
'index': 0,
|
mock_chunk1.choices = [mock_obj(), mock_obj()] # type: ignore
|
||||||
'delta': {
|
mock_chunk1.choices[0].index = 0 # type: ignore
|
||||||
'content': 'Answer 1'
|
mock_chunk1.choices[0].delta = mock_obj() # type: ignore
|
||||||
},
|
mock_chunk1.choices[0].delta.content = 'Answer 1' # type: ignore
|
||||||
'finish_reason': None
|
mock_chunk1.choices[0].finish_reason = None # type: ignore
|
||||||
},
|
mock_chunk1.choices[1].index = 1 # type: ignore
|
||||||
{
|
mock_chunk1.choices[1].delta = mock_obj() # type: ignore
|
||||||
'index': 1,
|
mock_chunk1.choices[1].delta.content = 'Answer 2' # type: ignore
|
||||||
'delta': {
|
mock_chunk1.choices[1].finish_reason = None # type: ignore
|
||||||
'content': 'Answer 2'
|
mock_chunk2 = mock_obj()
|
||||||
},
|
mock_chunk2.choices = [mock_obj(), mock_obj()] # type: ignore
|
||||||
'finish_reason': None
|
mock_chunk2.choices[0].index = 0 # type: ignore
|
||||||
}
|
mock_chunk2.choices[0].finish_reason = 'stop' # type: ignore
|
||||||
],
|
mock_chunk2.choices[1].index = 1 # type: ignore
|
||||||
}
|
mock_chunk2.choices[1].finish_reason = 'stop' # type: ignore
|
||||||
mock_chunk2 = {
|
|
||||||
'choices': [
|
|
||||||
{
|
|
||||||
'index': 0,
|
|
||||||
'finish_reason': 'stop'
|
|
||||||
},
|
|
||||||
{
|
|
||||||
'index': 1,
|
|
||||||
'finish_reason': 'stop'
|
|
||||||
}
|
|
||||||
],
|
|
||||||
}
|
|
||||||
mock_create.return_value = iter([mock_chunk1, mock_chunk2])
|
mock_create.return_value = iter([mock_chunk1, mock_chunk2])
|
||||||
|
|
||||||
# Create test data
|
# Create test data
|
||||||
|
|||||||
@@ -1,117 +0,0 @@
|
|||||||
import unittest
|
|
||||||
import tempfile
|
|
||||||
from pathlib import Path
|
|
||||||
from chatmastermind.glossary import Glossary, GlossaryError
|
|
||||||
|
|
||||||
|
|
||||||
glossary_suffix: str = Glossary.file_suffix
|
|
||||||
|
|
||||||
|
|
||||||
class TestGlossary(unittest.TestCase):
|
|
||||||
|
|
||||||
def test_from_file_valid_yaml(self) -> None:
|
|
||||||
# Prepare a temporary YAML file with valid content
|
|
||||||
with tempfile.NamedTemporaryFile('w', delete=False, suffix=glossary_suffix) as yaml_file:
|
|
||||||
yaml_file.write("Name: Sample\n"
|
|
||||||
"ID: '123'\n"
|
|
||||||
"SourceLang: en\n"
|
|
||||||
"TargetLang: es\n"
|
|
||||||
"Entries:\n"
|
|
||||||
" hello: hola\n"
|
|
||||||
" goodbye: adiós\n"
|
|
||||||
" 'yes': sí\n") # 'yes' is a YAML keyword and therefore quoted
|
|
||||||
yaml_file_path = Path(yaml_file.name)
|
|
||||||
|
|
||||||
glossary = Glossary.from_file(yaml_file_path)
|
|
||||||
self.assertEqual(glossary.name, "Sample")
|
|
||||||
self.assertEqual(glossary.source_lang, "en")
|
|
||||||
self.assertEqual(glossary.target_lang, "es")
|
|
||||||
self.assertEqual(glossary.entries, {"hello": "hola", "goodbye": "adiós", "yes": "sí"})
|
|
||||||
yaml_file_path.unlink() # Remove the temporary file
|
|
||||||
|
|
||||||
def test_to_file_writes_yaml(self) -> None:
|
|
||||||
# Create glossary instance
|
|
||||||
glossary = Glossary(name="Test", source_lang="en", target_lang="fr", entries={"yes": "oui"})
|
|
||||||
|
|
||||||
with tempfile.NamedTemporaryFile('w', delete=False, suffix=glossary_suffix) as tmp_file:
|
|
||||||
file_path = Path(tmp_file.name)
|
|
||||||
glossary.to_file(file_path)
|
|
||||||
|
|
||||||
with open(file_path, 'r') as file:
|
|
||||||
content = file.read()
|
|
||||||
|
|
||||||
self.assertIn("Name: Test", content)
|
|
||||||
self.assertIn("SourceLang: en", content)
|
|
||||||
self.assertIn("TargetLang: fr", content)
|
|
||||||
self.assertIn("Entries", content)
|
|
||||||
# 'yes' is a YAML keyword and therefore quoted
|
|
||||||
self.assertIn("'yes': oui", content)
|
|
||||||
file_path.unlink() # Remove the temporary file
|
|
||||||
|
|
||||||
def test_write_read_glossary(self) -> None:
|
|
||||||
# Create glossary instance
|
|
||||||
# -> use 'yes' in order to test if the YAML quoting is correctly removed when reading the file
|
|
||||||
glossary_write = Glossary(name="Test", source_lang="en", target_lang="fr", entries={"yes": "oui"})
|
|
||||||
|
|
||||||
with tempfile.NamedTemporaryFile('w', delete=False, suffix=glossary_suffix) as tmp_file:
|
|
||||||
file_path = Path(tmp_file.name)
|
|
||||||
glossary_write.to_file(file_path)
|
|
||||||
|
|
||||||
# create new instance from glossary file
|
|
||||||
glossary_read = Glossary.from_file(file_path)
|
|
||||||
self.assertEqual(glossary_write.name, glossary_read.name)
|
|
||||||
self.assertEqual(glossary_write.source_lang, glossary_read.source_lang)
|
|
||||||
self.assertEqual(glossary_write.target_lang, glossary_read.target_lang)
|
|
||||||
self.assertDictEqual(glossary_write.entries, glossary_read.entries)
|
|
||||||
|
|
||||||
file_path.unlink() # Remove the temporary file
|
|
||||||
|
|
||||||
def test_import_export_csv(self) -> None:
|
|
||||||
glossary = Glossary(name="Test", source_lang="en", target_lang="fr", entries={})
|
|
||||||
|
|
||||||
# First export to CSV
|
|
||||||
with tempfile.NamedTemporaryFile('w', delete=False, suffix=glossary_suffix) as csvfile:
|
|
||||||
csv_file_path = Path(csvfile.name)
|
|
||||||
glossary.entries = {"hello": "salut", "goodbye": "au revoir"}
|
|
||||||
glossary.export_csv(glossary.entries, csv_file_path)
|
|
||||||
|
|
||||||
# Now import CSV
|
|
||||||
glossary.import_csv(csv_file_path)
|
|
||||||
self.assertEqual(glossary.entries, {"hello": "salut", "goodbye": "au revoir"})
|
|
||||||
csv_file_path.unlink() # Remove the temporary file
|
|
||||||
|
|
||||||
def test_import_export_tsv(self) -> None:
|
|
||||||
glossary = Glossary(name="Test", source_lang="en", target_lang="fr", entries={})
|
|
||||||
|
|
||||||
# First export to TSV
|
|
||||||
with tempfile.NamedTemporaryFile('w', delete=False, suffix=glossary_suffix) as tsvfile:
|
|
||||||
tsv_file_path = Path(tsvfile.name)
|
|
||||||
glossary.entries = {"hello": "salut", "goodbye": "au revoir"}
|
|
||||||
glossary.export_tsv(glossary.entries, tsv_file_path)
|
|
||||||
|
|
||||||
# Now import TSV
|
|
||||||
glossary.import_tsv(tsv_file_path)
|
|
||||||
self.assertEqual(glossary.entries, {"hello": "salut", "goodbye": "au revoir"})
|
|
||||||
tsv_file_path.unlink() # Remove the temporary file
|
|
||||||
|
|
||||||
def test_to_file_wrong_suffix(self) -> None:
|
|
||||||
"""
|
|
||||||
Test for exception if suffix is wrong.
|
|
||||||
"""
|
|
||||||
glossary = Glossary(name="Test", source_lang="en", target_lang="fr", entries={"yes": "oui"})
|
|
||||||
with tempfile.NamedTemporaryFile('w', delete=False, suffix='.wrong') as tmp_file:
|
|
||||||
file_path = Path(tmp_file.name)
|
|
||||||
with self.assertRaises(GlossaryError) as err:
|
|
||||||
glossary.to_file(file_path)
|
|
||||||
self.assertEqual(str(err.exception), "File suffix '.wrong' is not supported")
|
|
||||||
|
|
||||||
def test_to_file_auto_suffix(self) -> None:
|
|
||||||
"""
|
|
||||||
Test if suffix is auto-generated if omitted.
|
|
||||||
"""
|
|
||||||
glossary = Glossary(name="Test", source_lang="en", target_lang="fr", entries={"yes": "oui"})
|
|
||||||
with tempfile.NamedTemporaryFile('w', delete=False, suffix='') as tmp_file:
|
|
||||||
file_path = Path(tmp_file.name)
|
|
||||||
glossary.to_file(file_path)
|
|
||||||
assert glossary.file_path is not None
|
|
||||||
self.assertEqual(glossary.file_path.suffix, glossary_suffix)
|
|
||||||
Reference in New Issue
Block a user