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         // be sure to set at least the given url
 119         $content['url'] = isset($content['url']) ? $content['url'] : $url;
 120
 121         try {
 122             $this->contentProxy->importEntry($entry, $content, $this->disableContentUpdate);
 123         } catch (\Exception $e) {
 124             $this->logger->error('Error trying to import an entry.', [
 125                 'entry_url' => $content['url'],
 126                 'error_msg' => $e->getMessage(),
 127             ]);
 128         }
 129     }
 130
 131     /**
 132      * Parse and insert all given entries.
 133      *
 134      * @param $entries
 135      */
 136     protected function parseEntries($entries)
 137     {
 138         $i = 1;
 139         $entryToBeFlushed = [];
 140
 141         foreach ($entries as $importedEntry) {
 142             if ($this->markAsRead) {
 143                 $importedEntry = $this->setEntryAsRead($importedEntry);
 144             }
 145
 146             $entry = $this->parseEntry($importedEntry);
 147
 148             if (null === $entry) {
 149                 continue;
 150             }
 151
 152             // store each entry to be flushed so we can trigger the entry.saved event for each of them
 153             // entry.saved needs the entry to be persisted in db because it needs it id to generate
 154             // images (at least)
 155             $entryToBeFlushed[] = $entry;
 156
 157             // flush every 20 entries
 158             if (($i % 20) === 0) {
 159                 $this->em->flush();
 160
 161                 foreach ($entryToBeFlushed as $entry) {
 162                     $this->eventDispatcher->dispatch(EntrySavedEvent::NAME, new EntrySavedEvent($entry));
 163                 }
 164
 165                 $entryToBeFlushed = [];
 166
 167                 // clear only affected entities
 168                 $this->em->clear(Entry::class);
 169                 $this->em->clear(Tag::class);
 170             }
 171             ++$i;
 172         }
 173
 174         $this->em->flush();
 175
 176         if (!empty($entryToBeFlushed)) {
 177             foreach ($entryToBeFlushed as $entry) {
 178                 $this->eventDispatcher->dispatch(EntrySavedEvent::NAME, new EntrySavedEvent($entry));
 179             }
 180         }
 181     }
 182
 183     /**
 184      * Parse entries and send them to the queue.
 185      * It should just be a simple loop on all item, no call to the database should be done
 186      * to speedup queuing.
 187      *
 188      * Faster parse entries for Producer.
 189      * We don't care to make check at this time. They'll be done by the consumer.
 190      *
 191      * @param array $entries
 192      */
 193     protected function parseEntriesForProducer(array $entries)
 194     {
 195         foreach ($entries as $importedEntry) {
 196             // set userId for the producer (it won't know which user is connected)
 197             $importedEntry['userId'] = $this->user->getId();
 198
 199             if ($this->markAsRead) {
 200                 $importedEntry = $this->setEntryAsRead($importedEntry);
 201             }
 202
 203             ++$this->queuedEntries;
 204
 205             $this->producer->publish(json_encode($importedEntry));
 206         }
 207     }
 208
 209     /**
 210      * {@inheritdoc}
 211      */
 212     public function getSummary()
 213     {
 214         return [
 215             'skipped' => $this->skippedEntries,
 216             'imported' => $this->importedEntries,
 217             'queued' => $this->queuedEntries,
 218         ];
 219     }
 220
 221     /**
 222      * Parse one entry.
 223      *
 224      * @param array $importedEntry
 225      *
 226      * @return Entry
 227      */
 228     abstract public function parseEntry(array $importedEntry);
 229
 230     /**
 231      * Set current imported entry to archived / read.
 232      * Implementation is different accross all imports.
 233      *
 234      * @param array $importedEntry
 235      *
 236      * @return array
 237      */
 238     abstract protected function setEntryAsRead(array $importedEntry);
 239 }