src/Wallabag/ImportBundle/Import/AbstractImport.php

   1 <?php
   2
   3 namespace Wallabag\ImportBundle\Import;
   4
   5 use Doctrine\ORM\EntityManager;
   6 use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;
   7 use Psr\Log\LoggerInterface;
   8 use Psr\Log\NullLogger;
   9 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
  10 use Wallabag\CoreBundle\Entity\Entry;
  11 use Wallabag\CoreBundle\Entity\Tag;
  12 use Wallabag\CoreBundle\Event\EntrySavedEvent;
  13 use Wallabag\CoreBundle\Helper\ContentProxy;
  14 use Wallabag\CoreBundle\Helper\TagsAssigner;
  15 use Wallabag\UserBundle\Entity\User;
  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      * {@inheritdoc}
 102      */
 103     public function getSummary()
 104     {
 105         return [
 106             'skipped' => $this->skippedEntries,
 107             'imported' => $this->importedEntries,
 108             'queued' => $this->queuedEntries,
 109         ];
 110     }
 111
 112     /**
 113      * Parse one entry.
 114      *
 115      * @param array $importedEntry
 116      *
 117      * @return Entry
 118      */
 119     abstract public function parseEntry(array $importedEntry);
 120
 121     /**
 122      * Fetch content from the ContentProxy (using graby).
 123      * If it fails return the given entry to be saved in all case (to avoid user to loose the content).
 124      *
 125      * @param Entry  $entry   Entry to update
 126      * @param string $url     Url to grab content for
 127      * @param array  $content An array with AT LEAST keys title, html, url, language & content_type to skip the fetchContent from the url
 128      */
 129     protected function fetchContent(Entry $entry, $url, array $content = [])
 130     {
 131         try {
 132             $this->contentProxy->updateEntry($entry, $url, $content, $this->disableContentUpdate);
 133         } catch (\Exception $e) {
 134             $this->logger->error('Error trying to import an entry.', [
 135                 'entry_url' => $url,
 136                 'error_msg' => $e->getMessage(),
 137             ]);
 138         }
 139     }
 140
 141     /**
 142      * Parse and insert all given entries.
 143      *
 144      * @param $entries
 145      */
 146     protected function parseEntries($entries)
 147     {
 148         $i = 1;
 149         $entryToBeFlushed = [];
 150
 151         foreach ($entries as $importedEntry) {
 152             if ($this->markAsRead) {
 153                 $importedEntry = $this->setEntryAsRead($importedEntry);
 154             }
 155
 156             $entry = $this->parseEntry($importedEntry);
 157
 158             if (null === $entry) {
 159                 continue;
 160             }
 161
 162             // store each entry to be flushed so we can trigger the entry.saved event for each of them
 163             // entry.saved needs the entry to be persisted in db because it needs it id to generate
 164             // images (at least)
 165             $entryToBeFlushed[] = $entry;
 166
 167             // flush every 20 entries
 168             if (0 === ($i % 20)) {
 169                 $this->em->flush();
 170
 171                 foreach ($entryToBeFlushed as $entry) {
 172                     $this->eventDispatcher->dispatch(EntrySavedEvent::NAME, new EntrySavedEvent($entry));
 173                 }
 174
 175                 $entryToBeFlushed = [];
 176
 177                 // clear only affected entities
 178                 $this->em->clear(Entry::class);
 179                 $this->em->clear(Tag::class);
 180             }
 181             ++$i;
 182         }
 183
 184         $this->em->flush();
 185
 186         if (!empty($entryToBeFlushed)) {
 187             foreach ($entryToBeFlushed as $entry) {
 188                 $this->eventDispatcher->dispatch(EntrySavedEvent::NAME, new EntrySavedEvent($entry));
 189             }
 190         }
 191     }
 192
 193     /**
 194      * Parse entries and send them to the queue.
 195      * It should just be a simple loop on all item, no call to the database should be done
 196      * to speedup queuing.
 197      *
 198      * Faster parse entries for Producer.
 199      * We don't care to make check at this time. They'll be done by the consumer.
 200      *
 201      * @param array $entries
 202      */
 203     protected function parseEntriesForProducer(array $entries)
 204     {
 205         foreach ($entries as $importedEntry) {
 206             // set userId for the producer (it won't know which user is connected)
 207             $importedEntry['userId'] = $this->user->getId();
 208
 209             if ($this->markAsRead) {
 210                 $importedEntry = $this->setEntryAsRead($importedEntry);
 211             }
 212
 213             ++$this->queuedEntries;
 214
 215             $this->producer->publish(json_encode($importedEntry));
 216         }
 217     }
 218
 219     /**
 220      * Set current imported entry to archived / read.
 221      * Implementation is different accross all imports.
 222      *
 223      * @param array $importedEntry
 224      *
 225      * @return array
 226      */
 227     abstract protected function setEntryAsRead(array $importedEntry);
 228 }