#!/usr/bin/python3
# -*- coding: utf-8 -*-
import psycopg2, os, sys, shutil, subprocess, urllib3, time, json, logging, errno
from importlib.machinery import SourceFileLoader
def postJson(address, data):
http = urllib3.PoolManager()
response = http.request('POST', address, headers={'Content-Type': 'application/json'},body=json.dumps(data).encode('utf-8'))
return json.loads(response.data.decode('utf-8'))
def add_examples(address, examplesData):
response = postJson(address, examplesData)
if response['status'] == 'error':
raise Exception(response['message'])
def file_len(fname):
with open(fname) as f:
for i, l in enumerate(f):
pass
return i + 1
def pid_exists(pid):
"""Check whether pid exists in the current process table.
UNIX only.
"""
if pid < 0:
return False
if pid == 0:
# According to "man 2 kill" PID 0 refers to every process
# in the process group of the calling process.
# On certain systems 0 is a valid PID but we have no way
# to know that in a portable fashion.
raise ValueError('invalid PID 0')
try:
os.kill(pid, 0)
except OSError as err:
if err.errno == errno.ESRCH:
# ESRCH == No such process
return False
elif err.errno == errno.EPERM:
# EPERM clearly means there's a process to deny access to
return True
else:
# According to "man 2 kill" possible error values are
# (EINVAL, EPERM, ESRCH)
raise
else:
return True
def updateRequestStatus(connection_string, request_id, new_status):
conn = psycopg2.connect(connection_string)
cur = conn.cursor()
cur.execute("UPDATE request SET status=%s WHERE id=%s", (new_status, request_id))
conn.commit()
cur.close()
conn.close()
def getPendingRequest(connection_string):
conn = psycopg2.connect(connection_string)
cur = conn.cursor()
cur.execute("""select request.id, request.source_file_path, request.target_file_path, request.name, src_lang.id as src_lang_id, src_lang.code as src_code, trg_lang.id as trg_lang_id, trg_lang.code as trg_code, request.status, request.type, request.tm_id from request inner join language as src_lang on src_lang.id = request.source_lang_id inner join language as trg_lang on trg_lang.id = request.target_lang_id where request.status = 0 order by request.created limit 1""")
request = cur.fetchone()
cur.close()
conn.close()
return request
def retrieve_and_append(connection_string, tm_id, src_file_path, trg_file_path):
conn = psycopg2.connect(connection_string)
cur = conn.cursor()
cur.execute("SELECT source_segment, target_segment FROM unit WHERE tm_id=%d" % tm_id)
with open(src_file_path, 'a') as src_file, open(trg_file_path, 'a') as trg_file:
for row in cur:
src_file.write(row[0]+'\n')
trg_file.write(row[1]+'\n')
cur.close()
conn.close()
def clearTm(connection_string, tm_id):
conn = psycopg2.connect(connection_string)
cur = conn.cursor()
cur.execute("DELETE FROM unit WHERE tm_id=%d" %tm_id)
conn.commit()
cur.close()
conn.close()
def getLangsFromTM(connection_string, tm_id):
conn = psycopg2.connect(connection_string)
cur = conn.cursor()
cur.execute("select src_lang.code, trg_lang.code from tm inner join language as src_lang on tm.source_lang_id = src_lang.id inner join language as trg_lang on tm.target_lang_id = trg_lang.id where tm.id = %d" % tm_id)
langs = cur.fetchone()
cur.close()
conn.close()
return langs
# ----------- main -------------------------------------------------------------
BUFFER_SIZE = 500
src_dir = os.path.dirname(os.path.realpath(__file__))
logger = logging.getLogger('request_handler')
hdlr = logging.FileHandler(src_dir+'/request_handler.log')
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.setLevel(logging.INFO)
pidfile_path = src_dir+'/request_handler.pid'
if os.path.isfile(pidfile_path):
with open(pidfile_path) as pidfile:
pid = int(pidfile.readline().strip())
logger.info('Found pid file with pid: %d' % pid)
if pid_exists(pid):
logger.info('The process with pid=%d is actually running, exiting' % pid)
sys.exit(0)
else:
logger.warning('The process with pid=%d is not running anymore. Ignoring the pid file.' % pid)
with open(pidfile_path, 'w') as pidfile:
pidfile.write(str(os.getpid()))
mgiza_path = src_dir+'/../mgiza-aligner'
connection_string = "dbname='concordia_server' user='concordia' host='localhost' port='6543' password='concordia'"
try:
request = getPendingRequest(connection_string)
if request is not None:
request_id, src_file_path, trg_file_path, tm_name, src_lang_id, src_lang_code, trg_lang_id, trg_lang_code, status, tm_type, tm_id = request
if tm_type == 1:
src_lang_code, trg_lang_code = getLangsFromTM(connection_string, tm_id)
logger.info("Working on request: %d. Starting the mgiza aligner." % request_id)
updateRequestStatus(connection_string, request_id, 1)
request_corpus_path = mgiza_path+'/corpora/request_'+str(request_id)
os.makedirs(request_corpus_path)
shutil.copy(src_file_path, request_corpus_path+'/src.txt')
shutil.copy(trg_file_path, request_corpus_path+'/trg.txt')
if tm_type == 1:
# extending existing TM. We must retrieve existing units from the db.
logger.info("This is an extend request. Appending units from the tm %d to the src and trg files."%tm_id)
retrieve_and_append(connection_string, tm_id, request_corpus_path+'/src.txt', request_corpus_path+'/trg.txt')
subprocess.run(["make","SRC_LANG="+src_lang_code, "TRG_LANG="+trg_lang_code, "CORPUS_NAME=request_"+str(request_id)], cwd=mgiza_path, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
logger.info("Mgiza alignment complete, starting adding the TM to Concordia.")
updateRequestStatus(connection_string, request_id, 2)
host = SourceFileLoader("host", os.path.dirname(os.path.realpath(__file__))+'/../tests/host.py').load_module()
address = 'http://'+host.concordia_host
if len(host.concordia_port) > 0:
address += ':'+host.concordia_port
sourceFile = request_corpus_path+'/src_final.txt'
targetFile = request_corpus_path+'/trg_final.txt'
alignmentsFile = request_corpus_path+'/aligned_final.txt'
if (file_len(sourceFile) != file_len(targetFile)):
raise Exception("source and target files are not of the same length!")
if (file_len(alignmentsFile) != 3*file_len(sourceFile)):
raise Exception("alignments file is not exactly 3 times longer than source and target")
if tm_type == 0:
# create new TM
data = {
'operation': 'addTm',
'sourceLangId':src_lang_id,
'targetLangId':trg_lang_id,
'name':tm_name,
'tmLemmatized':True
}
response = postJson(address, data)
tmId = int(response['newTmId'])
logger.info("Added new tm: %d" % tmId)
else:
clearTm(connection_string, tm_id)
tmId = tm_id
data = {
'operation': 'addAlignedLemmatizedSentences',
'tmId':tmId
}
examples = []
with open(sourceFile) as sf, open(targetFile) as tf, open(alignmentsFile) as af:
for sourceLine in sf:
sourceSentence = sourceLine.strip()
targetSentence = tf.readline().strip()
# skip to lines of the alignments file, these are lemmatized and we need the raw sentences from the source and target files.
af.readline()
af.readline()
alignmentString = af.readline().strip()
examples.append([sourceSentence, targetSentence, alignmentString])
if len(examples) >= BUFFER_SIZE:
data['examples'] = examples
add_examples(address, data)
examples = []
if len(examples) > 0:
data['examples'] = examples
add_examples(address, data)
updateRequestStatus(connection_string, request_id, 3)
logger.info("Generating index...")
start = time.time()
data = {
'operation': 'refreshIndex',
'tmId' : tmId
}
response = postJson(address, data)
end = time.time()
logger.info("Index regeneration complete. The operation took %.4f s" % (end - start))
except Exception as e:
logger.error("Error occured: "+str(e))
updateRequestStatus(connection_string, request_id, 5)
else:
updateRequestStatus(connection_string, request_id, 4)
finally:
os.remove(pidfile_path)