src/Wallabag/ImportBundle/Import/AbstractImport.php

   1 <?php
   2
   3 namespace Wallabag\ImportBundle\Import;
   4
   5 use Psr\Log\LoggerInterface;
   6 use Psr\Log\NullLogger;
   7 use Doctrine\ORM\EntityManager;
   8 use Wallabag\CoreBundle\Helper\ContentProxy;
   9 use Wallabag\CoreBundle\Entity\Entry;
  10 use Wallabag\CoreBundle\Entity\Tag;
  11 use Wallabag\UserBundle\Entity\User;
  12 use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;
  13 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
  14 use Wallabag\CoreBundle\Event\EntrySavedEvent;
  15
  16 abstract class AbstractImport implements ImportInterface
  17 {
  18     protected $em;
  19     protected $logger;
  20     protected $contentProxy;
  21     protected $eventDispatcher;
  22     protected $producer;
  23     protected $user;
  24     protected $markAsRead;
  25     protected $skippedEntries = 0;
  26     protected $importedEntries = 0;
  27     protected $queuedEntries = 0;
  28
  29     public function __construct(EntityManager $em, ContentProxy $contentProxy, EventDispatcherInterface $eventDispatcher)
  30     {
  31         $this->em = $em;
  32         $this->logger = new NullLogger();
  33         $this->contentProxy = $contentProxy;
  34         $this->eventDispatcher = $eventDispatcher;
  35     }
  36
  37     public function setLogger(LoggerInterface $logger)
  38     {
  39         $this->logger = $logger;
  40     }
  41
  42     /**
  43      * Set RabbitMQ/Redis Producer to send each entry to a queue.
  44      * This method should be called when user has enabled RabbitMQ.
  45      *
  46      * @param ProducerInterface $producer
  47      */
  48     public function setProducer(ProducerInterface $producer)
  49     {
  50         $this->producer = $producer;
  51     }
  52
  53     /**
  54      * Set current user.
  55      * Could the current *connected* user or one retrieve by the consumer.
  56      *
  57      * @param User $user
  58      */
  59     public function setUser(User $user)
  60     {
  61         $this->user = $user;
  62     }
  63
  64     /**
  65      * Set whether articles must be all marked as read.
  66      *
  67      * @param bool $markAsRead
  68      */
  69     public function setMarkAsRead($markAsRead)
  70     {
  71         $this->markAsRead = $markAsRead;
  72
  73         return $this;
  74     }
  75
  76     /**
  77      * Get whether articles must be all marked as read.
  78      */
  79     public function getMarkAsRead()
  80     {
  81         return $this->markAsRead;
  82     }
  83
  84     /**
  85      * Fetch content from the ContentProxy (using graby).
  86      * If it fails return the given entry to be saved in all case (to avoid user to loose the content).
  87      *
  88      * @param Entry  $entry   Entry to update
  89      * @param string $url     Url to grab content for
  90      * @param array  $content An array with AT LEAST keys title, html, url, language & content_type to skip the fetchContent from the url
  91      *
  92      * @return Entry
  93      */
  94     protected function fetchContent(Entry $entry, $url, array $content = [])
  95     {
  96         try {
  97             return $this->contentProxy->updateEntry($entry, $url, $content);
  98         } catch (\Exception $e) {
  99             return $entry;
 100         }
 101     }
 102
 103     /**
 104      * Parse and insert all given entries.
 105      *
 106      * @param $entries
 107      */
 108     protected function parseEntries($entries)
 109     {
 110         $i = 1;
 111         $entryToBeFlushed = [];
 112
 113         foreach ($entries as $importedEntry) {
 114             if ($this->markAsRead) {
 115                 $importedEntry = $this->setEntryAsRead($importedEntry);
 116             }
 117
 118             $entry = $this->parseEntry($importedEntry);
 119
 120             if (null === $entry) {
 121                 continue;
 122             }
 123
 124             // store each entry to be flushed so we can trigger the entry.saved event for each of them
 125             // entry.saved needs the entry to be persisted in db because it needs it id to generate
 126             // images (at least)
 127             $entryToBeFlushed[] = $entry;
 128
 129             // flush every 20 entries
 130             if (($i % 20) === 0) {
 131                 $this->em->flush();
 132
 133                 foreach ($entryToBeFlushed as $entry) {
 134                     $this->eventDispatcher->dispatch(EntrySavedEvent::NAME, new EntrySavedEvent($entry));
 135                 }
 136
 137                 $entryToBeFlushed = [];
 138
 139                 // clear only affected entities
 140                 $this->em->clear(Entry::class);
 141                 $this->em->clear(Tag::class);
 142             }
 143             ++$i;
 144         }
 145
 146         $this->em->flush();
 147
 148         if (!empty($entryToBeFlushed)) {
 149             foreach ($entryToBeFlushed as $entry) {
 150                 $this->eventDispatcher->dispatch(EntrySavedEvent::NAME, new EntrySavedEvent($entry));
 151             }
 152         }
 153     }
 154
 155     /**
 156      * Parse entries and send them to the queue.
 157      * It should just be a simple loop on all item, no call to the database should be done
 158      * to speedup queuing.
 159      *
 160      * Faster parse entries for Producer.
 161      * We don't care to make check at this time. They'll be done by the consumer.
 162      *
 163      * @param array $entries
 164      */
 165     protected function parseEntriesForProducer(array $entries)
 166     {
 167         foreach ($entries as $importedEntry) {
 168             // set userId for the producer (it won't know which user is connected)
 169             $importedEntry['userId'] = $this->user->getId();
 170
 171             if ($this->markAsRead) {
 172                 $importedEntry = $this->setEntryAsRead($importedEntry);
 173             }
 174
 175             ++$this->queuedEntries;
 176
 177             $this->producer->publish(json_encode($importedEntry));
 178         }
 179     }
 180
 181     /**
 182      * {@inheritdoc}
 183      */
 184     public function getSummary()
 185     {
 186         return [
 187             'skipped' => $this->skippedEntries,
 188             'imported' => $this->importedEntries,
 189             'queued' => $this->queuedEntries,
 190         ];
 191     }
 192
 193     /**
 194      * Parse one entry.
 195      *
 196      * @param array $importedEntry
 197      *
 198      * @return Entry
 199      */
 200     abstract public function parseEntry(array $importedEntry);
 201
 202     /**
 203      * Set current imported entry to archived / read.
 204      * Implementation is different accross all imports.
 205      *
 206      * @param array $importedEntry
 207      *
 208      * @return array
 209      */
 210     abstract protected function setEntryAsRead(array $importedEntry);
 211 }