QueueRepository::unsetProcessScheduledAndProcessIdForQueueEntries() - Code Metrics - Inspection of "[TASK] Move Queries to Repositories" - AOEpeople/crawler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — cleanup/crawlercontroller ( c98170...0af167 )

by Tomas Norre

created 2020-04-25 13:43 UTC

unsetProcessScheduledAndProcessIdForQueueEntries() A

↳ Parent: QueueRepository

Complexity

Conditions	1
Paths	1

Size

Total Lines	12
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	1.0156

Importance

Changes

Metric	Value
cc	1
eloc	9
nc	1
nop	1
dl	0
loc	12
ccs	9
cts	12
cp	0.75
crap	1.0156
rs	9.9666
c	0
b	0
f	0

<?php

declare(strict_types=1);

namespace AOE\Crawler\Domain\Repository;

/***************************************************************
 *  Copyright notice
 *
 *  (c) 2020 AOE GmbH <[email protected]>
 *
 *  All rights reserved
 *
 *  This script is part of the TYPO3 project. The TYPO3 project is
 *  free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  The GNU General Public License can be found at
 *  http://www.gnu.org/copyleft/gpl.html.
 *
 *  This script is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  This copyright notice MUST APPEAR in all copies of the script!
 ***************************************************************/

use AOE\Crawler\Configuration\ExtensionConfigurationProvider;
use AOE\Crawler\Domain\Model\Process;
use Psr\Log\LoggerAwareInterface;
use TYPO3\CMS\Core\Database\Connection;
use TYPO3\CMS\Core\Database\ConnectionPool;
use TYPO3\CMS\Core\Utility\GeneralUtility;
use TYPO3\CMS\Core\Utility\MathUtility;

/**
 * Class QueueRepository
 *
 * @package AOE\Crawler\Domain\Repository
 */
class QueueRepository extends AbstractRepository implements LoggerAwareInterface
{
    use \Psr\Log\LoggerAwareTrait;

    /**
     * @var string
     */
    protected $tableName = 'tx_crawler_queue';

    public function __construct()
    {
        // Left empty intentional
    }

