Pipedrive activities support

plugins/MauticCrmBundle/Api/PipedriveApi.php

@@ -11,13 +11,18 @@ class PipedriveApi extends CrmApi
     const ORGANIZATIONS_API_ENDPOINT = 'organizations';
     const PERSONS_API_ENDPOINT       = 'persons';
     const USERS_API_ENDPOINT         = 'users';
+    const ACTIVITY_TYPES             = 'activityTypes';
+    const ACTIVITIES                 = 'activities';
     /**
      * @var TransportInterface
      */
     private $transport;
 
     private $apiFields = [];
 
+    /** @var null|array */
+    private $activities = null;
+
     /**
      * PipedriveApi constructor.
      *
@@ -168,8 +173,7 @@ public function getDataByEndpoint(array $query, $endpoint)
             'query' => array_merge($this->getAuthQuery(), $query),
         ];
 
-        $url = sprintf('%s/%s', $this->integration->getApiUrl(), $endpoint);
-
+        $url      = sprintf('%s/%s', $this->integration->getApiUrl(), $endpoint);
         $response = $this->transport->get($url, $params);
 
         return json_decode($response->getBody(), true);
@@ -230,6 +234,97 @@ private function getRequestParameters(array $data = [])
         ];
     }
 
+    /**
+     * @param array $data
+     *
+     * @return array
+     */
+    public function addActivity($data)
+    {
+        $params     = $this->getRequestParameters($data);
+        $url        = sprintf('%s/%s', $this->integration->getApiUrl(), self::ACTIVITIES);
+        $response   = $this->transport->post($url, $params);
+
+        return $data = $this->getResponseData($response);
+    }
+
+    /**
+     * @param array $event
+     *
+     * @return array|mixed
+     */
+    public function getActivityType($event)
+    {
+        $activityName = $event['eventType'];
+        $activities   = $this->getActivityTypes();
+        // return If already exists
+        if (isset($activities[$activityName])) {
+            return $activities[$activityName];
+        }
+
+        return false;
+    }
+
+    /**
+     * Create activity log on Pipedrive.
+     *
+     * @param array $event
+     *
+     * @return array
+     */
+    public function createActivityType($event)
+    {
+        $icon = 'pricetag';
+
+        switch ($event['event']) {
+            case 'email.sent':
+            case 'email.read':
+            case 'email.replied':
+            case 'email.failed':
+                $icon = 'email';
+                break;
+            case 'campaign.event':
+            case 'campaign.event.scheduled':
+                $icon = 'task';
+                break;
+            case 'lead.source.identified':
+            case 'lead.source.created':
+            case 'segment_membership':
+            case 'lead.donotcontact':
+                $icon = 'addressbook';
+                break;
+        }
+
+        $activityName     = $event['eventType'];
+        $data             = [];
+        $data['name']     = $activityName;
+        $data['icon_key'] = $icon;
+        $data['color']    = '4e5e9e';
+
+        $params   = $this->getRequestParameters($data);
+        $url      = sprintf('%s/%s', $this->integration->getApiUrl(), self::ACTIVITY_TYPES);
+        $response = $this->transport->post($url, $params);
+
+        $data = $this->getResponseData($response);
+
+        return $this->activities[$activityName] = $data;
+    }
+
+    /**
+     * @return array
+     */
+    private function getActivityTypes()
+    {
+        if (null !== $this->activities) {
+            return $this->activities;
+        }
+
+        $activities = $this->getDataByEndpoint([], 'activityTypes');
+
+        //return data by name
+        return $this->activities = array_combine(array_column($activities['data'], 'name'), $activities['data']);
+    }
+
     /**
      * @return array
      */

plugins/MauticCrmBundle/Command/PushActivitiesToPipedriveCommand.php

