itslenny/lenny.py

#!/usr/bin/env python2
# -*- coding: utf-8 -*-

import random
import ConfigParser
import audioop
import contextlib
import glob
import io
import os
import re
import signal
import smtplib
import string
import subprocess
import sys
import syslog
import tempfile
import threading
import urllib2
import wave
from datetime import datetime
from email.mime.text import MIMEText

import linphone
import yaml
from enum import Enum

VOLUME_THRESHOLD = 100

current_dir = os.path.dirname(os.path.realpath(__file__))


def slugify(s):
    """
    Normalizes string, converts to lowercase, removes non-alpha characters,
    and converts spaces to hyphens, wich is url/filename friendly.
    """
    valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
    filename = ''.join(cc for cc in s if cc in valid_chars)
    filename = filename.replace(' ', '_')  # I don't like spaces in filenames.
    return filename


def get_ip_from_logfile(remote_sip_client_tag, log_file):
    # Le log file contiendra, via la mise a jour dyndns custom du routeur, une ligne du type:
    # 84.227.205.102 - jf [23/Jun/2017:17:06:19 +0200] "GET /itslenny/domain=[essai]/ip=[84.227.205.102] HTTP/1.1" 404 169 "-" "Fritz!Box DDNS/1.0.1"
    cmd = 'cat ' + log_file + ' | grep "' + remote_sip_client_tag + '" | tail -n 1'
    reg = "domain=\[" + remote_sip_client_tag + "\]\/ip=\[(.*)\]"
    s = re.findall(reg, subprocess.check_output(["bash", "-c", cmd]))
    if len(s) > 0:
        return s[0]


class RepliesFactory(object):

    def __init__(self, base_path):
        self._base_path = base_path
        self._start_replies = None
        self._random_replies = None
        self._pos = None
        self._latest = None
        self.reset()

    def reset(self):
        self._pos = 0
        self.reset_random()
        self.reset_start()

    def reset_start(self):
        self._start_replies = glob.glob(self._base_path + "/start_sequence/*.opus")
        self._start_replies.sort()

    def reset_random(self):
        self._random_replies = glob.glob(self._base_path + "/random/*.opus")

    def clear_latest(self):
        os.remove(self._latest)

    @staticmethod
    def convert_from_opus(opus_fname):
        wav_file = tempfile.NamedTemporaryFile(delete=False, suffix=".wav").name
        subprocess.call("opusdec --quiet " + opus_fname + " " + wav_file, shell=True)
        return wav_file

    def next(self):
        # On commence par la sequence du début
        if self._pos < len(self._start_replies):
            result = self._start_replies[self._pos]
            self._pos += 1
        else:
            # Puis une fois épuisée, on puisse dans celle random
            if len(self._random_replies) == 0:
                self.reset_random()
            result = random.choice(self._random_replies)
            self._random_replies.remove(result)

        original_file = result
        result = self.convert_from_opus(result)

        self._latest = result
        return result, original_file


class ConversationStatus(Enum):
    READY_TO_TALK = 0
    IMTALKING = 1
    WAITFORANSWER = 2


class Conversation(object):
    def __init__(self):
        self._status = ConversationStatus.READY_TO_TALK

    @property
    def status(self):
        return self._status

    @status.setter
    def status(self, value):
        if value != self._status:
            self._status = value


THREADS_MUST_QUIT = False


def get_wav_duration(fname):
    with contextlib.closing(wave.open(fname, 'r')) as f:
        frames = f.getnframes()
        rate = f.getframerate()
        return frames / float(rate)


def sleep(duration):
    dummy_event = threading.Event()
    dummy_event.wait(timeout=duration)