    public function unsetQueueProcessId(string $processId): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $queryBuilder
            ->update($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId))
            )
            ->set('process_id', '')
            ->execute();
    }

    /**
     * This method is used to find the youngest entry for a given process.
     *
     * @param Process $process
     *
     * @return array
     */
    public function findYoungestEntryForProcess(Process $process): array
    {
        return $this->getFirstOrLastObjectByProcess($process, 'exec_time');
    }

    /**
     * This method is used to find the oldest entry for a given process.
     *
     * @param Process $process
     *
     * @return array
     */
    public function findOldestEntryForProcess(Process $process): array
    {
        return $this->getFirstOrLastObjectByProcess($process, 'exec_time', 'DESC');
    }

    /**
     * This internal helper method is used to create an instance of an entry object
     *
     * @param Process $process
     * @param string $orderByField first matching item will be returned as object
     * @param string $orderBySorting sorting direction
     *
     * @return array
     */
    protected function getFirstOrLastObjectByProcess($process, $orderByField, $orderBySorting = 'ASC'): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $first = $queryBuilder
            ->select('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
                $queryBuilder->expr()->gt('exec_time', 0)
            )
            ->setMaxResults(1)
            ->addOrderBy($orderByField, $orderBySorting)
            ->execute()->fetch(0);

        return $first ?: [];
    }

    /**
     * Counts all executed items of a process.
     *
     * @param Process $process
     *
     * @return int
     */
    public function countExecutedItemsByProcess($process): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
                $queryBuilder->expr()->gt('exec_time', 0)
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * Counts items of a process which yet have not been processed/executed
     *
     * @param Process $process
     *
     * @return int
     */
    public function countNonExecutedItemsByProcess($process): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($process->getProcessId())),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * get items which have not been processed yet
     *
     * @return array
     */
    public function getUnprocessedItems(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->select('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->execute()->fetchAll();
    }

    /**
     * Count items which have not been processed yet
     *
     * @return int
     */
    public function countUnprocessedItems(): int
    {
        return count($this->getUnprocessedItems());
    }

    /**
     * This method can be used to count all queue entrys which are
     * scheduled for now or a earlier date.
     *
     * @return int
     */
    public function countAllPendingItems(): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_scheduled', 0),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * This method can be used to count all queue entries which are
     * scheduled for now or a earlier date and are assigned to a process.
     *
     * @return int
     */
    public function countAllAssignedPendingItems(): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->neq('process_id', '""'),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * This method can be used to count all queue entrys which are
     * scheduled for now or a earlier date and are not assigned to a process.
     *
     * @return int
     */
    public function countAllUnassignedPendingItems(): int
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);

        return $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('process_id', '""'),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->execute()
            ->fetchColumn(0);
    }

    /**
     * Count pending queue entries grouped by configuration key
     *
     * @return array
     */
    public function countPendingItemsGroupedByConfigurationKey(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->selectLiteral('count(*) as unprocessed', 'sum(process_id != \'\') as assignedButUnprocessed')
            ->addSelect('configuration')
            ->where(
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lt('scheduled', time())
            )
            ->groupBy('configuration')
            ->execute();

        return $statement->fetchAll();
    }

    /**
     * Get set id with unprocessed entries
     *
     * @return array array of set ids
     */
    public function getSetIdWithUnprocessedEntries(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->select('set_id')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->lt('scheduled', time()),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->addGroupBy('set_id')
            ->execute();

        $setIds = [];
        while ($row = $statement->fetch()) {
            $setIds[] = intval($row['set_id']);
        }

        return $setIds;
    }

    /**
     * Get total queue entries by configuration
     *
     * @param array $setIds
     *
     * @return array totals by configuration (keys)
     */
    public function getTotalQueueEntriesByConfiguration(array $setIds): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $totals = [];
        if (count($setIds) > 0) {
            $statement = $queryBuilder
                ->from($this->tableName)
                ->selectLiteral('count(*) as c')
                ->addSelect('configuration')
                ->where(
                    $queryBuilder->expr()->in('set_id', implode(',', $setIds)),
                    $queryBuilder->expr()->lt('scheduled', time())
                )
                ->groupBy('configuration')
                ->execute();

            while ($row = $statement->fetch()) {
                $totals[$row['configuration']] = $row['c'];
            }
        }

        return $totals;
    }

    /**
     * Get the timestamps of the last processed entries
     *
     * @param int $limit
     *
     * @return array
     */
    public function getLastProcessedEntriesTimestamps($limit = 100): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->select('exec_time')
            ->from($this->tableName)
            ->addOrderBy('exec_time', 'desc')
            ->setMaxResults($limit)
            ->execute();

        $rows = [];
        while ($row = $statement->fetch()) {
            $rows[] = $row['exec_time'];
        }

        return $rows;
    }

    /**
     * Get the last processed entries
     *
     * @param int $limit
     *
     * @return array
     */
    public function getLastProcessedEntries($limit = 100): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->select('*')
            ->orderBy('exec_time', 'desc')
            ->setMaxResults($limit)
            ->execute();

        $rows = [];
        while (($row = $statement->fetch()) !== false) {
            $rows[] = $row;
        }

        return $rows;
    }

    /**
     * Get performance statistics data
     *
     * @param int $start timestamp
     * @param int $end timestamp
     *
     * @return array performance data
     */
    public function getPerformanceData($start, $end): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->selectLiteral('min(exec_time) as start', 'max(exec_time) as end', 'count(*) as urlcount')
            ->addSelect('process_id_completed')
            ->where(
                $queryBuilder->expr()->neq('exec_time', 0),
                $queryBuilder->expr()->gte('exec_time', $queryBuilder->createNamedParameter($start, \PDO::PARAM_INT)),
                $queryBuilder->expr()->lte('exec_time', $queryBuilder->createNamedParameter($end, \PDO::PARAM_INT))
            )
            ->groupBy('process_id_completed')
            ->execute();

        $rows = [];
        while ($row = $statement->fetch()) {
            $rows[$row['process_id_completed']] = $row;
        }

        return $rows;
    }

    /**
     * Determines if a page is queued
     */
    public function isPageInQueue(int $uid, bool $unprocessed_only = true, bool $timed_only = false, int $timestamp = 0): bool
    {
        // TODO: Looks like the check is obsolete as PHP doesn't allow other than ints for the $uid, PHP 7.2 Strict Mode
        if (!MathUtility::canBeInterpretedAsInteger($uid)) {
            throw new \InvalidArgumentException('Invalid parameter type', 1468931945);
        }

        $isPageInQueue = false;

        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->from($this->tableName)
            ->count('*')
            ->where(
                $queryBuilder->expr()->eq('page_id', $queryBuilder->createNamedParameter($uid, \PDO::PARAM_INT))
            );

        if (false !== $unprocessed_only) {
            $statement->andWhere(
                $queryBuilder->expr()->eq('exec_time', 0)
            );
        }

        if (false !== $timed_only) {
            $statement->andWhere(
                $queryBuilder->expr()->neq('scheduled', 0)
            );
        }

        if ($timestamp) {
            $statement->andWhere(
                $queryBuilder->expr()->eq('scheduled', $queryBuilder->createNamedParameter($timestamp, \PDO::PARAM_INT))
            );
        }

        // TODO: Currently it's not working if page doesn't exists. See tests
        $count = $statement
            ->execute()
            ->fetchColumn(0);

        if (false !== $count && $count > 0) {
            $isPageInQueue = true;
        }

        return $isPageInQueue;
    }

    /**
     * Method to check if a page is in the queue which is timed for a
     * date when it should be crawled
     */
    public function isPageInQueueTimed(int $uid, bool $show_unprocessed = true): bool
    {
        return $this->isPageInQueue($uid, $show_unprocessed);
    }

    public function getAvailableSets(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $statement = $queryBuilder
            ->selectLiteral('count(*) as count_value')
            ->addSelect('set_id', 'scheduled')
            ->from($this->tableName)
            ->orderBy('scheduled', 'desc')
            ->groupBy('set_id', 'scheduled')
            ->execute();

        $rows = [];
        while ($row = $statement->fetch()) {
            $rows[] = $row;
        }

        return $rows;
    }

    public function findByQueueId(string $queueId): ?array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $queueRec = $queryBuilder
            ->select('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('qid', $queryBuilder->createNamedParameter($queueId))
            )
            ->execute()
            ->fetch();
        return is_array($queueRec) ? $queueRec : null;
    }

    public function cleanupQueue(): void
    {
        $extensionSettings = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class)->getExtensionConfiguration();
        $purgeDays = (int)$extensionSettings['purgeQueueDays'];

        if ($purgeDays > 0) {
            $purgeDate = time() - 24 * 60 * 60 * $purgeDays;

            $queryBuilderDelete = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
            $del = $queryBuilderDelete
                ->delete($this->tableName)
                ->where(
                    'exec_time != 0 AND exec_time < ' . $purgeDate
                )->execute();

            if (false === $del) {
                $this->logger->info(
                    'Records could not be deleted.'
                );
            }
        }
    }

    /**
     * @param int $countInARun
     * @return mixed
     */
    public function fetchRecordsToBeCrawled(int $countInARun)
    {
        $queryBuilderSelect = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $rows = $queryBuilderSelect
            ->select('qid', 'scheduled')
            ->from($this->tableName)
            ->where(
                $queryBuilderSelect->expr()->eq('exec_time', 0),
                $queryBuilderSelect->expr()->eq('process_scheduled', 0),
                $queryBuilderSelect->expr()->lte('scheduled', time())
            )
            ->orderBy('scheduled')
            ->addOrderBy('qid')
            ->setMaxResults($countInARun)
            ->execute()
            ->fetchAll();
        return $rows;
    }

    /**
     * @param array $quidList
     * @param string $processId
     * @return mixed
     */
    public function updateProcessIdAndSchedulerForQueueIds(array $quidList, string $processId)
    {
        $queryBuilderUpdate = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $numberOfAffectedRows = $queryBuilderUpdate
            ->update($this->tableName)
            ->where(
                $queryBuilderUpdate->expr()->in('qid', $quidList)
            )
            ->set('process_scheduled', time())
            ->set('process_id', $processId)
            ->execute();
        return $numberOfAffectedRows;
    }

    /**
     * @param array $processIds
     */
    public function unsetProcessScheduledAndProcessIdForQueueEntries(array $processIds): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $queryBuilder
            ->update($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->in('process_id', $queryBuilder->createNamedParameter($processIds, Connection::PARAM_STR_ARRAY))
            )
            ->set('process_scheduled', 0)
            ->set('process_id', '')
            ->execute();
    }

    /**
     * This method is used to count if there are ANY unprocessed queue entries
     * of a given page_id and the configuration which matches a given hash.
     * If there if none, we can skip an inner detail check
     *
     * @param int $uid
     * @param string $configurationHash
     * @return boolean
     */
    public function noUnprocessedQueueEntriesForPageWithConfigurationHashExist($uid, $configurationHash): bool
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
        $noUnprocessedQueueEntriesFound = true;

        $result = $queryBuilder
            ->count('*')
            ->from($this->tableName)
            ->where(
                $queryBuilder->expr()->eq('page_id', (int)$uid),
                $queryBuilder->expr()->eq('configuration_hash', $queryBuilder->createNamedParameter($configurationHash)),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->execute()
            ->fetchColumn();

        if ($result) {
            $noUnprocessedQueueEntriesFound = false;
        }

        return $noUnprocessedQueueEntriesFound;
    }
}


