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