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