<?php

/***
 * @version $Id: ContactClean.php 80 2010-06-13 23:13:35Z root $
 *
 *
 * The bulk-search functions require the php ssh2 bindings. Found in libssh2-php on ubuntu.
 */
require_once 'Mail/RFC822.php';

class ContactClean {
    /* user vars */
    var $debug = 0;
    var $acc_key = 0;
    var $acc_key_secret = 0;
    var $client_name = 'ContactClean rest class';
    var $temp_dir = '/tmp';
    
    /* internal vars */
    var $rest_ssl = 1;
    var $rest_host = 'www.contactclean.com';
    var $bulk_name = 0;
    
    protected $rest_url = '/ws/rest.php';
    protected $_bulk_user = 'bulk';
    protected $bulk_data_handle = 0;
    
    protected $ssh = 0;
    protected $seen_status = array();
    
    /*
     * This equates to a payload of around 1MB. If you have search list bigger than this then
     * you should use the asynchronous, bulk search method.
     */
    const sync_limit = 25000;
    
    /* max 320MB uncompressed per bulk search */
    const async_limit = 1e7;
    
    function _debug() {
        if ($this->debug) {
            $args = func_get_args();
            error_log(implode('', $args));
        }
    }
    
    /**
    * instance contructor
    *
    * @param string - # https://www.contactclean.com/user/license_key.php?gdet=yes
    * @param string - # https://www.contactclean.com/user/license_key.php?gdet=yes
    * @param string - optional
    * @return object
    */
    function __construct($acc_key, $acc_key_secret, $name=0) {
        if (! preg_match('/[0-9a-f]{32}/', $acc_key)) die("Invalid ContactClean access key $acc_key");
        if (! preg_match('/[0-9a-f]{32}/', $acc_key_secret)) die("Invalid ContactClean secret access key $acc_key_secret");
        
        $this->acc_key = $acc_key;
        $this->acc_key_secret = $acc_key_secret;
        
        if ($name) $this->client_name = $name;
    }
    
    function endpoint() {
        return(($this->rest_ssl ? 'https://' : 'http://') . $this->rest_host . $this->rest_url);
    }

    private function _client_name() {
        return $this->client_name . ' ' . phpversion();
    }
    
    function _set_sig(&$payload) {
        $keys = '';
        $load = $this->acc_key_secret;
        foreach ($payload as $k => $v) {
            if ($keys) $keys .= ':';
            $keys .= $k;
            $load .= $v;
        }
        $payload['sig_keys']    = $keys;
        $payload['sig']            = md5($load);
    }
    
    function _fix_args(&$payload) {
        foreach ($payload as $k => $v) {
            if (is_array($v)) $payload[$k] = implode(',', $v);
        }
    }
    
    private function _rest($method, $payload) {
        static $ch = 0;
        
        $this->_fix_args($payload);
        $payload['method'] = $method;
        $payload['acc_key'] = $this->acc_key;
        $this->_set_sig($payload);
        
        if (function_exists('http_post_data')) {
            $this->_debug('ContactClean - excellent, use built-in function');
            $ret = http_post_data(
                $this->endpoint(),
                $payload,
                array('useragent' => $this->_client_name())
            );
        } else {
            $args = array();
            foreach ($payload as $k => $v) {
                $args[] = "$k=" . urlencode($v);
            }
            $post_string = implode('&', $args);
                
            if (function_exists('curl_init')) {
                $this->_debug('ContactClean - Using CURL');
                if (! $ch) {
                    $ch = curl_init();
                    curl_setopt($ch, CURLOPT_URL, $this->endpoint());
                    curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
                }
                
                curl_setopt($ch, CURLOPT_POSTFIELDS, $post_string);
                curl_setopt($ch, CURLOPT_USERAGENT, $this->_client_name());
                $ret = curl_exec($ch);
            } else {
                $this->_debug('ContactClean - No post function, no curl! Oh dear, have to use plaintext!');
                $context =
                    array('http' =>
                        array('method'    => 'POST',
                            'header'    => "Content-type: application/x-www-form-urlencoded\r\n"
                                . 'User-Agent: ' . $this->_client_name() . "\r\n"
                                . 'Content-length: ' . strlen($post_string),
                            'content'    => $post_string));
                $contextid = stream_context_create($context);
                $sock = fopen(preg_replace('/^https/', 'http', $this->endpoint()), 'r', false, $contextid);
                  if ($sock) {
                    $ret='';
                    while (!feof($sock)) $ret .= fgets($sock, 4096);
                    fclose($sock);
                } else {
                    die("Couldn't open http connection to " . $this->rest_url);
                }
            }
        }
        # echo 'raw - '; print_r($ret); echo "\n";
        $j = is_numeric($ret) # this json stuff seems pretty arbitrary sometimes!
            ? $ret
            : json_decode($ret);
        # echo 'decoded - '; print_r($j); echo "\n";
        if (is_object($j)) {
            $j = (array) $j;
            if (isset($j['error'])) $this->_debug("Method:$method Returned error:" . $j['error']);
        }
        return($j);
    }
    