1		<?php
2
3		declare(strict_types=1);
4
5		namespace AOE\Crawler\Domain\Repository;
6
7		/***************************************************************
8		* Copyright notice
9		*
10		* (c) 2020 AOE GmbH <[email protected]>
11		*
12		* All rights reserved
13		*
14		* This script is part of the TYPO3 project. The TYPO3 project is
15		* free software; you can redistribute it and/or modify
16		* it under the terms of the GNU General Public License as published by
17		* the Free Software Foundation; either version 3 of the License, or
18		* (at your option) any later version.
19		*
20		* The GNU General Public License can be found at
21		* http://www.gnu.org/copyleft/gpl.html.
22		*
23		* This script is distributed in the hope that it will be useful,
24		* but WITHOUT ANY WARRANTY; without even the implied warranty of
25		* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
26		* GNU General Public License for more details.
27		*
28		* This copyright notice MUST APPEAR in all copies of the script!
29		***************************************************************/
30
31		use AOE\Crawler\Configuration\ExtensionConfigurationProvider;
32		use AOE\Crawler\Domain\Model\Process;
33		use Psr\Log\LoggerAwareInterface;
34		use TYPO3\CMS\Core\Database\Connection;
35		use TYPO3\CMS\Core\Database\ConnectionPool;
36		use TYPO3\CMS\Core\Utility\GeneralUtility;
37		use TYPO3\CMS\Core\Utility\MathUtility;
38
39		/**
40		* Class QueueRepository
41		*
42		* @package AOE\Crawler\Domain\Repository
43		*/
44		class QueueRepository extends AbstractRepository implements LoggerAwareInterface
45		{
46		use \Psr\Log\LoggerAwareTrait;
47
48		/**
49		* @var string
50		*/
51		protected $tableName = 'tx_crawler_queue';
52
53	88	public function __construct()
54		{
55		// Left empty intentional
56	88	}
57
58	3	public function unsetQueueProcessId(string $processId): void
59		{
60	3	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
61		$queryBuilder
62	3	->update($this->tableName)
63	3	->where(
64	3	$queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId))
65		)
66	3	->set('process_id', '')
67	3	->execute();
68	3	}
69
70		/**
71		* This method is used to find the youngest entry for a given process.
72		*
73		* @param Process $process
74		*
75		* @return array
76		*/
77	1	public function findYoungestEntryForProcess(Process $process): array
78		{
79	1	return $this->getFirstOrLastObjectByProcess($process, 'exec_time');
80		}
81
82		/**
83		* This method is used to find the oldest entry for a given process.
84		*
85		* @param Process $process
86		*
87		* @return array
88		*/
89	1	public function findOldestEntryForProcess(Process $process): array
90		{
91	1	return $this->getFirstOrLastObjectByProcess($process, 'exec_time', 'DESC');
92		}
93
94		/**
95		* This internal helper method is used to create an instance of an entry object
96		*
97		* @param Process $process
98		* @param string $orderByField first matching item will be returned as object
99		* @param string $orderBySorting sorting direction
100		*
101		* @return array
102		*/
103	5	protected function getFirstOrLastObjectByProcess($process, $orderByField, $orderBySorting = 'ASC'): array
104		{
105	5	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
106		$first = $queryBuilder
107	5	->select('*')
108	5	->from($this->tableName)
109	5	->where(
110	5	$queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
111	5	$queryBuilder->expr()->gt('exec_time', 0)
112		)
113	5	->setMaxResults(1)
114	5	->addOrderBy($orderByField, $orderBySorting)
115	5	->execute()->fetch(0);
116
117	5	return $first ?: [];
118		}
119
120		/**
121		* Counts all executed items of a process.
122		*
123		* @param Process $process
124		*
125		* @return int
126		*/
127	1	public function countExecutedItemsByProcess($process): int
128		{
129	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
130
131		return $queryBuilder
132	1	->count('*')
133	1	->from($this->tableName)
134	1	->where(
135	1	$queryBuilder->expr()->eq('process_id_completed', $queryBuilder->createNamedParameter($process->getProcessId())),
136	1	$queryBuilder->expr()->gt('exec_time', 0)
137		)
138	1	->execute()
139	1	->fetchColumn(0);
140		}
141
142		/**
143		* Counts items of a process which yet have not been processed/executed
144		*
145		* @param Process $process
146		*
147		* @return int
148		*/
149	1	public function countNonExecutedItemsByProcess($process): int
150		{
151	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
152
153		return $queryBuilder
154	1	->count('*')
155	1	->from($this->tableName)
156	1	->where(
157	1	$queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($process->getProcessId())),
158	1	$queryBuilder->expr()->eq('exec_time', 0)
159		)
160	1	->execute()
161	1	->fetchColumn(0);
162		}
163
164		/**
165		* get items which have not been processed yet
166		*
167		* @return array
168		*/
169	3	public function getUnprocessedItems(): array
170		{
171	3	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
172
173		return $queryBuilder
174	3	->select('*')
175	3	->from($this->tableName)
176	3	->where(
177	3	$queryBuilder->expr()->eq('exec_time', 0)
178		)
179	3	->execute()->fetchAll();
180		}
181
182		/**
183		* Count items which have not been processed yet
184		*
185		* @return int
186		*/
187	3	public function countUnprocessedItems(): int
188		{
189	3	return count($this->getUnprocessedItems());
190		}
191
192		/**
193		* This method can be used to count all queue entrys which are
194		* scheduled for now or a earlier date.
195		*
196		* @return int
197		*/
198	2	public function countAllPendingItems(): int
199		{
200	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
201
202		return $queryBuilder
203	2	->count('*')
204	2	->from($this->tableName)
205	2	->where(
206	2	$queryBuilder->expr()->eq('process_scheduled', 0),
207	2	$queryBuilder->expr()->eq('exec_time', 0),
208	2	$queryBuilder->expr()->lte('scheduled', time())
209		)
210	2	->execute()
211	2	->fetchColumn(0);
212		}
213
214		/**
215		* This method can be used to count all queue entries which are
216		* scheduled for now or a earlier date and are assigned to a process.
217		*
218		* @return int
219		*/
220	2	public function countAllAssignedPendingItems(): int
221		{
222	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
223
224		return $queryBuilder
225	2	->count('*')
226	2	->from($this->tableName)
227	2	->where(
228	2	$queryBuilder->expr()->neq('process_id', '""'),
229	2	$queryBuilder->expr()->eq('exec_time', 0),
230	2	$queryBuilder->expr()->lte('scheduled', time())
231		)
232	2	->execute()
233	2	->fetchColumn(0);
234		}
235
236		/**
237		* This method can be used to count all queue entrys which are
238		* scheduled for now or a earlier date and are not assigned to a process.
239		*
240		* @return int
241		*/
242	2	public function countAllUnassignedPendingItems(): int
243		{
244	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
245
246		return $queryBuilder
247	2	->count('*')
248	2	->from($this->tableName)
249	2	->where(
250	2	$queryBuilder->expr()->eq('process_id', '""'),
251	2	$queryBuilder->expr()->eq('exec_time', 0),
252	2	$queryBuilder->expr()->lte('scheduled', time())
253		)
254	2	->execute()
255	2	->fetchColumn(0);
256		}
257
258		/**
259		* Count pending queue entries grouped by configuration key
260		*
261		* @return array
262		*/
263	1	public function countPendingItemsGroupedByConfigurationKey(): array
264		{
265	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
266		$statement = $queryBuilder
267	1	->from($this->tableName)
268	1	->selectLiteral('count(*) as unprocessed', 'sum(process_id != \'\') as assignedButUnprocessed')
269	1	->addSelect('configuration')
270	1	->where(
271	1	$queryBuilder->expr()->eq('exec_time', 0),
272	1	$queryBuilder->expr()->lt('scheduled', time())
273		)
274	1	->groupBy('configuration')
275	1	->execute();
276
277	1	return $statement->fetchAll();
278		}
279
280		/**
281		* Get set id with unprocessed entries
282		*
283		* @return array array of set ids
284		*/
285	1	public function getSetIdWithUnprocessedEntries(): array
286		{
287	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
288		$statement = $queryBuilder
289	1	->select('set_id')
290	1	->from($this->tableName)
291	1	->where(
292	1	$queryBuilder->expr()->lt('scheduled', time()),
293	1	$queryBuilder->expr()->eq('exec_time', 0)
294		)
295	1	->addGroupBy('set_id')
296	1	->execute();
297
298	1	$setIds = [];
299	1	while ($row = $statement->fetch()) {
300	1	$setIds[] = intval($row['set_id']);
301		}
302
303	1	return $setIds;
304		}
305
306		/**
307		* Get total queue entries by configuration
308		*
309		* @param array $setIds
310		*
311		* @return array totals by configuration (keys)
312		*/
313	1	public function getTotalQueueEntriesByConfiguration(array $setIds): array
314		{
315	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
316	1	$totals = [];
317	1	if (count($setIds) > 0) {
318		$statement = $queryBuilder
319	1	->from($this->tableName)
320	1	->selectLiteral('count(*) as c')
321	1	->addSelect('configuration')
322	1	->where(
323	1	$queryBuilder->expr()->in('set_id', implode(',', $setIds)),
324	1	$queryBuilder->expr()->lt('scheduled', time())
325		)
326	1	->groupBy('configuration')
327	1	->execute();
328
329	1	while ($row = $statement->fetch()) {
330	1	$totals[$row['configuration']] = $row['c'];
331		}
332		}
333
334	1	return $totals;
335		}
336
337		/**
338		* Get the timestamps of the last processed entries
339		*
340		* @param int $limit
341		*
342		* @return array
343		*/
344	1	public function getLastProcessedEntriesTimestamps($limit = 100): array
345		{
346	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
347		$statement = $queryBuilder
348	1	->select('exec_time')
349	1	->from($this->tableName)
350	1	->addOrderBy('exec_time', 'desc')
351	1	->setMaxResults($limit)
352	1	->execute();
353
354	1	$rows = [];
355	1	while ($row = $statement->fetch()) {
356	1	$rows[] = $row['exec_time'];
357		}
358
359	1	return $rows;
360		}
361
362		/**
363		* Get the last processed entries
364		*
365		* @param int $limit
366		*
367		* @return array
368		*/
369	1	public function getLastProcessedEntries($limit = 100): array
370		{
371	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
372		$statement = $queryBuilder
373	1	->from($this->tableName)
374	1	->select('*')
375	1	->orderBy('exec_time', 'desc')
376	1	->setMaxResults($limit)
377	1	->execute();
378
379	1	$rows = [];
380	1	while (($row = $statement->fetch()) !== false) {
381	1	$rows[] = $row;
382		}
383
384	1	return $rows;
385		}
386
387		/**
388		* Get performance statistics data
389		*
390		* @param int $start timestamp
391		* @param int $end timestamp
392		*
393		* @return array performance data
394		*/
395	1	public function getPerformanceData($start, $end): array
396		{
397	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
398		$statement = $queryBuilder
399	1	->from($this->tableName)
400	1	->selectLiteral('min(exec_time) as start', 'max(exec_time) as end', 'count(*) as urlcount')
401	1	->addSelect('process_id_completed')
402	1	->where(
403	1	$queryBuilder->expr()->neq('exec_time', 0),
404	1	$queryBuilder->expr()->gte('exec_time', $queryBuilder->createNamedParameter($start, \PDO::PARAM_INT)),
405	1	$queryBuilder->expr()->lte('exec_time', $queryBuilder->createNamedParameter($end, \PDO::PARAM_INT))
406		)
407	1	->groupBy('process_id_completed')
408	1	->execute();
409
410	1	$rows = [];
411	1	while ($row = $statement->fetch()) {
412	1	$rows[$row['process_id_completed']] = $row;
413		}
414
415	1	return $rows;
416		}
417
418		/**
419		* Determines if a page is queued
420		*/
421	5	public function isPageInQueue(int $uid, bool $unprocessed_only = true, bool $timed_only = false, int $timestamp = 0): bool
422		{
423		// TODO: Looks like the check is obsolete as PHP doesn't allow other than ints for the $uid, PHP 7.2 Strict Mode
424	5	if (!MathUtility::canBeInterpretedAsInteger($uid)) {
425		throw new \InvalidArgumentException('Invalid parameter type', 1468931945);
426		}
427
428	5	$isPageInQueue = false;
429
430	5	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
431		$statement = $queryBuilder
432	5	->from($this->tableName)
433	5	->count('*')
434	5	->where(
435	5	$queryBuilder->expr()->eq('page_id', $queryBuilder->createNamedParameter($uid, \PDO::PARAM_INT))
436		);
437
438	5	if (false !== $unprocessed_only) {
439	2	$statement->andWhere(
440	2	$queryBuilder->expr()->eq('exec_time', 0)
441		);
442		}
443
444	5	if (false !== $timed_only) {
445	1	$statement->andWhere(
446	1	$queryBuilder->expr()->neq('scheduled', 0)
447		);
448		}
449
450	5	if ($timestamp) {
451	1	$statement->andWhere(
452	1	$queryBuilder->expr()->eq('scheduled', $queryBuilder->createNamedParameter($timestamp, \PDO::PARAM_INT))
453		);
454		}
455
456		// TODO: Currently it's not working if page doesn't exists. See tests
457		$count = $statement
458	5	->execute()
459	5	->fetchColumn(0);
460
461	5	if (false !== $count && $count > 0) {
462	4	$isPageInQueue = true;
463		}
464
465	5	return $isPageInQueue;
466		}
467
468		/**
469		* Method to check if a page is in the queue which is timed for a
470		* date when it should be crawled
471		*/
472	1	public function isPageInQueueTimed(int $uid, bool $show_unprocessed = true): bool
473		{
474	1	return $this->isPageInQueue($uid, $show_unprocessed);
475		}
476
477	1	public function getAvailableSets(): array
478		{
479	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
480		$statement = $queryBuilder
481	1	->selectLiteral('count(*) as count_value')
482	1	->addSelect('set_id', 'scheduled')
483	1	->from($this->tableName)
484	1	->orderBy('scheduled', 'desc')
485	1	->groupBy('set_id', 'scheduled')
486	1	->execute();
487
488	1	$rows = [];
489	1	while ($row = $statement->fetch()) {
490	1	$rows[] = $row;
491		}
492
493	1	return $rows;
494		}
495
496	1	public function findByQueueId(string $queueId): ?array
497		{
498	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
499		$queueRec = $queryBuilder
500	1	->select('*')
501	1	->from($this->tableName)
502	1	->where(
503	1	$queryBuilder->expr()->eq('qid', $queryBuilder->createNamedParameter($queueId))
504		)
505	1	->execute()
506	1	->fetch();
507	1	return is_array($queueRec) ? $queueRec : null;
508		}
509
510	1	public function cleanupQueue(): void
511		{
512	1	$extensionSettings = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class)->getExtensionConfiguration();
513	1	$purgeDays = (int)$extensionSettings['purgeQueueDays'];
514
515	1	if ($purgeDays > 0) {
516	1	$purgeDate = time() - 24 * 60 * 60 * $purgeDays;
517
518	1	$queryBuilderDelete = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
519		$del = $queryBuilderDelete
520	1	->delete($this->tableName)
521	1	->where(
522	1	'exec_time != 0 AND exec_time < ' . $purgeDate
523	1	)->execute();
524
525	1	if (false === $del) {
526		$this->logger->info(
527		'Records could not be deleted.'
528		);
529		}
530		}
531	1	}
532
533		/**
534		* @param int $countInARun
535		* @return mixed
536		*/
537	1	public function fetchRecordsToBeCrawled(int $countInARun)
538		{
539	1	$queryBuilderSelect = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
540		$rows = $queryBuilderSelect
541	1	->select('qid', 'scheduled')
542	1	->from($this->tableName)
543	1	->where(
544	1	$queryBuilderSelect->expr()->eq('exec_time', 0),
545	1	$queryBuilderSelect->expr()->eq('process_scheduled', 0),
546	1	$queryBuilderSelect->expr()->lte('scheduled', time())
547		)
548	1	->orderBy('scheduled')
549	1	->addOrderBy('qid')
550	1	->setMaxResults($countInARun)
551	1	->execute()
552	1	->fetchAll();
553	1	return $rows;
554		}
555
556		/**
557		* @param array $quidList
558		* @param string $processId
559		* @return mixed
560		*/
561	1	public function updateProcessIdAndSchedulerForQueueIds(array $quidList, string $processId)
562		{
563	1	$queryBuilderUpdate = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
564		$numberOfAffectedRows = $queryBuilderUpdate
565	1	->update($this->tableName)
566	1	->where(
567	1	$queryBuilderUpdate->expr()->in('qid', $quidList)
568		)
569	1	->set('process_scheduled', time())
570	1	->set('process_id', $processId)
571	1	->execute();
572	1	return $numberOfAffectedRows;
573		}
574
575		/**
576		* @param array $processIds
577		*/
578	1	public function unsetProcessScheduledAndProcessIdForQueueEntries(array $processIds): void
579		{
580	1	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
581		$queryBuilder
582	1	->update($this->tableName)
583	1	->where(
584	1	$queryBuilder->expr()->eq('exec_time', 0),
585	1	$queryBuilder->expr()->in('process_id', $queryBuilder->createNamedParameter($processIds, Connection::PARAM_STR_ARRAY))
586		)
587	1	->set('process_scheduled', 0)
588	1	->set('process_id', '')
589	1	->execute();
590	1	}
591
592		/**
593		* This method is used to count if there are ANY unprocessed queue entries
594		* of a given page_id and the configuration which matches a given hash.
595		* If there if none, we can skip an inner detail check
596		*
597		* @param int $uid
598		* @param string $configurationHash
599		* @return boolean
600		*/
601	2	public function noUnprocessedQueueEntriesForPageWithConfigurationHashExist($uid, $configurationHash): bool
602		{
603	2	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable($this->tableName);
604	2	$noUnprocessedQueueEntriesFound = true;
605
606		$result = $queryBuilder
607	2	->count('*')
608	2	->from($this->tableName)
609	2	->where(
610	2	$queryBuilder->expr()->eq('page_id', (int)$uid),
611	2	$queryBuilder->expr()->eq('configuration_hash', $queryBuilder->createNamedParameter($configurationHash)),
612	2	$queryBuilder->expr()->eq('exec_time', 0)
613		)
614	2	->execute()
615	2	->fetchColumn();
616
617	2	if ($result) {
618	2	$noUnprocessedQueueEntriesFound = false;
619		}
620
621	2	return $noUnprocessedQueueEntriesFound;
622		}
623		}
624

AOEpeople / crawler

Push — cleanup/crawlercontroller ( c98170...0af167 )

unsetProcessScheduledAndProcessIdForQueueEntries() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like