@@ -0,0 +1,138 @@
+<?php
+
+namespace MauticPlugin\MauticCrmBundle\Command;
+
+use Mautic\CoreBundle\Helper\ProgressBarHelper;
+use Mautic\PluginBundle\Entity\IntegrationEntity;
+use MauticPlugin\MauticCrmBundle\Integration\Pipedrive\Export\ActivitiesPipedriveExport;
+use MauticPlugin\MauticCrmBundle\Integration\Pipedrive\Export\LeadExport;
+use MauticPlugin\MauticCrmBundle\Integration\PipedriveIntegration;
+use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class PushActivitiesToPipedriveCommand extends ContainerAwareCommand
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function configure()
+    {
+        $this->setName('mautic:integration:pipedrive:pushactivities')
+            ->setDescription('Pushes the data from Mautic to Pipedrive')
+            ->addOption(
+                'time-interval',
+                't',
+                InputOption::VALUE_OPTIONAL,
+                'Send time interval to check updates, it should be a correct php formatted time interval in the past eg:(15 minutes)',
+                '15 minutes'
+            );
+
+        parent::configure();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $integrationHelper = $this->getContainer()->get('mautic.helper.integration');
+        /** @var PipedriveIntegration $integrationObject */
+        $integrationObject = $integrationHelper->getIntegrationObject(PipedriveIntegration::INTEGRATION_NAME);
+        $em                = $this->getContainer()->get('doctrine')->getManager();
+
+        if (!$integrationObject->getIntegrationSettings()->getIsPublished()) {
+            $output->writeln('Pipedrive integration id disabled.');
+
+            return;
+        }
+
+        $output->writeln('<comment>Push activities</comment>');
+
+        /** @var ActivitiesPipedriveExport $activitiesExport */
+        $activitiesExport = $this->getContainer()->get('mautic_integration.pipedrive.export.activities');
+        $activitiesExport->setIntegration($integrationObject);
+
+        $pushed = 0;
+        $start  = 0;
+        $limit  = 50;
+
+        // now() - time-interval
+        $timeInterval     = str_replace('--', '-', '-'.$input->getOption('time-interval'));
+        $lastPossibleSync = (new \DateTime())->setTimestamp(strtotime($timeInterval));
+
+        $progress = ProgressBarHelper::init($output, $limit);
+
+        while (true) {
+            $integrationEntities = $this->getIntegrationEntities($integrationObject, $start, $limit);
+
+            if (!$integrationEntities) {
+                break;
+            }
+
+            /** @var IntegrationEntity $integrationEntity */
+            foreach ($integrationEntities as $integrationEntity) {
+                $internal         = $integrationEntity->getInternal();
+                $lastActivitySync = null;
+                if (isset($internal['last_activity_sync'])) {
+                    $lastActivitySync = (new \DateTime())->setTimestamp($internal['last_activity_sync']);
+                    // stop If last activity sync is over timeout
+                    if ($lastActivitySync > $lastPossibleSync) {
+                        continue;
+                    }
+                }
+                if ($activitiesExport->createActivities($integrationEntity, $lastActivitySync) !== false) {
+                    ++$pushed;
+                    if ($pushed % $limit == 0) {
+                        $progress->setProgress($pushed);
+                    }
+                }
+            }
+            $start = $start + $limit;
+            $em->clear();
+        }
+
+        $progress->finish();
+        $output->writeln('');
+        $output->writeln('Pushed activities to '.$pushed.' contacts');
+        $output->writeln('Execution time: '.number_format(microtime(true) - $_SERVER['REQUEST_TIME_FLOAT'], 3));
+    }
+
+    /**
+     * @param PipedriveIntegration $integrationObject
+     * @param int                  $start
+     * @param int                  $limit
+     *
+     * @return array|null
+     */
+    private function getIntegrationEntities($integrationObject, $start, $limit)
+    {
+        return $integrationObject->getIntegrationEntityRepository()->getEntities(
+            [
+                'filter' => [
+                    'force' => [
+                        [
+                            'column' => 'e.integration',
+                            'expr'   => 'eq',
+                            'value'  => $integrationObject->getName(),
+                        ],
+                        [
+                            'column' => 'e.integrationEntity',
+                            'expr'   => 'eq',
+                            'value'  => LeadExport::PERSON_ENTITY_TYPE,
+                        ],
+                        [
+                            'column' => 'e.internalEntity',
+                            'expr'   => 'eq',
+                            'value'  => LeadExport::LEAD_ENTITY_TYPE,
+                        ],
+                    ],
+                ],
+                'start'            => $start,
+                'limit'            => $limit,
+                'ignore_paginator' => true,
+            ]
+        );
+    }
+}