class SipConnection(object):
    class MailType(Enum):
        Notify_Incoming_Call = 1
        Notify_Incoming_Telemarketer_Call = 2

    def log(self, msg):
        to_show = str(self) + ": " + msg
        print(to_show)
        syslog.syslog(to_show)

    def mail(self, mail_cfg, text):
        try:
            server = smtplib.SMTP(mail_cfg["smtp_host"])

            msg = MIMEText(text)
            msg['Subject'] = text
            msg['From'] = mail_cfg["from"]
            msg['To'] = mail_cfg["to"]

            server.sendmail(mail_cfg["from"], [mail_cfg["to"]], msg.as_string())
            server.quit()
        except smtplib.SMTPException as e:
            self.log("Error sending email " + e.message)

    def say(self, core):
        if self._conversation.status is not ConversationStatus.IMTALKING:
            self._conversation.status = ConversationStatus.IMTALKING

            voice_filename, original_file = self._replies.next()

            duration = get_wav_duration(voice_filename)
            self.log("Saying : " + original_file)

            core.play_file = voice_filename

            sleep(duration)
            core.play_file = ""

            self._replies.clear_latest()

            # On laisse l'autre l'occassion de reparler
            self._conversation.status = ConversationStatus.WAITFORANSWER

    def storage_path(self):
        path = current_dir + "/out/" + str(self) + "/"
        if not os.path.isdir(path):
            os.makedirs(path)
        return path

    def shoul_daccept_first_call(self, number):
        first_call_file = self.storage_path() + "first_calls.txt"
        if not os.path.isfile(first_call_file):
            os.system("touch " + first_call_file)
        if "accept_first_call" in self._config_info and self._config_info["accept_first_call"] and number not in open(
                first_call_file).read():
            os.system("echo " + number + " `date` >> " + first_call_file)
            self.log("Accepting first call of " + number)
            return True
        return False

    def incoming_stream_worker(self, core, call):
        f = open(self._incoming_stream_file, "rb")
        f.seek(0, io.SEEK_END)
        p = f.tell()
        buf = ''

        previous_status = self._conversation.status

        while call.state is not linphone.CallState.End and not self._is_quitting:
            if self._conversation.status is ConversationStatus.IMTALKING:
                f.seek(0, io.SEEK_END)
                p = f.tell()
            else:

                if previous_status != self._conversation.status:
                    f.seek(0, io.SEEK_END)
                    p = f.tell()

                f.seek(p)
                buf += f.read(4096)
                p = f.tell()

                if len(buf) >= 20000:
                    volume = audioop.rms(buf, 2)
                    # print("State : " + str(conversation.status))
                    buf = ''
                    if volume < self._volume_threshold:
                        if self._conversation.status is ConversationStatus.READY_TO_TALK:
                            threading.Thread(target=self.say, args=[core]).start()
                    else:
                        self._conversation.status = ConversationStatus.READY_TO_TALK

            # We must sleep a bit to avoid cpu hog
            sleep(0.05)
            previous_status = self._conversation.status

        self.log("Worker is quitting")

    def registration_state_changed(self, core, call, state, message):
        # Le client se ré-enregistre a de multiple reprise, on
        # s'en tappe un peu d'en être informé.
        if message != self._registration_previous_message:
            self.log("Registration status: " + message)
            self._registration_previous_message = message

    def call_state_changed(self, core, call, state, message):
        self.log("state changed : " + message)

        if state == linphone.CallState.Released:
            # Let's convert wav to mp3
            if call.current_params.record_file is not None and os.path.isfile(call.current_params.record_file):
                self.log("Saving to mp3 : " + call.current_params.record_file)
                subprocess.call('lame --quiet --preset medium %s' % call.current_params.record_file, shell=True)
                os.remove(call.current_params.record_file)

        if state == linphone.CallState.IncomingReceived:
            self.log("Incoming call : {}".format(call.remote_address.username))

            self.mail_if_needed(call.remote_address.username, self.MailType.Notify_Incoming_Call)

            self._replies.reset()

            if self.shoul_daccept_first_call(call.remote_address.username) or self.is_in_blacklists(
                    call.remote_address.username):
                self.log("telemarketer calling : " + call.remote_address.username)
                self.mail_if_needed(call.remote_address.username, self.MailType.Notify_Incoming_Telemarketer_Call)

                call_params = core.create_call_params(call)
                if not os.path.isdir(current_dir + "/out"):
                    os.makedirs(current_dir + "/out")

                a_file = self.storage_path() + "call_from_" + slugify(call.remote_address.username) + \
                         "_" + datetime.now().strftime(
                    '%Y-%m-%d_%Hh%Mmn%Ss') + ".wav"

                self.log("Recording to : " + a_file)

                call_params.record_file = a_file

                # Let ring some time
                sleep(4)

                core.accept_call_with_params(call, call_params)
                call.start_recording()

                sleep(2)

                t = threading.Thread(target=self.incoming_stream_worker, args=[core, call])
                t.start()

                self.say(core)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.log(str(self) + ": cleaning on exit ...")
        os.unlink(self._incoming_stream_file)

    def get_domain_info(self):
        if "domain" in self._config_info:
            return self._config_info["domain"]

        if "domain_dyn_tag" in self._config_info and "domain_dyn_log" in self._config_info:
            return get_ip_from_logfile(self._config_info["domain_dyn_tag"], self._config_info["domain_dyn_log"])

    def set_connection_info(self):
        self._core.clear_proxy_config()

        proxy_cfg = self._core.create_proxy_config()

        domain = self.get_domain_info()

        proxy_cfg.identity_address = self._core.create_address(
            'sip:' + self._config_info["username"] + '@' + domain + ':5060')
        proxy_cfg.server_addr = 'sip:' + domain + ':5060'
        proxy_cfg.register_enabled = True
        self._core.add_proxy_config(proxy_cfg)
        auth_info = self._core.create_auth_info(self._config_info["username"], None, self._config_info["password"],
                                                None, None, domain)
        self._core.add_auth_info(auth_info)

    def start(self):
        self.log("starting ")
        self._core.use_files = True
        self._core.record_file = self._incoming_stream_file

        self.set_connection_info()

        while not self._is_quitting:
            sleep(0.05)
            self._core.iterate()

    def request_quit(self):
        self._is_quitting = True
        self.__exit__(None, None, None)

    def __str__(self):
        return self._config_info["username"] + "@" + self.get_domain_info()

    def __init__(self, config_info):

        callbacks = {
            'call_state_changed': self.call_state_changed,
            'registration_state_changed': self.registration_state_changed,
        }

        self._config_info = config_info

        self._replies = RepliesFactory(current_dir + "/replies/")

        self._core = linphone.Core.new(callbacks, None, None)
        self._is_quitting = False

        self._registration_previous_message = ""

        self._conversation = Conversation()
        self._volume_threshold = VOLUME_THRESHOLD
        self._incoming_stream_file = tempfile.NamedTemporaryFile(delete=False).name
        self._core.iterate()

    def mail_if_needed(self, number, type):
        if "mailer" in self._config_info:
            mail_cfg = self._config_info["mailer"]
            if type == self.MailType.Notify_Incoming_Call:
                if mail_cfg["log_all_call"]:
                    self.mail(mail_cfg, "Appel entrant : " + number)

            if type == self.MailType.Notify_Incoming_Telemarketer_Call:
                self.mail(mail_cfg, "Appel telemarketeur entrant : " + number)

    def is_in_blacklists(self, a_number):
        return self.is_in_local_blacklist(a_number) or self.is_in_directory_ch_blacklist(
            a_number) or self.is_in_ktipp_blacklist(a_number) \
               or self.is_in_shiansw_blacklist(a_number)

    def is_in_local_blacklist(self, a_number):
        black_list = current_dir + "/blacklist.txt"
        if os.path.isfile(black_list):
            res = a_number in open(current_dir + "/blacklist.txt").read()
            if res:
                self.log(a_number + " Found in localblacklist")
            return res

    def is_in_ktipp_blacklist(self, a_number):
        # On peut interroger le site ktipp:
        # https://www.ktipp.ch/service/warnlisten/detail/?warnliste_id=7&ajax=ajax-search-form&keyword=0445510503
        # Si argument keyword pas trouvé, ca donne ca dans la réponse :
        # 0 Einträge

        base_url = "https://www.ktipp.ch/service/warnlisten/detail/?warnliste_id=7&ajax=ajax-search-form&keyword={" \
                   "$number$}"

        the_number = a_number.lstrip("0")
        the_number = the_number.replace("+", "")

        url = base_url.replace("{$number$}", the_number)
        response = ""
        try:
            opener = urllib2.build_opener()
            opener.addheaders = [('User-Agent', 'Mozilla/5.0')]
            response = opener.open(url).read()
        except urllib2.HTTPError as e:
            self.log(" Error during ktipp lookup : url="+url+", exception : "+str(e))

        res = "0 Eintr" not in response
        if res:
            self.log(a_number + " found in ktipp blacklist")
        return res

    def is_in_shiansw_blacklist(self, a_number):
        base_url = "https://ch.shouldianswer.net/telefonnummer/{$number$}"
        url = base_url.replace("{$number$}", a_number)
        response = ""
        try:
            response = urllib2.urlopen(url).read()
        except urllib2.HTTPError:
            pass

        res = '<div class="review_score negative"></div>' in response
        if res:
            self.log("Found in ch.shouldianswer.net blacklist")
        return res

    def is_in_directory_ch_blacklist(self, a_number):
        base_url = "https://tel.local.ch/fr/{$number$}"
        url = base_url.replace("{$number$}", a_number)
        response = ""
        try:
            response = urllib2.urlopen(url).read()
        except urllib2.HTTPError:
            pass

        res = 'https://tel.local.ch/fr/spamnumber/' in response

        if res:
            self.log(a_number + " found in directories.ch blacklist")
        return res


if __name__ == "__main__":

    cfg = ConfigParser.SafeConfigParser()

    cfg_path = current_dir + "/config.yml"

    if len(sys.argv) == 2:
        cfg_path = sys.argv[1]

    connections = []

    for connection_cfg in yaml.load(file(cfg_path)):
        connections.append(SipConnection(connection_cfg))

    for sip_c in connections:
        threading.Thread(target=sip_c.start).start()


    # Ensuring clean quit and ressource releasing
    # when receiving ctrl-c from console or SIGTERM
    # from daemon manager.
    def signal_handler(sig, frame):
        print('External stop request!')
        for conn in connections:
            conn.request_quit()


    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    signal.pause()