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 string
import subprocess
import sys
import syslog
import tempfile
import threading
import urllib2
import wave
from datetime import datetime

import linphone
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

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

SHOULDIANSWER_LOOKUP_URL = "https://ch.shouldianswer.net/telefonnummer/{$number$}"


def is_in_blacklists(a_number):
    return is_in_local_blacklist(a_number) or is_in_ktipp_blacklist(a_number) or is_in_shiansw_blacklist(a_number)


def is_in_local_blacklist(a_number):
    black_list = current_dir + "/blacklist.txt"
    if os.path.isfile(black_list):
        return a_number in open(current_dir + "/blacklist.txt").read()


def is_in_ktipp_blacklist(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

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

    url = KTIP_LOOKUP_URL.replace("{$number$}", the_number)
    response = ""
    try:
        response = urllib2.urlopen(url).read()
    except urllib2.HTTPError:
        pass

    return "0 Eintr" not in response


def is_in_shiansw_blacklist(a_number):
    url = SHOULDIANSWER_LOOKUP_URL.replace("{$number$}", a_number)
    response = ""
    try:
        response = urllib2.urlopen(url).read()
    except urllib2.HTTPError:
        pass

    return '<div class="review_score negative"></div>' in response


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):
    def log(self, msg):
        to_show = str(self) + ": " + msg
        print(to_show)
        syslog.syslog(to_show)

    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._replies_pos = 0

            if is_in_blacklists(call.remote_address.username):
                self.log("telemarketer calling : " + call.remote_address.username)

                call_params = core.create_call_params(call)
                os.makedirs(current_dir + "/out", exist_ok=True)
                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._username + '@' + self._domain + ':5060')
        proxy_cfg.server_addr = 'sip:' + self._domain + ':5060'
        proxy_cfg.register_enabled = True
        self._core.add_proxy_config(proxy_cfg)
        auth_info = self._core.create_auth_info(self._username, None, self._password, None, None, self._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._username + "@" + self._domain

    def __init__(self, domain, username, password):

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

        self._core = linphone.Core.new(callbacks, None, None)
        self._domain = domain
        self._username = username
        self._password = password
        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()


if __name__ == "__main__":

    cfg = ConfigParser.SafeConfigParser()

    cfg_path = current_dir + "/config.ini"

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

    connections = []

    cfg.read(cfg_path)

    for c in cfg.sections():
        connections.append(SipConnection(cfg.get(c, "domain"), cfg.get(c, "username"), cfg.get(c, "password")))

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