plugins/MauticCrmBundle/Command/PushDataToPipedriveCommand.php

@@ -2,6 +2,8 @@
 
 namespace MauticPlugin\MauticCrmBundle\Command;
 
+use Doctrine\ORM\EntityManager;
+use Mautic\CoreBundle\Helper\ProgressBarHelper;
 use Mautic\LeadBundle\Entity\Company;
 use Mautic\LeadBundle\Entity\Lead;
 use MauticPlugin\MauticCrmBundle\Integration\PipedriveIntegration;
@@ -41,6 +43,7 @@ protected function configure()
     protected function execute(InputInterface $input, OutputInterface $output)
     {
         $integrationHelper = $this->getContainer()->get('mautic.helper.integration');
+        /** @var PipedriveIntegration $integrationObject */
         $integrationObject = $integrationHelper->getIntegrationObject(PipedriveIntegration::INTEGRATION_NAME);
         $this->io          = new SymfonyStyle($input, $output);
         $em                = $this->getContainer()->get('doctrine')->getManager();
@@ -77,18 +80,75 @@ protected function execute(InputInterface $input, OutputInterface $output)
             $this->io->text('Pushed '.$pushed);
         }
 
-        $leads = $em->getRepository(Lead::class)->findAll();
         $this->io->title('Pushing Leads');
+
         $leadExport = $this->getContainer()->get('mautic_integration.pipedrive.export.lead');
         $leadExport->setIntegration($integrationObject);
-        $pushed = 0;
-        foreach ($leads as $lead) {
-            if ($leadExport->create($lead)) {
-                ++$pushed;
+
+        $pushed   = 0;
+        $start    = 0;
+        $limit    = 50;
+        $progress = ProgressBarHelper::init($output, $limit);
+        while (true) {
+            $leads = $this->getLeads($em, $start, $limit);
+
+            if (!$leads) {
+                break;
+            }
+
+            foreach ($leads as $lead) {
+                if ($leadExport->create($lead)) {
+                    ++$pushed;
+                    if ($pushed % $limit == 0) {
+                        $progress->setProgress($pushed);
+                    }
+                }
             }
+            $start = $start + $limit;
+            $em->clear();
         }
-        $this->io->text('Pushed '.$pushed);
 
+        $progress->finish();
+
+        $output->writeln('');
+        $this->io->text('Pushed total '.$pushed);
         $this->io->success('Execution time: '.number_format(microtime(true) - $_SERVER['REQUEST_TIME_FLOAT'], 3));
     }
+
+    /**
+     * @param EntityManager $em
+     * @param               $start
+     * @param               $limit
+     *
+     * @return array
+     */
+    private function getLeads(EntityManager $em, $start, $limit)
+    {
+        return $em->getRepository(Lead::class)->getEntities(
+            [
+                'filter' => [
+                    'force' => [
+                        [
+                            'column' => 'l.email',
+                            'expr'   => 'neq',
+                            'value'  => '',
+                        ],
+                        [
+                            'column' => 'l.firstname',
+                            'expr'   => 'neq',
+                            'value'  => '',
+                        ],
+                        [
+                            'column' => 'l.lastname',
+                            'expr'   => 'neq',
+                            'value'  => '',
+                        ],
+                    ],
+                ],
+                'start'            => $start,
+                'limit'            => $limit,
+                'ignore_paginator' => true,
+            ]
+        );
+    }
 }

plugins/MauticCrmBundle/Config/config.php

@@ -144,6 +144,13 @@
                     'mautic_integration.pipedrive.export.company',
                 ],
             ],
+            'mautic_integration.pipedrive.export.activities' => [
+                'class'     => \MauticPlugin\MauticCrmBundle\Integration\Pipedrive\Export\ActivitiesPipedriveExport::class,
+                'arguments' => [
+                    'doctrine.orm.entity_manager',
+                    'mautic.lead.model.lead',
+                ],
+            ],
         ],
         'forms' => [
             'mautic.form.type.connectwise.campaignaction' => [