itslenny/lenny.py

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

import ConfigParser
import audioop
import contextlib
import glob
import io
import os
import signal
import smtplib
import string
import subprocess
import sys
import syslog
import tempfile
import threading
import urllib2
import wave
from datetime import datetime

import linphone
import yaml
from enum import Enum

VOLUME_THRESHOLD = 100


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


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


current_dir = os.path.dirname(os.path.realpath(__file__))
replies_seq = glob.glob(current_dir + "/replies/sequence/*.wav")
replies_seq.sort()

replies_generic = glob.glob(current_dir + "/replies/generic/*.wav")
replies_generic.sort()

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"])
            server.sendmail(mail_cfg["from"], mail_cfg["to"], text)
            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

            # On joue les repliques en sequence, puis quand
            # on arrive au bout, on en joue une au hasard
            # du groupe 'generic'

            voice_filename = replies_seq[self._replies_pos]
            self._replies_pos = (self._replies_pos + 1) % len(replies_seq)
            if self._replies_pos == 0:
                # On ne rejoue jamais la première réplique "allo"
                self._replies_pos = 1

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

            core.play_file = voice_filename

            sleep(duration)
            core.play_file = ""

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

    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.01)
            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 insane %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_pos = 0

            if 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 = current_dir + "/out/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 start(self):
        self.log("starting ")
        self._core.use_files = True
        self._core.record_file = self._incoming_stream_file

        proxy_cfg = self._core.create_proxy_config()
        proxy_cfg.identity_address = self._core.create_address('sip:' + self._config_info["username"] +
                                                               '@' + self._config_info["domain"] + ':5060')
        proxy_cfg.server_addr = 'sip:' + self._config_info["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, self._config_info["domain"])
        self._core.add_auth_info(auth_info)

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

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

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

    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._core = linphone.Core.new(callbacks, None, None)
        self._is_quitting = False

        self._registration_previous_message = ""

        self._conversation = Conversation()
        self._replies_pos = 0
        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 télémarketeur 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:
            response = urllib2.urlopen(url).read()
        except urllib2.HTTPError:
            pass

        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()