11463038090

Committed 22 Oct 2024 03:04PM UTC coverage: 69.182% (+0.2%) from 68.96%

Build # 11463038090

Build Type

Pull #1113

github

Committed by

web-flow

Commit Message

Merge 3cf3baf1f into e54dc1476

Pull Request Pull Request #1113: CI: Add TYPO3 v13 to test matrix

Run Details

40 of 40 new or added lines in 3 files covered. (100.0%)

3 existing lines in 2 files now uncovered.

1870 of 2703 relevant lines covered (69.18%)

3.28 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.33

/Classes/Domain/Repository/QueueRepository.php

<?php

declare(strict_types=1);

namespace AOE\Crawler\Domain\Repository;

/*
 * (c) 2021 Tomas Norre Mikkelsen <tomasnorre@gmail.com>
 *
 * This file is part of the TYPO3 Crawler Extension.
 *
 * It is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License, either version 2
 * of the License, or any later version.
 *
 * For the full copyright and license information, please read the
 * LICENSE.txt file that was distributed with this source code.
 *
 * The TYPO3 project - inspiring people to share!
 */

use AOE\Crawler\Configuration\ExtensionConfigurationProvider;
use AOE\Crawler\Domain\Model\Process;
use AOE\Crawler\Value\QueueFilter;
use Doctrine\DBAL\ArrayParameterType;
use Doctrine\DBAL\Exception;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use Symfony\Contracts\Service\Attribute\Required;
use TYPO3\CMS\Core\Database\Connection;
use TYPO3\CMS\Core\Database\ConnectionPool;
use TYPO3\CMS\Core\Utility\GeneralUtility;

/**
 * @internal since v9.2.5
 */
class QueueRepository implements LoggerAwareInterface
{
    use LoggerAwareTrait;

    final public const TABLE_NAME = 'tx_crawler_queue';

    protected array $extensionSettings;