    function clean_email($addr) {
        static $rfc;
        
        if (! $rfc) $rfc = new Mail_RFC822;
        
        $str = strtolower(trim($addr));
        if ($rfc->validateMailbox($str)) { # call by reference
            if ($str->host == 'localhost') {
                die("no domain found in $addr");
            } else {
                return($str->mailbox . '@' . $str->host);
            }
        } else {
            die("$addr not a valid email address");
        }
    }
    
    
    /* That's the infrastructure done, now for the actual API.
     *
     * Synchronous functions first
     *
     */
    function add_address_pair($current, $previous, $upload_key) {
        return (array) $this->_rest(
            'add_addr_pair',
            array(
                'current'    => $this->clean_email($current),
                'previous'    => $this->clean_email($previous),
                'upload_key' => $upload_key
            )
        );
    }
    
    function addr_search($emails) {
        if (count($emails) > self::sync_limit) {
            return(array('error' => 'Email search list too large. Use bulk_search_start()'));
        }
        
        $md_map = array();
        foreach ($emails as $e) {
            $c = $this->clean_email($e);
            $md_map[md5($c)] = array($c, $e);
        }
        
        #print_r($md_map);
        
        $ret = $this->_rest(
            'addr_search',
            array(
                'emails'    => array_keys($md_map),
                'encoding'    => 'md5',
                'client'    => $this->client_name
            )
        );
        
        if (isset($ret['error'])) {
            $this->_debug($ret);
            return($ret);
        }
        
        $search_result = (array) $ret['search_result'];
        $replace = array();
        foreach ($search_result as $r) {
            $target = $md_map[$r->query];
            $rep = $r->result;
            if ($rep != 'norecord' and $rep != $target[0]) {
                # if the replacement != clean target then return replacement for original target
                $replace[$target[1]] = $rep;
            }
        }
        $ret['search_result'] = $replace;
        $ret['report'] = (array) $ret['report'];
        
        #echo 'addr_search '; print_r($ret);
        return($ret);
    }
        
    function account_quota() {
        return $this->_rest('account_quota', array());
    }

    function account_details() {
        return (array) $this->_rest('key_account', array());
    }
    
    
    /*
     * Asynchronous bulk-search API
     *
     * http://www.contactclean.com/doc/contactclean_bulk_search.pdf
     */
    function bulk_search_start($private, $public, $name=0) {
        if (! file_exists($private)) die("Private ssl key file $private not found");
        if (! file_exists($public)) die("Public ssl key file $public not found");
        if (! $name) $name = 'cc_bulk_' . getmypid();
        
        $this->_private_key = $private;
        $this->_public_key = $public;
        $this->bulk_name = $name;
        $this->bulk_data_handle = gzopen($this->temp_dir . "/${name}_data.gz", 'w');
    }
    
    function bulk_search_add_email($email) {
        
        $md = md5($this->clean_email($email));
        
        gzwrite($this->bulk_data_handle, "$md\n");
        return($md);
    }
    
