Perbarui status

CHANGES.txt

+1.0.1 2023-07-05
+--------------
+- Perbarui status
+
 1.0 2023-06-30
 --------------
 - Tidak perlu lagi install paket Debian im-gw karena sudah dipindahkan ke sini.

imgw/scripts/agent_job.py

 import sys
 import os
 import json
+from time import sleep
 from logging import getLogger
 from configparser import ConfigParser
 from textwrap import wrap
@@ -19,16 +20,88 @@ from ..models import (
 from .tools import (
     make_pid_file,
     clean_log,
+    file2dict,
+    file_time,
     )
 from .logger import setup_logging
 
 
-my_registry = dict()
+registry = dict()
 
 
+##############################
+# Baca Status Agent (Mailer) #
+##############################
+
+def update_agent(d):
+    db_session = registry['db_session']
+    q = db_session.query(Agent).filter_by(id=d['id'])
+    row = q.first()
+    if not row:
+        row = Agent(id=d['id'], jalur=6)
+    row.status = d['status']
+    row.startup = d['tgl']
+    if d['status'] == 0:
+        row.lasterr = None
+    else:
+        row.lasterr = d['jawaban']
+    with transaction.manager:
+        db_session.add(row)
+
+
+def update_job(d):
+    db_session = registry['db_session']
+    q = db_session.query(Selesai).filter_by(id=d['id'])
+    row = q.first()
+    agent = None
+    if row:
+        q = db_session.query(Agent).filter_by(id=row.pengirim)
+        agent = q.first()
+        row.status = d['status']
+        row.jawaban = d['jawaban']
+        row.tgl_operator = d['tgl']
+        if agent:
+            agent.lastjob = d['tgl']
+    with transaction.manager:
+        db_session.add(row)
+        if agent:
+            db_session.add(agent)
+
+
+def read_status_file():
+    log = getLogger('read_status_file')
+    dir_name = registry['conf']['result_dir']
+    files = os.listdir(dir_name)
+    files.sort()
+    for filename in files:
+        fullpath = os.path.join(dir_name, filename)
+        try:
+            d = file2dict(fullpath)
+            log.info(f'Read {fullpath}: {d}')
+        except Exception:
+            pass
+        if 'id' not in d:
+            prefix = os.path.splitext(filename)[0]
+            try:
+                d['id'] = int(prefix)
+            except ValueError:
+                pass
+        if 'id' in d:
+            d['tgl'] = file_time(fullpath)
+            if filename == 'status.json':
+                update_agent(d)
+            else:
+                update_job(d)
+        os.remove(fullpath)
+
+
+##################################
+# Kirim Job untuk Agent (Mailer) #
+##################################
+
 def write_job_file(s, mail=None):
-    conf = my_registry['conf']
-    db_session = my_registry['db_session']
+    conf = registry['conf']
+    db_session = registry['db_session']
     log = getLogger()
     job_file = f'{s.id}.json'
     job_file = os.path.join(conf['job_dir'], job_file)
@@ -46,7 +119,6 @@ def write_job_file(s, mail=None):
     json_d = json.dumps(d)
     with open(job_file, 'w') as f:
         f.write(json_d)
-    return job_file
 
 
 def create_mail_row(a):
@@ -57,18 +129,8 @@ def create_mail_row(a):
     return Mail(id=a.id, name=name, subject=subject)
 
 
-def main(argv=sys.argv[1:]):
-    conf_file = argv[0]
-    conf = ConfigParser()
-    conf.read(conf_file)
-    my_registry['conf'] = cf = dict(conf.items('main'))
-    if not make_pid_file(cf['pid_file']):
-        sys.exit(1)
-    setup_logging(conf_file)
-    engine = engine_from_config(cf, 'db_')
-    factory = sessionmaker(bind=engine)
-    my_registry['db_session'] = db_session = factory()
-    register(db_session)
+def read_queue():
+    db_session = registry['db_session']
     q = db_session.query(Antrian).filter_by(kirim=True)
     q_mail = db_session.query(Mail)
     while True:
@@ -102,5 +164,23 @@ def main(argv=sys.argv[1:]):
                 db_session.add(mail)
             db_session.flush()
             db_session.expunge_all()
-        json_file = write_job_file(s, mail)
-    os.remove(cf['pid_file'])
+        if s.status > 0:
+            write_job_file(s, mail)
+
+
+def main(argv=sys.argv[1:]):
+    conf_file = argv[0]
+    conf = ConfigParser()
+    conf.read(conf_file)
+    registry['conf'] = cf = dict(conf.items('main'))
+    if not make_pid_file(cf['pid_file']):
+        sys.exit(1)
+    setup_logging(conf_file)
+    engine = engine_from_config(cf, 'db_')
+    factory = sessionmaker(bind=engine)
+    registry['db_session'] = db_session = factory()
+    register(db_session)
+    while True:
+        read_status_file()
+        read_queue()
+        sleep(1)

imgw/scripts/common.py

-import os
-
-
-def is_live(pid):
-    try:
-        os.kill(pid, 0)
-    except OSError:
-        return
-    return True
-
-
-def read_pid_file(filename):
-    try:
-        f = open(filename)
-        s = f.read()
-        f.close()
-        s = s.split()
-        s = s[0]
-        return int(s)
-    except IOError:
-        return
-    except ValueError:
-        return
-    except IndexError:
-        return
-
-
-def write_pid_file(filename):
-    pid = os.getpid()
-    f = open(filename, 'w')
-    f.write(str(pid))
-    f.close()
-    return pid
-
-
-def make_pid_file(filename):
-    pid = read_pid_file(filename)
-    if pid and is_live(pid):
-        print(f'PID saya {pid} masih aktif.')
-        return
-    return write_pid_file(filename)

imgw/scripts/data/status_agent.csv

+id,ket
+0,Siap
+-1,Masalah network / sinyal
+-2,Login gagal
+-3,User non aktif / account locked
+-4,User tidak ada
+-5,Kirim pesan gagal / pulsa habis
+-9,Offline

imgw/scripts/init_db.py

@@ -11,6 +11,7 @@ from ..models import (
     Base,
     Jalur,
     Status,
+    StatusAgent,
     )
 
 
@@ -63,3 +64,4 @@ def main(argv=sys.argv[1:]):
     with transaction.manager:
         append_csv(Jalur, 'jalur.csv', ['id'])
         append_csv(Status, 'status.csv', ['id'])
+        append_csv(StatusAgent, 'status_agent.csv', ['id'])

imgw/scripts/mailer.py

 import os
 import sys
 import json
+import socket
+from time import sleep
 from logging import getLogger
 from argparse import ArgumentParser
 from configparser import ConfigParser
@@ -11,84 +13,65 @@ from .tools import (
     make_pid_file,
     clean_log,
     create_doc,
+    file2dict,
     )
 from .logger import setup_logging
 
 
 registry = dict()
+socket.setdefaulttimeout(30)
 
 
-def send(doc=None):
-    conf = registry['conf']
-    log = getLogger('send')
-    try:
-        with smtplib.SMTP(conf['host'], conf['port']) as server:
-            if conf['tls']:
-                server.starttls()
-            err = None
-            log.debug(f'Login sebagai {conf["username"]}')
-            try:
-                server.login(conf['username'], conf['password'])
-            except OSError as err:
-                return dict(status=-1, jawaban=str(err).strip())
-            except smtplib.SMTPAuthenticationError as err:
-                return dict(status=-2, jawaban=str(err).strip())
-            if doc:
-                log.info(f'KIRIM {clean_log(doc._data)}')
-                server.sendmail(
-                    conf['username'], doc._data['penerima'], doc.as_string())
-    except ConnectionRefusedError as err:
-        return dict(status=-1, jawaban=str(err).strip())
-    return dict(status=0, jawaban='OK')
-
-
-def job_file_to_doc(filename):
-    log = getLogger('job_file_to_doc')
-    with open(filename) as f:
-        s = f.read()
-    os.remove(filename)
-    try:
-        d = json.loads(s)
-    except Exception:
-        log.error(f'eval({s}): {exception_message()}')
-        return
+def dict2doc(d):
     cf = registry['conf']
     files = []
     for fname, content in d.get('files', []):
         data = base64.b64decode(content)
         files.append([fname, data])
-    doc = create_doc(
+    return create_doc(
             cf['username'], d['penerima'], d.get('subject'), d['pesan'],
             cf['name'], d.get('name'), files)
-    # Hack
-    doc._data = d
-    return doc
 
 
 def save_result(d, filename):
+    log = getLogger('save_result')
     result_dir = registry['conf']['result_dir']
     fullpath = os.path.join(result_dir, filename)
     with open(fullpath, 'w') as f:
         f.write(json.dumps(d))
+    log.info(f'Save {fullpath}: {d}')
 
 
-def do_job():
-    job_dir = registry['conf']['job_dir']
+def save_status(status=0, jawaban='OK'):
+    conf = registry['conf']
+    d = dict(id=conf['username'], status=status, jawaban=jawaban)
+    save_result(d, 'status.json')
+
+
+def send(server):
+    conf = registry['conf']
+    job_dir = conf['job_dir']
     job_files = os.listdir(job_dir)
     if not job_files:
         return
+    log = getLogger('send')
     job_files.sort()
     job_file = os.path.join(job_dir, job_files[0])
-    doc = job_file_to_doc(job_file)
-    if not doc:
-        return
-    log = getLogger('do_job')
     result_file = os.path.split(job_file)[-1]
-    result = send(doc)
-    if result['status'] == 0:
-        log.info(result)
-    else:
-        log.error(result)
+    d = file2dict(job_file)
+    os.remove(job_file)
+    result = 'id' in d and dict(id=d['id']) or dict()
+    try:
+        doc = dict2doc(d)
+        log.info(f'Send {clean_log(d)}')
+        server.sendmail(
+            conf['username'], d['penerima'], doc.as_string())
+        result['status'] = 0
+        result['jawaban'] = 'OK'
+    except Exception as e:
+        result['status'] = -3
+        result['jawaban'] = exception_message()
+        log.error(f'Result {result}')
     save_result(result, result_file)
 
 
@@ -106,15 +89,30 @@ def main(argv=sys.argv[1:]):
     if not make_pid_file(cf['pid_file']):
         return
     setup_logging(option.conf)
-    log = getLogger()
+    log = getLogger('main')
     cf['tls'] = cf['tls'] == 'true'
     cf['port'] = int(cf['port'])
-    # result = send()
-    # save_result(result, 'status.json')
+    log.debug(f'Connect to {cf["host"]}:{cf["port"]}')
     try:
-        do_job()
-    except Exception as err:
-        d = dict(status=-9, jawaban=exception_message())
-        save_result(d, 'status.json')
-        log.error(d['jawaban'])
+        with smtplib.SMTP(cf['host'], cf['port']) as server:
+            if cf['tls']:
+                server.starttls()
+            log.debug(f'Login sebagai {cf["username"]}')
+            server.login(cf['username'], cf['password'])
+            save_status()
+            while True:
+                send(server)
+                sleep(1)
+    except ConnectionRefusedError as err:
+        msg = str(err)
+    except OSError as err:
+        msg = str(err)
+    except smtplib.SMTPAuthenticationError as err:
+        msg = str(err)
+    except KeyboardInterrupt:
+        msg = 'Keyboard Interrupt'
+    except Exception:
+        msg = exception_message()
+    log.error(msg)
+    save_status(-2, msg)
     os.remove(cf['pid_file'])

imgw/scripts/tools.py

@@ -4,6 +4,10 @@ from email.mime.text import MIMEText
 from email.mime.base import MIMEBase
 from email.mime.multipart import MIMEMultipart
 import mimetypes
+import json
+from logging import getLogger
+from time import localtime
+from opensipkd.waktu import create_datetime
 
 
 ##########
@@ -103,3 +107,21 @@ def clean_log(p):
         files.append(fname)
     r['files'] = files
     return r
+
+
+def file2dict(filename):
+    log = getLogger('file2dict')
+    with open(filename) as f:
+        s = f.read()
+    try:
+        return json.loads(s)
+    except Exception as e:
+        log.error(f'eval({s}): {exception_message()}')
+        raise e
+
+
+def file_time(filename):
+    t = os.stat(filename).st_mtime
+    t = localtime(t)
+    return create_datetime(
+            t.tm_year, t.tm_mon, t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec)