Commit 90b67e91 authored by Jakob Lerch's avatar Jakob Lerch
Browse files

add logging

parent f7867512
......@@ -7,19 +7,22 @@ from difflib import SequenceMatcher
from itertools import takewhile
from math import ceil
from pathlib import Path
from typing import Tuple
import requests
from typing import List, Tuple
import utils
from pad import Etherpad, HedgedocNote, Pad
from translation import DeepL, LibreTranslate, Translation
# TODO: implement initial test of config file (if values are ok and stuff), provide fallback values, error msgs
# TODO: scheint nur bei fem-instanz zu funktionieren. bei libretranslate.de kackt das ab
# TODO: use .get of configparser
# TODO: document config properly
# TODO: proper typing
def main():
# setup logging
log = utils.MyLogger()
# read config files
api_keys = ConfigParser()
config = ConfigParser()
......@@ -28,9 +31,7 @@ def main():
api_keys.read(Path(__file__).parent.parent / "api_keys.conf")
config.read(Path(__file__).parent.parent / "config.conf")
except Error as e:
print(e)
print("exiting...")
sys.exit()
log.exit_with_error(e)
# verify that neccessary parameters are set
try:
......@@ -46,21 +47,20 @@ def main():
config.get("destination", "url")
config.get("destination", "language")
except Error as e:
print(e)
print("exiting...")
sys.exit()
log.exit_with_error(e)
# read main
header_size = config.getint("main", "header_size", fallback=0)
if header_size < 0:
raise ValueError(
log.exit_with_error(
"config.conf: 'header_size' in [main] is < 0. It should be >= 0.")
# create src pad
src_pad: Pad
if config["source"]["url"] == "":
raise ValueError("config.conf: 'url' in [source] must not be empty.")
log.exit_with_error(
"config.conf: 'url' in [source] must not be empty.")
if config["source"]["kind"] == "hedgedoc":
src_pad = HedgedocNote(config["source"]["url"])
......@@ -68,7 +68,7 @@ def main():
src_pad = Etherpad(
config["source"]["url"], api_keys["default"]["etherpad_key"])
else:
raise ValueError(
log.exit_with_error(
"config.conf: 'kind = " + config["source"]["kind"] + "' in [source] is not valid. It should either be 'hedgedoc' or 'etherpad'.")
# create dst pad
......@@ -76,14 +76,14 @@ def main():
dst_pad: Pad
if config["destination"]["url"] == "":
raise ValueError(
log.exit_with_error(
"config.conf: 'url' in [destination] must not be empty.")
if config["destination"]["kind"] == "etherpad":
dst_pad = Etherpad(
config["destination"]["url"], api_keys["default"]["etherpad_key"])
else:
raise ValueError(
log.exit_with_error(
"config.conf: 'kind = " + config["destination"]["kind"] + "' in [source] is not valid. It should be 'etherpad'.")
# write termination message on exit
......@@ -97,7 +97,7 @@ def main():
t: Translation
if config["translation"]["service_url"] == "":
raise ValueError(
log.exit_with_error(
"config.conf: 'service_url' in [translation] must not be empty.")
# TODO: support DeepL
......@@ -105,54 +105,57 @@ def main():
t = LibreTranslate(
config["translation"]["service_url"], api_keys["default"]["libretranslate_key"])
else:
raise ValueError(
log.exit_with_error(
"config.conf: 'service = " + config["translation"]["service"] + "' in [translation] is not valid. It should be 'libretranslate'")
# ..check if src_lang and dst_lang are supported
supported_languages = t.get_supported_languages()
if src_lang not in supported_languages:
raise ValueError("config.conf: 'language = " +
config["destination"]["kind"] +
"' in [source] is not valid. Supported languages are: " +
', '.join(supported_languages))
log.exit_with_error(
"config.conf: 'language = " +
config["destination"]["kind"] +
"' in [source] is not valid. Supported languages are: " +
', '.join(supported_languages))
if dst_lang not in supported_languages:
raise ValueError("config.conf: 'language = " +
config["destination"]["kind"] +
"' in [source] is not valid. Supported languages are: " +
', '.join(supported_languages))
log.exit_with_error(
"config.conf: 'language = " +
config["destination"]["kind"] +
"' in [source] is not valid. Supported languages are: " +
', '.join(supported_languages))
# initial translation
def translate_with_leading_whitespace(lines: List[str]) -> List[str]:
""" translate list of strings with trailing whitespace lines:
* strip leading lines consisting of whitespace,
* add them when writing to dst_pad
This has to be done, because libretranslate ignores leading whitespace,
leading to incorrect order of the translated lines.
"""
leading_space_lines = list(
takewhile(lambda line: line.isspace() or not line, lines))
text_to_translate = '\n'.join(
lines[len(leading_space_lines):])
translated_lines = t.translate(text_to_translate, src_lang,
dst_lang).splitlines()
return leading_space_lines + translated_lines
dst_pad.write("initializing...")
initial_text = src_pad.read().splitlines()
header = initial_text[0:header_size]
body = initial_text[header_size:]
# ..only include line break after header if header size > 0
joined_header = '\n'.join(header) + '\n' if len(header) > 0 else ""
# ..strip leading lines consisting of whitespace
# ..add them when writing to dst_pad
# ..this has to be done, because libretranslate ignores leading whitespace,
# ..leading to incorrect order of the translated lines
leading_space_lines = list(
takewhile(lambda line: line.isspace() or not line, body))
# ..only include line break after lines if size > 0
joined_leading_space_lines = '\n'.join(
leading_space_lines) + '\n' if len(leading_space_lines) > 0 else ""
body = body[len(leading_space_lines):]
joined_translated_body = t.translate(
'\n'.join(body), src_lang, dst_lang)
translated_body = translate_with_leading_whitespace(body)
dst_pad.write(joined_header + joined_leading_space_lines +
joined_translated_body)
dst_pad.write('\n'.join(header + translated_body))
# ..further translation
s = SequenceMatcher()
translated_body = joined_translated_body.splitlines()
joined_header = '\n'.join(header) + '\n' if len(header) > 0 else ""
while True:
# it is assumed, that the translation service definitely translates the protocol line by line
......@@ -164,19 +167,18 @@ def main():
body = src_pad.read().splitlines()[header_size:]
# again, handle leading space lines
leading_space_lines = list(
takewhile(lambda line: line.isspace() or not line, body))
joined_leading_space_lines = '\n'.join(
leading_space_lines) + '\n' if len(leading_space_lines) > 0 else ""
# leading_space_lines = list(
# takewhile(lambda line: line.isspace() or not line, body))
# joined_leading_space_lines = '\n'.join(
# leading_space_lines) + '\n' if len(leading_space_lines) > 0 else ""
body = body[len(leading_space_lines):]
# body = body[len(leading_space_lines):]
# sequence matcher get_opcodes
s.set_seqs(old_body, body)
opcodes = s.get_opcodes()
# translate the changed text
# ..define locale function
def translate_single_opcode(opcode: Tuple[str, int, int, int, int]):
"""
compute the text in the lines that have been changed and translate it,
......@@ -185,10 +187,10 @@ def main():
if opcode[0] == "delete":
return None
else:
text_to_translate = '\n'.join(body[opcode[3]:opcode[4]])
return t.translate(text_to_translate, src_lang, dst_lang).splitlines()
lines = body[opcode[3]:opcode[4]]
return translate_with_leading_whitespace(lines)
# ..create a generator
def translate_opcodes():
for val in opcodes:
# ..add all elements of 'opcodes' to it except those with the 'equal'-tag
......@@ -218,8 +220,7 @@ def main():
assert False, f"The value '{val[0]}' should not be possible"
# write
dst_pad.write(joined_header + joined_leading_space_lines +
'\n'.join(translated_body))
dst_pad.write(joined_header + '\n'.join(translated_body))
if __name__ == "__main__":
......
......@@ -3,6 +3,8 @@ from urllib.parse import quote
import requests
import utils
class Pad(ABC):
......@@ -33,13 +35,15 @@ class Etherpad(Pad):
self.main_url, self.pad_id = url_parted
self.api_key = api_key
self.log = utils.MyLogger()
def read(self) -> str:
r = requests.get(self.main_url + "/api/1/getText?apikey=" + self.api_key +
"&padID=" + self.pad_id)
r_json = r.json()
if not r_json["code"] == 0:
raise Exception(r_json["message"])
self.log.exit_with_error(r_json["message"])
return r_json["data"]["text"]
......@@ -51,7 +55,7 @@ class Etherpad(Pad):
r_json = r.json()
if not r_json["code"] == 0:
raise Exception(r_json["message"])
self.log.exit_with_error(r_json["message"])
def append(self, string: str) -> None:
string = quote(string)
......@@ -60,7 +64,7 @@ class Etherpad(Pad):
r_json = r.json()
if not r_json["code"] == 0:
raise Exception(r_json["message"])
self.log.exit_with_error(r_json["message"])
class HedgedocNote(Pad):
......
import logging
import sys
from rich.logging import RichHandler
class MyLogger:
def __init__(self):
FORMAT = "%(message)s"
logging.basicConfig(
level="ERROR", format=FORMAT, datefmt="[%X]", handlers=[RichHandler()]
)
self.log = logging.getLogger("rich")
def exit_with_error(self, err_msg):
self.log.error(err_msg)
sys.exit(1)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment