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