3 namespace Wallabag\ImportBundle\Import
;
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
;
17 abstract class AbstractImport
implements ImportInterface
21 protected $contentProxy;
22 protected $tagsAssigner;
23 protected $eventDispatcher;
26 protected $markAsRead;
27 protected $disableContentUpdate = false;
28 protected $skippedEntries = 0;
29 protected $importedEntries = 0;
30 protected $queuedEntries = 0;
32 public function __construct(EntityManager
$em, ContentProxy
$contentProxy, TagsAssigner
$tagsAssigner, EventDispatcherInterface
$eventDispatcher)
35 $this->logger
= new NullLogger();
36 $this->contentProxy
= $contentProxy;
37 $this->tagsAssigner
= $tagsAssigner;
38 $this->eventDispatcher
= $eventDispatcher;
41 public function setLogger(LoggerInterface
$logger)
43 $this->logger
= $logger;
47 * Set RabbitMQ/Redis Producer to send each entry to a queue.
48 * This method should be called when user has enabled RabbitMQ.
50 * @param ProducerInterface $producer
52 public function setProducer(ProducerInterface
$producer)
54 $this->producer
= $producer;
59 * Could the current *connected* user or one retrieve by the consumer.
63 public function setUser(User
$user)
69 * Set whether articles must be all marked as read.
71 * @param bool $markAsRead
73 public function setMarkAsRead($markAsRead)
75 $this->markAsRead
= $markAsRead;
81 * Get whether articles must be all marked as read.
83 public function getMarkAsRead()
85 return $this->markAsRead
;
89 * Set whether articles should be fetched for updated content.
91 * @param bool $disableContentUpdate
93 public function setDisableContentUpdate($disableContentUpdate)
95 $this->disableContentUpdate
= $disableContentUpdate;
103 public function getSummary()
106 'skipped' => $this->skippedEntries
,
107 'imported' => $this->importedEntries
,
108 'queued' => $this->queuedEntries
,
115 * @param array $importedEntry
119 abstract public function parseEntry(array $importedEntry);
122 * Validate that an entry is valid (like has some required keys, etc.).
124 * @param array $importedEntry
128 abstract public function validateEntry(array $importedEntry);
131 * Fetch content from the ContentProxy (using graby).
132 * If it fails return the given entry to be saved in all case (to avoid user to loose the content).
134 * @param Entry $entry Entry to update
135 * @param string $url Url to grab content for
136 * @param array $content An array with AT LEAST keys title, html, url, language & content_type to skip the fetchContent from the url
138 protected function fetchContent(Entry
$entry, $url, array $content = [])
141 $this->contentProxy
->updateEntry($entry, $url, $content, $this->disableContentUpdate
);
142 } catch (\Exception
$e) {
143 $this->logger
->error('Error trying to import an entry.', [
145 'error_msg' => $e->getMessage(),
151 * Parse and insert all given entries.
153 * @param array $entries
155 protected function parseEntries(array $entries)
158 $entryToBeFlushed = [];
160 foreach ($entries as $importedEntry) {
161 if ($this->markAsRead
) {
162 $importedEntry = $this->setEntryAsRead($importedEntry);
165 if (false === $this->validateEntry($importedEntry)) {
169 $entry = $this->parseEntry($importedEntry);
171 if (null === $entry) {
175 // store each entry to be flushed so we can trigger the entry.saved event for each of them
176 // entry.saved needs the entry to be persisted in db because it needs it id to generate
178 $entryToBeFlushed[] = $entry;
180 // flush every 20 entries
181 if (0 === ($i %
20)) {
184 foreach ($entryToBeFlushed as $entry) {
185 $this->eventDispatcher
->dispatch(EntrySavedEvent
::NAME
, new EntrySavedEvent($entry));
188 $entryToBeFlushed = [];
190 // clear only affected entities
191 $this->em
->clear(Entry
::class);
192 $this->em
->clear(Tag
::class);
199 if (!empty($entryToBeFlushed)) {
200 foreach ($entryToBeFlushed as $entry) {
201 $this->eventDispatcher
->dispatch(EntrySavedEvent
::NAME
, new EntrySavedEvent($entry));
207 * Parse entries and send them to the queue.
208 * It should just be a simple loop on all item, no call to the database should be done
209 * to speedup queuing.
211 * Faster parse entries for Producer.
212 * We don't care to make check at this time. They'll be done by the consumer.
214 * @param array $entries
216 protected function parseEntriesForProducer(array $entries)
218 foreach ($entries as $importedEntry) {
219 // set userId for the producer (it won't know which user is connected)
220 $importedEntry['userId'] = $this->user
->getId();
222 if ($this->markAsRead
) {
223 $importedEntry = $this->setEntryAsRead($importedEntry);
226 ++
$this->queuedEntries
;
228 $this->producer
->publish(json_encode($importedEntry));
233 * Set current imported entry to archived / read.
234 * Implementation is different accross all imports.
236 * @param array $importedEntry
240 abstract protected function setEntryAsRead(array $importedEntry);