    function _ssh_open() {
        if (! $this->ssh) {
            $this->ssh = ssh2_connect($this->rest_host, 22);
            if (! $this->ssh) die('Connection to ' . $this->rest_host . ' failed');
            
            if (! ssh2_auth_pubkey_file(
                $this->ssh, $this->_bulk_user, $this->_public_key, $this->_private_key, ''
            )) {
                  die('Public Key Authentication Failed');
            }
        }
    }
    
    function _ssh_close($already_closed=0) {
        if ($this->ssh) {
            if (! $already_closed) ssh2_exec($this->ssh, 'exit');
            $this->ssh = 0;
        }
    }
    
    function _ssh_get($cmd, $blocksize=4096) {
        $this->_ssh_open();
        
        $stream = ssh2_exec($this->ssh, $cmd);
        stream_set_blocking($stream, true);

        $contents = '';
        while (!feof($stream)) {
            $contents .= fread($stream, $blocksize);
        }
        
        $this->_ssh_close(1);
        return($contents);
    }
    
    function bulk_search_execute($notify_email=0) {
        gzclose($this->bulk_data_handle);
        
        $tmp = $this->temp_dir;
        $name = $this->bulk_name;

        $control = array('key' => $this->acc_key);
        if ($notify_email) $control['notify_email'] = $notify_email;
        file_put_contents(
            "$tmp/${name}_control.json",
            json_encode($control)
        );
        file_put_contents("$tmp/${name}_ready", '');
        
        $this->_ssh_open();
        ssh2_scp_send($this->ssh, "$tmp/${name}_data.gz", "${name}_data.gz");
        ssh2_scp_send($this->ssh, "$tmp/${name}_control.json", "${name}_control.json");
        ssh2_scp_send($this->ssh, "$tmp/${name}_ready", "${name}_ready");
        $this->_ssh_close();
        
        $this->seen_status = array('ready' => 0, 'processing' => 0, 'done' => 0);
    }
    
    function bulk_search_status() {
        static $types = array('done', 'processing', 'ready');
        $name = $this->bulk_name;
        # TODO limit checks to once every 10s
        
        $str = $this->_ssh_get('status ' . $name, 1024);
        $stati = explode("\n", $str);
        foreach ($types as $t) {
            if (array_search("${name}_$t", $stati) !== false) {
                return $t;
            }
        }
        return 'unknown';
    }
    
    function bulk_wait_for_status($wait) {
        if (isset($this->seen_status[$wait]) && $this->seen_status[$wait]) return($wait);
        
        while (($stat = $this->bulk_search_status()) !== 'done') {
            $this->_debug(date('h:i:s ') . " bulk_search_result() status:$stat");
            sleep(10);
        }
        $this->seen_status[$stat] = 1;
        if ($stat === 'processing') {
            $this->seen_status['ready'] = 1;
        }
        if ($stat === 'done') {
            $this->seen_status['ready'] = 1;
            $this->seen_status['processing'] = 1;
        }
        return(1);
    }
    
    function bulk_search_report() {
        $this->bulk_wait_for_status('done');
        
        $str = $this->_ssh_get('report ' . $this->bulk_name);
        # $this->_debug('bulk_search_report() str:', $str);
        $report = json_decode($str, true);
        return($report);
    }
    
    function bulk_search_result($callback=0) {
        $result = array();
        $name    = $this->bulk_name;
        $tmp    = $this->temp_dir;
        
        $this->bulk_wait_for_status('done');
        
        $this->_ssh_open();
        ssh2_scp_recv($this->ssh, "${name}_results.gz", "$tmp/${name}_results.gz");
        if (! file_exists("$tmp/${name}_results.gz")) die("Failed to download results file to $tmp/${name}_results.gz");
        
        $gz = gzopen("$tmp/${name}_results.gz", 'r') or die("Failed to open $tmp/${name}_results.gz");
        
        if (! $callback) {
            while (!gzeof($gz)) {
                   $str = trim(gzgets($gz, 128));
                   $line = explode(',', $str, 2);
                   $result[$line[0]] = $line[1];
            }
        } else {
            while (!gzeof($gz)) {
                   call_user_func_array($callback, explode(',', trim(gzgets($gz, 128)), 2));
            }
        }
        gzclose($gz);
        
        $this->_ssh_close();
        
        return $result;
    }
}