    #[Required]
    public function setExtensionSettings(): void
    {
        /** @var ExtensionConfigurationProvider $configurationProvider */
        $configurationProvider = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class);
        $this->extensionSettings = $configurationProvider->getExtensionConfiguration();
    }

    // TODO: Should be a property on the QueueObject
    public function unsetQueueProcessId(string $processId): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $queryBuilder
            ->update(self::TABLE_NAME)
            ->where($queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId)))
            ->set('process_id', '')
            ->executeStatement();
    }

    /**
     * This method is used to find the youngest entry for a given process.
     */
    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.
     */
    public function findOldestEntryForProcess(Process $process): array
    {
        return $this->getFirstOrLastObjectByProcess($process, 'exec_time', 'DESC');
    }

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

        return (int) $queryBuilder
            ->count('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq(
                    'process_id_completed',
                    $queryBuilder->createNamedParameter($process->getProcessId())
                ),
                $queryBuilder->expr()->gt('exec_time', 0)
            )
            ->executeQuery()
            ->fetchOne();
    }

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

        return (int) $queryBuilder
            ->count('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($process->getProcessId())),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->executeQuery()
            ->fetchOne();
    }

    public function getQueueEntriesByQid(int $queueId, bool $force): array|false
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);

        $queryBuilder
            ->select('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq('qid', $queryBuilder->createNamedParameter($queueId, Connection::PARAM_INT))
            );
        if (!$force) {
            $queryBuilder
                ->andWhere('exec_time = 0')
                ->andWhere('process_scheduled > 0');
        }
        return $queryBuilder->executeQuery()->fetchAssociative();
    }

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

        return $queryBuilder
            ->select('*')
            ->from(self::TABLE_NAME)
            ->where($queryBuilder->expr()->eq('exec_time', 0))
            ->executeQuery()->fetchAllAssociative();
    }

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

        return (int) $queryBuilder
            ->count('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq('process_scheduled', 0),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->executeQuery()
            ->fetchOne();
    }

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

        return (int) $queryBuilder
            ->count('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->neq('process_id', $queryBuilder->createNamedParameter(0, Connection::PARAM_INT)),
                $queryBuilder->expr()->eq('exec_time', 0),
                $queryBuilder->expr()->lte('scheduled', time())
            )
            ->executeQuery()
            ->fetchOne();
    }

    /**
     * Determines if a page is queued
     */
    public function isPageInQueue(
        int $uid,
        bool $unprocessed_only = true,
        bool $timed_only = false,
        int $timestamp = 0
    ): bool {
        $isPageInQueue = false;

        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $statement = $queryBuilder
            ->from(self::TABLE_NAME)
            ->count('*')
            ->where(
                $queryBuilder->expr()->eq(
                    'page_id',
                    $queryBuilder->createNamedParameter($uid, Connection::PARAM_INT)
                )
            );

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

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

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

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

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

        return $isPageInQueue;
    }

    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(
                self::TABLE_NAME
            );
            $del = $queryBuilderDelete
                ->delete(self::TABLE_NAME)
                ->where('exec_time != 0 AND exec_time < ' . $purgeDate)->executeStatement();

            if ($del === 0) {
                $this->logger?->info('No records was deleted');
            }
        }
    }

    /**
     * Cleans up entries that stayed for too long in the queue. These are default:
     * - processed entries that are over 1.5 days in age
     * - scheduled entries that are over 7 days old
     */
    public function cleanUpOldQueueEntries(): void
    {
        $extensionSettings = GeneralUtility::makeInstance(
            ExtensionConfigurationProvider::class
        )->getExtensionConfiguration();
        // 24*60*60 Seconds in 24 hours
        $processedAgeInSeconds = $extensionSettings['cleanUpProcessedAge'] * 86400;
        $scheduledAgeInSeconds = $extensionSettings['cleanUpScheduledAge'] * 86400;

        $now = time();
        $condition = '(exec_time<>0 AND exec_time<' . ($now - $processedAgeInSeconds) . ') OR scheduled<=' . ($now - $scheduledAgeInSeconds);

        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $del = $queryBuilder
            ->delete(self::TABLE_NAME)
            ->where($condition)->executeStatement();

        if ($del === 0) {
            $this->logger?->info('No records was deleted.');
        }
    }

    public function fetchRecordsToBeCrawled(int $countInARun): array
    {
        $queryBuilderSelect = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(
            self::TABLE_NAME
        );
        return $queryBuilderSelect
            ->select('qid', 'scheduled', 'page_id', 'sitemap_priority')
            ->from(self::TABLE_NAME)
            ->leftJoin(
                self::TABLE_NAME,
                'pages',
                'p',
                $queryBuilderSelect->expr()->eq(
                    'p.uid',
                    $queryBuilderSelect->quoteIdentifier(self::TABLE_NAME . '.page_id')
                )
            )
            ->where(
                $queryBuilderSelect->expr()->eq('exec_time', 0),
                $queryBuilderSelect->expr()->eq('process_scheduled', 0),
                $queryBuilderSelect->expr()->lte('scheduled', time())
            )
            ->orderBy('sitemap_priority', 'DESC')
            ->addOrderBy('scheduled')
            ->addOrderBy('qid')
            ->setMaxResults($countInARun)
            ->executeQuery()
            ->fetchAllAssociative();
    }

    public function updateProcessIdAndSchedulerForQueueIds(array $quidList, string $processId): int
    {
        $queryBuilderUpdate = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(
            self::TABLE_NAME
        );
        return $queryBuilderUpdate
            ->update(self::TABLE_NAME)
            ->where($queryBuilderUpdate->expr()->in('qid', $quidList))
            ->set('process_scheduled', (string) time())
            ->set('process_id', $processId)
            ->executeStatement();
    }

    public function unsetProcessScheduledAndProcessIdForQueueEntries(array $processIds): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $queryBuilder
            ->update(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq('exec_time', $queryBuilder->createNamedParameter(0, Connection::PARAM_INT)),
                $queryBuilder->expr()->in(
                    'process_id',
                    $queryBuilder->createNamedParameter($processIds, ArrayParameterType::STRING)
                )
            )
            ->set('process_scheduled', '0')
            ->set('process_id', '0')
            ->executeStatement();
    }

    /**
     * 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 is none, we can skip an inner detail check
     */
    public function noUnprocessedQueueEntriesForPageWithConfigurationHashExist(
        int $uid,
        string $configurationHash
    ): bool {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $noUnprocessedQueueEntriesFound = true;

        $result = $queryBuilder
            ->count('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq('page_id', $uid),
                $queryBuilder->expr()->eq(
                    'configuration_hash',
                    $queryBuilder->createNamedParameter($configurationHash)
                ),
                $queryBuilder->expr()->eq('exec_time', 0)
            )
            ->executeQuery()
            ->fetchOne();

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

        return $noUnprocessedQueueEntriesFound;
    }

    /**
     * Removes queue entries
     */
    public function flushQueue(QueueFilter $queueFilter): void
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);

        switch ($queueFilter) {
            case 'all':
                // No where claus needed delete everything
                break;
            case 'pending':
                $queryBuilder->andWhere($queryBuilder->expr()->eq('exec_time', 0));
                break;
            case 'finished':
            default:
                $queryBuilder->andWhere($queryBuilder->expr()->gt('exec_time', 0));
                break;
        }

        $queryBuilder
            ->delete(self::TABLE_NAME)
            ->executeStatement();
    }

    public function getDuplicateQueueItemsIfExists(
        bool $enableTimeslot,
        int $timestamp,
        int $currentTime,
        int $pageId,
        string $parametersHash
    ): array {
        $rows = [];

        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $queryBuilder
            ->select('qid')
            ->from(self::TABLE_NAME);
        //if this entry is scheduled with "now"
        if ($timestamp <= $currentTime) {
            if ($enableTimeslot) {
                $timeBegin = $currentTime - 100;
                $timeEnd = $currentTime + 100;
                $queryBuilder
                    ->where('scheduled BETWEEN ' . $timeBegin . ' AND ' . $timeEnd . ' ')
                    ->orWhere($queryBuilder->expr()->lte('scheduled', $currentTime));
            } else {
                $queryBuilder
                    ->where($queryBuilder->expr()->lte('scheduled', $currentTime));
            }
        } elseif ($timestamp > $currentTime) {
            //entry with a timestamp in the future need to have the same schedule time
            $queryBuilder
                ->where($queryBuilder->expr()->eq('scheduled', $timestamp));
        }

        $queryBuilder
            ->andWhere($queryBuilder->expr()->eq('exec_time', 0))
            ->andWhere($queryBuilder->expr()->eq('process_id', "''"))
            ->andWhere(
                $queryBuilder->expr()->eq(
                    'page_id',
                    $queryBuilder->createNamedParameter($pageId, Connection::PARAM_INT)
                )
            )
            ->andWhere(
                $queryBuilder->expr()->eq('parameters_hash', $queryBuilder->createNamedParameter(
                    $parametersHash,
                    Connection::PARAM_STR
                ))
            );

        $statement = $queryBuilder->executeQuery();

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

        return $rows;
    }

    public function getQueueEntriesForPageId(int $id, int $itemsPerPage, QueueFilter $queueFilter): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $queryBuilder
            ->select('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq('page_id', $queryBuilder->createNamedParameter(
                    $id,
                    Connection::PARAM_INT
                ))
            )
            ->orderBy('scheduled', 'DESC');

        $expressionBuilder = GeneralUtility::makeInstance(ConnectionPool::class)
            ->getConnectionForTable(self::TABLE_NAME)
            ->getExpressionBuilder();
        $query = $expressionBuilder->and();
        // PHPStorm adds the highlight that the $addWhere is immediately overwritten,
        // but the $query = $expressionBuilder->andX() ensures that the $addWhere is written correctly with AND
        // between the statements, it's not a mistake in the code.
        switch ($queueFilter) {
            case 'pending':
                $queryBuilder->andWhere($queryBuilder->expr()->eq('exec_time', 0));
                break;
            case 'finished':
                $queryBuilder->andWhere($queryBuilder->expr()->gt('exec_time', 0));
                break;
        }

        if ($itemsPerPage > 0) {
            $queryBuilder
                ->setMaxResults($itemsPerPage);
        }

        return $queryBuilder->executeQuery()->fetchAllAssociative();
    }

    /**
     * @throws Exception
     */
    public function findAll(): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $queryBuilder
            ->select('*')
            ->from(self::TABLE_NAME);
        return $queryBuilder->executeQuery()->fetchAllAssociative();
    }

    public function findByProcessId(string $processId): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        return $queryBuilder
            ->select('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter(
                    $processId,
                    Connection::PARAM_INT
                ))
            )
            ->executeQuery()->fetchAllAssociative();
    }

    /**
     * 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
     */
    protected function getFirstOrLastObjectByProcess($process, $orderByField, $orderBySorting = 'ASC'): array
    {
        $queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);
        $first = $queryBuilder
            ->select('*')
            ->from(self::TABLE_NAME)
            ->where(
                $queryBuilder->expr()->eq(
                    'process_id_completed',
                    $queryBuilder->createNamedParameter($process->getProcessId())
                ),
                $queryBuilder->expr()->gt('exec_time', 0)
            )
            ->setMaxResults(1)
            ->addOrderBy($orderByField, $orderBySorting)
            ->executeQuery()->fetchAssociative();

        return $first ?: [];
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	namespace AOE\Crawler\Domain\Repository;
6
7	/*
8	* (c) 2021 Tomas Norre Mikkelsen <tomasnorre@gmail.com>
9	*
10	* This file is part of the TYPO3 Crawler Extension.
11	*
12	* It is free software; you can redistribute it and/or modify it under
13	* the terms of the GNU General Public License, either version 2
14	* of the License, or any later version.
15	*
16	* For the full copyright and license information, please read the
17	* LICENSE.txt file that was distributed with this source code.
18	*
19	* The TYPO3 project - inspiring people to share!
20	*/
21
22	use AOE\Crawler\Configuration\ExtensionConfigurationProvider;
23	use AOE\Crawler\Domain\Model\Process;
24	use AOE\Crawler\Value\QueueFilter;
25	use Doctrine\DBAL\ArrayParameterType;
26	use Doctrine\DBAL\Exception;
27	use Psr\Log\LoggerAwareInterface;
28	use Psr\Log\LoggerAwareTrait;
29	use Symfony\Contracts\Service\Attribute\Required;
30	use TYPO3\CMS\Core\Database\Connection;
31	use TYPO3\CMS\Core\Database\ConnectionPool;
32	use TYPO3\CMS\Core\Utility\GeneralUtility;
33
34	/**
35	* @internal since v9.2.5
36	*/
37	class QueueRepository implements LoggerAwareInterface
38	{
39	use LoggerAwareTrait;
40
41	final public const TABLE_NAME = 'tx_crawler_queue';
42
43	protected array $extensionSettings;
44
45	#[Required]
46	public function setExtensionSettings(): void
47	{
48	/** @var ExtensionConfigurationProvider $configurationProvider */
UNCOV 49	$configurationProvider = GeneralUtility::makeInstance(ExtensionConfigurationProvider::class);	×
UNCOV 50	$this->extensionSettings = $configurationProvider->getExtensionConfiguration();	×
51	}
52
53	// TODO: Should be a property on the QueueObject
54	public function unsetQueueProcessId(string $processId): void
55	{
56	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	3✔
57	$queryBuilder	3✔
58	->update(self::TABLE_NAME)	3✔
59	->where($queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($processId)))	3✔
60	->set('process_id', '')	3✔
61	->executeStatement();	3✔
62	}
63
64	/**
65	* This method is used to find the youngest entry for a given process.
66	*/
67	public function findYoungestEntryForProcess(Process $process): array
68	{
69	return $this->getFirstOrLastObjectByProcess($process, 'exec_time');	1✔
70	}
71
72	/**
73	* This method is used to find the oldest entry for a given process.
74	*/
75	public function findOldestEntryForProcess(Process $process): array
76	{
77	return $this->getFirstOrLastObjectByProcess($process, 'exec_time', 'DESC');	1✔
78	}
79
80	/**
81	* Counts all executed items of a process.
82	*
83	* @param Process $process
84	*/
85	public function countExecutedItemsByProcess($process): int
86	{
87	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	1✔
88
89	return (int) $queryBuilder	1✔
90	->count('*')	1✔
91	->from(self::TABLE_NAME)	1✔
92	->where(	1✔
93	$queryBuilder->expr()->eq(	1✔
94	'process_id_completed',	1✔
95	$queryBuilder->createNamedParameter($process->getProcessId())	1✔
96	),	1✔
97	$queryBuilder->expr()->gt('exec_time', 0)	1✔
98	)	1✔
99	->executeQuery()	1✔
100	->fetchOne();	1✔
101	}
102
103	/**
104	* Counts items of a process which yet have not been processed/executed
105	*
106	* @param Process $process
107	*/
108	public function countNonExecutedItemsByProcess($process): int
109	{
110	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	1✔
111
112	return (int) $queryBuilder	1✔
113	->count('*')	1✔
114	->from(self::TABLE_NAME)	1✔
115	->where(	1✔
116	$queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter($process->getProcessId())),	1✔
117	$queryBuilder->expr()->eq('exec_time', 0)	1✔
118	)	1✔
119	->executeQuery()	1✔
120	->fetchOne();	1✔
121	}
122
123	public function getQueueEntriesByQid(int $queueId, bool $force): array\|false
124	{
125	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	3✔
126
127	$queryBuilder	3✔
128	->select('*')	3✔
129	->from(self::TABLE_NAME)	3✔
130	->where(	3✔
131	$queryBuilder->expr()->eq('qid', $queryBuilder->createNamedParameter($queueId, Connection::PARAM_INT))	3✔
132	);	3✔
133	if (!$force) {	3✔
134	$queryBuilder	2✔
135	->andWhere('exec_time = 0')	2✔
136	->andWhere('process_scheduled > 0');	2✔
137	}
138	return $queryBuilder->executeQuery()->fetchAssociative();	3✔
139	}
140
141	/**
142	* get items which have not been processed yet
143	*/
144	public function getUnprocessedItems(): array
145	{
146	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	5✔
147
148	return $queryBuilder	5✔
149	->select('*')	5✔
150	->from(self::TABLE_NAME)	5✔
151	->where($queryBuilder->expr()->eq('exec_time', 0))	5✔
152	->executeQuery()->fetchAllAssociative();	5✔
153	}
154
155	/**
156	* This method can be used to count all queue entrys which are
157	* scheduled for now or a earlier date.
158	*/
159	public function countAllPendingItems(): int
160	{
161	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	3✔
162
163	return (int) $queryBuilder	3✔
164	->count('*')	3✔
165	->from(self::TABLE_NAME)	3✔
166	->where(	3✔
167	$queryBuilder->expr()->eq('process_scheduled', 0),	3✔
168	$queryBuilder->expr()->eq('exec_time', 0),	3✔
169	$queryBuilder->expr()->lte('scheduled', time())	3✔
170	)	3✔
171	->executeQuery()	3✔
172	->fetchOne();	3✔
173	}
174
175	/**
176	* This method can be used to count all queue entries which are
177	* scheduled for now or an earlier date and are assigned to a process.
178	*/
179	public function countAllAssignedPendingItems(): int
180	{
181	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	3✔
182
183	return (int) $queryBuilder	3✔
184	->count('*')	3✔
185	->from(self::TABLE_NAME)	3✔
186	->where(	3✔
187	$queryBuilder->expr()->neq('process_id', $queryBuilder->createNamedParameter(0, Connection::PARAM_INT)),	3✔
188	$queryBuilder->expr()->eq('exec_time', 0),	3✔
189	$queryBuilder->expr()->lte('scheduled', time())	3✔
190	)	3✔
191	->executeQuery()	3✔
192	->fetchOne();	3✔
193	}
194
195	/**
196	* Determines if a page is queued
197	*/
198	public function isPageInQueue(
199	int $uid,
200	bool $unprocessed_only = true,
201	bool $timed_only = false,
202	int $timestamp = 0
203	): bool {
204	$isPageInQueue = false;	4✔
205
206	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	4✔
207	$statement = $queryBuilder	4✔
208	->from(self::TABLE_NAME)	4✔
209	->count('*')	4✔
210	->where(	4✔
211	$queryBuilder->expr()->eq(	4✔
212	'page_id',	4✔
213	$queryBuilder->createNamedParameter($uid, Connection::PARAM_INT)	4✔
214	)	4✔
215	);	4✔
216
217	if ($unprocessed_only !== false) {	4✔
218	$statement->andWhere($queryBuilder->expr()->eq('exec_time', 0));	1✔
219	}
220
221	if ($timed_only !== false) {	4✔
222	$statement->andWhere($queryBuilder->expr()->neq('scheduled', 0));	1✔
223	}
224
225	if ($timestamp) {	4✔
226	$statement->andWhere(	1✔
227	$queryBuilder->expr()->eq(	1✔
228	'scheduled',	1✔
229	$queryBuilder->createNamedParameter($timestamp, Connection::PARAM_INT)	1✔
230	)	1✔
231	);	1✔
232	}
233
234	// TODO: Currently it's not working if page doesn't exists. See tests
235	$count = $statement	4✔
236	->executeQuery()	4✔
237	->fetchOne();	4✔
238
239	if ($count !== false && $count > 0) {	4✔
240	$isPageInQueue = true;	3✔
241	}
242
243	return $isPageInQueue;	4✔
244	}
245
246	public function cleanupQueue(): void
247	{
248	$extensionSettings = GeneralUtility::makeInstance(	3✔
249	ExtensionConfigurationProvider::class	3✔
250	)->getExtensionConfiguration();	3✔
251	$purgeDays = (int) $extensionSettings['purgeQueueDays'];	3✔
252
253	if ($purgeDays > 0) {	3✔
254	$purgeDate = time() - 24 * 60 * 60 * $purgeDays;	3✔
255
256	$queryBuilderDelete = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(	3✔
257	self::TABLE_NAME	3✔
258	);	3✔
259	$del = $queryBuilderDelete	3✔
260	->delete(self::TABLE_NAME)	3✔
261	->where('exec_time != 0 AND exec_time < ' . $purgeDate)->executeStatement();	3✔
262
263	if ($del === 0) {	3✔
264	$this->logger?->info('No records was deleted');	1✔
265	}
266	}
267	}
268
269	/**
270	* Cleans up entries that stayed for too long in the queue. These are default:
271	* - processed entries that are over 1.5 days in age
272	* - scheduled entries that are over 7 days old
273	*/
274	public function cleanUpOldQueueEntries(): void
275	{
276	$extensionSettings = GeneralUtility::makeInstance(	6✔
277	ExtensionConfigurationProvider::class	6✔
278	)->getExtensionConfiguration();	6✔
279	// 246060 Seconds in 24 hours
280	$processedAgeInSeconds = $extensionSettings['cleanUpProcessedAge'] * 86400;	6✔
281	$scheduledAgeInSeconds = $extensionSettings['cleanUpScheduledAge'] * 86400;	6✔
282
283	$now = time();	6✔
284	$condition = '(exec_time<>0 AND exec_time<' . ($now - $processedAgeInSeconds) . ') OR scheduled<=' . ($now - $scheduledAgeInSeconds);	6✔
285
286	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	6✔
287	$del = $queryBuilder	6✔
288	->delete(self::TABLE_NAME)	6✔
289	->where($condition)->executeStatement();	6✔
290
291	if ($del === 0) {	6✔
292	$this->logger?->info('No records was deleted.');	6✔
293	}
294	}
295
296	public function fetchRecordsToBeCrawled(int $countInARun): array
297	{
298	$queryBuilderSelect = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(	4✔
299	self::TABLE_NAME	4✔
300	);	4✔
301	return $queryBuilderSelect	4✔
302	->select('qid', 'scheduled', 'page_id', 'sitemap_priority')	4✔
303	->from(self::TABLE_NAME)	4✔
304	->leftJoin(	4✔
305	self::TABLE_NAME,	4✔
306	'pages',	4✔
307	'p',	4✔
308	$queryBuilderSelect->expr()->eq(	4✔
309	'p.uid',	4✔
310	$queryBuilderSelect->quoteIdentifier(self::TABLE_NAME . '.page_id')	4✔
311	)	4✔
312	)	4✔
313	->where(	4✔
314	$queryBuilderSelect->expr()->eq('exec_time', 0),	4✔
315	$queryBuilderSelect->expr()->eq('process_scheduled', 0),	4✔
316	$queryBuilderSelect->expr()->lte('scheduled', time())	4✔
317	)	4✔
318	->orderBy('sitemap_priority', 'DESC')	4✔
319	->addOrderBy('scheduled')	4✔
320	->addOrderBy('qid')	4✔
321	->setMaxResults($countInARun)	4✔
322	->executeQuery()	4✔
323	->fetchAllAssociative();	4✔
324	}
325
326	public function updateProcessIdAndSchedulerForQueueIds(array $quidList, string $processId): int
327	{
328	$queryBuilderUpdate = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(	3✔
329	self::TABLE_NAME	3✔
330	);	3✔
331	return $queryBuilderUpdate	3✔
332	->update(self::TABLE_NAME)	3✔
333	->where($queryBuilderUpdate->expr()->in('qid', $quidList))	3✔
334	->set('process_scheduled', (string) time())	3✔
335	->set('process_id', $processId)	3✔
336	->executeStatement();	3✔
337	}
338
339	public function unsetProcessScheduledAndProcessIdForQueueEntries(array $processIds): void
340	{
341	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	3✔
342	$queryBuilder	3✔
343	->update(self::TABLE_NAME)	3✔
344	->where(	3✔
345	$queryBuilder->expr()->eq('exec_time', $queryBuilder->createNamedParameter(0, Connection::PARAM_INT)),	3✔
346	$queryBuilder->expr()->in(	3✔
347	'process_id',	3✔
348	$queryBuilder->createNamedParameter($processIds, ArrayParameterType::STRING)	3✔
349	)	3✔
350	)	3✔
351	->set('process_scheduled', '0')	3✔
352	->set('process_id', '0')	3✔
353	->executeStatement();	3✔
354	}
355
356	/**
357	* This method is used to count if there are ANY unprocessed queue entries
358	* of a given page_id and the configuration which matches a given hash.
359	* If there is none, we can skip an inner detail check
360	*/
361	public function noUnprocessedQueueEntriesForPageWithConfigurationHashExist(
362	int $uid,
363	string $configurationHash
364	): bool {
365	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	8✔
366	$noUnprocessedQueueEntriesFound = true;	8✔
367
368	$result = $queryBuilder	8✔
369	->count('*')	8✔
370	->from(self::TABLE_NAME)	8✔
371	->where(	8✔
372	$queryBuilder->expr()->eq('page_id', $uid),	8✔
373	$queryBuilder->expr()->eq(	8✔
374	'configuration_hash',	8✔
375	$queryBuilder->createNamedParameter($configurationHash)	8✔
376	),	8✔
377	$queryBuilder->expr()->eq('exec_time', 0)	8✔
378	)	8✔
379	->executeQuery()	8✔
380	->fetchOne();	8✔
381
382	if ($result) {	8✔
383	$noUnprocessedQueueEntriesFound = false;	3✔
384	}
385
386	return $noUnprocessedQueueEntriesFound;	8✔
387	}
388
389	/**
390	* Removes queue entries
391	*/
392	public function flushQueue(QueueFilter $queueFilter): void
393	{
394	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	8✔
395
396	switch ($queueFilter) {
397	case 'all':	8✔
398	// No where claus needed delete everything
399	break;	4✔
400	case 'pending':	4✔
401	$queryBuilder->andWhere($queryBuilder->expr()->eq('exec_time', 0));	2✔
402	break;	2✔
403	case 'finished':	2✔
404	default:
405	$queryBuilder->andWhere($queryBuilder->expr()->gt('exec_time', 0));	2✔
406	break;	2✔
407	}
408
409	$queryBuilder	8✔
410	->delete(self::TABLE_NAME)	8✔
411	->executeStatement();	8✔
412	}
413
414	public function getDuplicateQueueItemsIfExists(
415	bool $enableTimeslot,
416	int $timestamp,
417	int $currentTime,
418	int $pageId,
419	string $parametersHash
420	): array {
421	$rows = [];	8✔
422
423	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	8✔
424	$queryBuilder	8✔
425	->select('qid')	8✔
426	->from(self::TABLE_NAME);	8✔
427	//if this entry is scheduled with "now"
428	if ($timestamp <= $currentTime) {	8✔
429	if ($enableTimeslot) {	3✔
430	$timeBegin = $currentTime - 100;	2✔
431	$timeEnd = $currentTime + 100;	2✔
432	$queryBuilder	2✔
433	->where('scheduled BETWEEN ' . $timeBegin . ' AND ' . $timeEnd . ' ')	2✔
434	->orWhere($queryBuilder->expr()->lte('scheduled', $currentTime));	2✔
435	} else {
436	$queryBuilder	3✔
437	->where($queryBuilder->expr()->lte('scheduled', $currentTime));	3✔
438	}
439	} elseif ($timestamp > $currentTime) {	5✔
440	//entry with a timestamp in the future need to have the same schedule time
441	$queryBuilder	5✔
442	->where($queryBuilder->expr()->eq('scheduled', $timestamp));	5✔
443	}
444
445	$queryBuilder	8✔
446	->andWhere($queryBuilder->expr()->eq('exec_time', 0))	8✔
447	->andWhere($queryBuilder->expr()->eq('process_id', "''"))	8✔
448	->andWhere(	8✔
449	$queryBuilder->expr()->eq(	8✔
450	'page_id',	8✔
451	$queryBuilder->createNamedParameter($pageId, Connection::PARAM_INT)	8✔
452	)	8✔
453	)	8✔
454	->andWhere(	8✔
455	$queryBuilder->expr()->eq('parameters_hash', $queryBuilder->createNamedParameter(	8✔
456	$parametersHash,	8✔
457	Connection::PARAM_STR	8✔
458	))	8✔
459	);	8✔
460
461	$statement = $queryBuilder->executeQuery();	8✔
462
463	while ($row = $statement->fetchAssociative()) {	8✔
464	$rows[] = $row['qid'];	2✔
465	}
466
467	return $rows;	8✔
468	}
469
470	public function getQueueEntriesForPageId(int $id, int $itemsPerPage, QueueFilter $queueFilter): array
471	{
472	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	4✔
473	$queryBuilder	4✔
474	->select('*')	4✔
475	->from(self::TABLE_NAME)	4✔
476	->where(	4✔
477	$queryBuilder->expr()->eq('page_id', $queryBuilder->createNamedParameter(	4✔
478	$id,	4✔
479	Connection::PARAM_INT	4✔
480	))	4✔
481	)	4✔
482	->orderBy('scheduled', 'DESC');	4✔
483
484	$expressionBuilder = GeneralUtility::makeInstance(ConnectionPool::class)	4✔
485	->getConnectionForTable(self::TABLE_NAME)	4✔
486	->getExpressionBuilder();	4✔
487	$query = $expressionBuilder->and();	4✔
488	// PHPStorm adds the highlight that the $addWhere is immediately overwritten,
489	// but the $query = $expressionBuilder->andX() ensures that the $addWhere is written correctly with AND
490	// between the statements, it's not a mistake in the code.
491	switch ($queueFilter) {
492	case 'pending':	4✔
493	$queryBuilder->andWhere($queryBuilder->expr()->eq('exec_time', 0));	1✔
494	break;	1✔
495	case 'finished':	3✔
496	$queryBuilder->andWhere($queryBuilder->expr()->gt('exec_time', 0));	1✔
497	break;	1✔
498	}
499
500	if ($itemsPerPage > 0) {	4✔
501	$queryBuilder	4✔
502	->setMaxResults($itemsPerPage);	4✔
503	}
504
505	return $queryBuilder->executeQuery()->fetchAllAssociative();	4✔
506	}
507
508	/**
509	* @throws Exception
510	*/
511	public function findAll(): array
512	{
513	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	16✔
514	$queryBuilder	16✔
515	->select('*')	16✔
516	->from(self::TABLE_NAME);	16✔
517	return $queryBuilder->executeQuery()->fetchAllAssociative();	16✔
518	}
519
520	public function findByProcessId(string $processId): array
521	{
522	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	1✔
523	return $queryBuilder	1✔
524	->select('*')	1✔
525	->from(self::TABLE_NAME)	1✔
526	->where(	1✔
527	$queryBuilder->expr()->eq('process_id', $queryBuilder->createNamedParameter(	1✔
528	$processId,	1✔
529	Connection::PARAM_INT	1✔
530	))	1✔
531	)	1✔
532	->executeQuery()->fetchAllAssociative();	1✔
533	}
534
535	/**
536	* This internal helper method is used to create an instance of an entry object
537	*
538	* @param Process $process
539	* @param string $orderByField first matching item will be returned as object
540	* @param string $orderBySorting sorting direction
541	*/
542	protected function getFirstOrLastObjectByProcess($process, $orderByField, $orderBySorting = 'ASC'): array
543	{
544	$queryBuilder = GeneralUtility::makeInstance(ConnectionPool::class)->getQueryBuilderForTable(self::TABLE_NAME);	5✔
545	$first = $queryBuilder	5✔
546	->select('*')	5✔
547	->from(self::TABLE_NAME)	5✔
548	->where(	5✔
549	$queryBuilder->expr()->eq(	5✔
550	'process_id_completed',	5✔
551	$queryBuilder->createNamedParameter($process->getProcessId())	5✔
552	),	5✔
553	$queryBuilder->expr()->gt('exec_time', 0)	5✔
554	)	5✔
555	->setMaxResults(1)	5✔
556	->addOrderBy($orderByField, $orderBySorting)	5✔
557	->executeQuery()->fetchAssociative();	5✔
558
559	return $first ?: [];	5✔
560	}
561	}

tomasnorre / crawler / 11463038090

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous