<?php
/**
* SeekQuarry/Yioop --
* Open Source Pure PHP Search Engine, Crawler, and Indexer
*
* Copyright (C) 2009 - 2026 Chris Pollett chris@pollett.org
*
* LICENSE:
*
* This program 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.
*
* This program 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.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
* END LICENSE
*
* @author Chris Pollett chris@pollett.org
* @license https://www.gnu.org/licenses/ GPL3
* @link https://www.seekquarry.com/
* @copyright 2009 - 2026
* @filesource
*/
namespace seekquarry\yioop\library\storage_formats;
use seekquarry\yioop\library as L;
use seekquarry\yioop\configs as C;
use seekquarry\yioop\library\UrlParser;
use seekquarry\yioop\library\data_structures\BloomFilterFile;
use seekquarry\yioop\library\data_structures\UndoLog;
use seekquarry\yioop\library\language_processing\PhraseParser;
/**
* Used for crawlLog, crawlHash, and garbageCollect
*/
require_once __DIR__ . '/../Utility.php';
/**
* FeedDocumentBundle is the index of news feed items, kept in the
* cache folder named IndexDataFeed, that FeedsUpdateJob adds items to
* and searches of news read. It is an IndexDocumentBundle with two
* Bloom filters, files that answer quickly whether an item's id was seen
* before, so an item is added once.
*
* @author Chris Pollett
*/
class FeedDocumentBundle extends IndexDocumentBundle
{
/**
* OLD_ITEM_TIME is how long, in seconds, a feed item is kept before it
* counts as too old to index.
*/
const OLD_ITEM_TIME = 4 * C\ONE_WEEK;
/**
* $db is the database connection the bundle writes the counts of
* trending terms to.
* @var DatasourceManager
*/
public $db;
/**
* $feeds holds the settings of the news feeds whose items are stored
* in this bundle, such as each feed's language and category.
* FeedsUpdateJob sets it before the bundle builds a partition.
* @var array
*/
public $feeds;
/**
* $filter_a is the Bloom filter that holds the ids of the feed items
* stored in this bundle, and is the one asked whether an item is new.
* Once it holds more than half of URL_FILTER_SIZE ids, each id is also
* added to $filter_b. Once it holds more than URL_FILTER_SIZE ids,
* $filter_b takes its place, so the filter never grows too full to
* answer well.
* @var BloomFilterFile
*/
public $filter_a;
/**
* $filter_b is the second Bloom filter of feed item ids. It is started
* once $filter_a is more than half full, gets every id added after
* that, and replaces $filter_a once $filter_a is full. It is null
* until it is started.
* @var BloomFilterFile
*/
public $filter_b;
/**
* __construct opens the feed bundle in $dir_name, making it if it is
* not there, and loads its Bloom filters from their files.
* FeedsUpdateJob::getFeedBundle calls it each time MediaUpdater starts.
*
* @param string $dir_name folder name to store this bundle
* @param object $db datasource handle used by the feed bundle to
* read/write its associated FEED_ITEM table rows
* @param bool $read_only_archive whether to open archive only for reading
* or reading and writing
* @param string $description a text name/serialized info about this
* IndexDocumentBundle
* @param int $num_docs_per_partition the number of pages to be stored
* in a single shard
*/
public function __construct($dir_name, $db, $read_only_archive = true,
$description = null, $num_docs_per_partition =
C\NUM_DOCS_PER_PARTITION)
{
parent::__construct($dir_name, $read_only_archive, $description,
$num_docs_per_partition);
$this->feeds = []; // set by FeedsUpdateJob
$this->db = $db;
if (file_exists($dir_name . "/filter_a.ftr")) {
$this->filter_a = BloomFilterFile::load($dir_name .
"/filter_a.ftr");
$this->filter_a->filename = $dir_name . "/filter_a.ftr";
} else {
$this->filter_a = new BloomFilterFile($dir_name . "/filter_a.ftr",
C\URL_FILTER_SIZE);
set_error_handler(null);
@chmod($dir_name . "/filter_a.ftr", 0755);
restore_error_handler();
}
if (file_exists($dir_name . "/filter_b.ftr")) {
$this->filter_b = BloomFilterFile::load($dir_name .
"/filter_b.ftr");
$this->filter_b->filename = $dir_name . "/filter_b.ftr";
} else {
$this->filter_b = null;
}
}
/**
* addPagesAndSeenKeys adds the pages of feed items to the bundle and
* adds each item's id to the Bloom filters, so the item is not added
* again. FeedsUpdateJob::addFoundItemsPartition calls it.
*
* @param array $pages array of feed items
* @param int $visited_urls_count number of feed items
* @return bool whether or not succeeded in adding pages
*/
public function addPagesAndSeenKeys($pages, $visited_urls_count)
{
foreach ($pages as $page) {
$key = $page[self::SUMMARY][self::HASH];
$this->addFilters($key);
}
return parent::addPages($pages, $visited_urls_count);
}
/**
* addFilters adds the id of a feed item to the Bloom filters. The id
* always goes to $filter_a. Once $filter_a is more than half full,
* $filter_b is started and gets the id too. Once $filter_a is full,
* $filter_b's file replaces $filter_a's and $filter_b becomes
* $filter_a.
*
* @param string $key the id of a feed item
*/
public function addFilters($key)
{
if ($this->filter_a->count > C\URL_FILTER_SIZE/2 &&
!$this->filter_b) {
if (file_exists($this->dir_name . "/filter_b.ftr")) {
$this->filter_b = BloomFilterFile::load($this->dir_name .
"/filter_b.ftr");
$this->filter_b->filename = $this->dir_name . "/filter_b.ftr";
} else {
$this->filter_b = new BloomFilterFile(
$this->dir_name . "/filter_b.ftr", C\URL_FILTER_SIZE);
}
}
if ($this->filter_a->count > C\URL_FILTER_SIZE && $this->filter_b) {
$this->filter_b->save();
rename($this->dir_name . "/filter_b.ftr",
$this->dir_name . "/filter_a.ftr");
$this->filter_a = $this->filter_b;
$this->filter_a->filename = $this->dir_name . "/filter_a.ftr";
$this->filter_b = null;
}
$this->filter_a->add($key);
$this->filter_b?->add($key);
}
/**
* buildInvertedIndexPartition builds the inverted index of a partition
* of feed items: the word lists of each item's title and description,
* its meta words, and the trending term counts. The four files of the
* index are saved as one step of the undo log, so a build ended part
* way is undone the next time the bundle is opened. forceSave calls
* it.
*
* @param int $partition bundle partition to build inverted index for
* @param string $taking_too_long_touch name of file to touch if building
* inverted index takes too long (whether SCHEDULES_DIR/ .
* "/{$this->channel}-" . CrawmConstants::crawl_status_file
* has been recently modified) is used in crawling to see if have run out
* of new data and the crawl can stopped.
* @param bool $just_stats whether to just compute stats on the inverted
* or to actually save the results
* @return mixed false if the partition has no items, an array with
* NUM_DOCS, NUM_LINKS and TERM_STATISTICS if $just_stats, and
* nothing otherwise
*/
public function buildInvertedIndexPartition($partition = -1,
$taking_too_long_touch = null, $just_stats = false)
{
$age = self::OLD_ITEM_TIME;
$pre_feeds = $this->feeds ?? [];
if ($partition < 0) {
$partition = $this->documents->parameters["SAVE_PARTITION"];
}
$dockeys_filename = $this->documents->getPartitionIndex($partition);
if(!file_exists($dockeys_filename)) {
L\crawlLog("----.. No new items to index returning.");
return false;
}
L\crawlLog("----.. Loading guids of items to index.");
$doc_index = $this->documents->loadPartitionIndex($partition, true);
$doc_ids = array_keys($doc_index);
$num_ids = count($doc_ids);
L\crawlLog("----.. Loaded $num_ids items in partition.");
$feeds = [];
foreach ($pre_feeds as $pre_feed) {
if (!isset($pre_feed['NAME'])) {
continue;
}
$feeds[$pre_feed['NAME']] = $pre_feed;
}
$time = time();
$num_sites = 0;
$completed = true;
L\crawlLog("----.. Creating inverted index of items.");
$i = 0;
$term_counts = [];
$seen_url_count = 0;
$base_folder = $this->getPartitionBaseFolder($partition);
if (!file_exists($base_folder)) {
mkdir($base_folder);
}
$doc_map_tools = $this->doc_map_tools;
$postings_tools = $this->postings_tools;
$last_entries_tools = $this->last_entries_tools;
$doc_map_filename = $base_folder . "/" . self::DOC_MAP_FILENAME;
$postings_filename = $base_folder . "/" . self::POSTINGS_FILENAME;
$positions_filename = $base_folder . "/" . self::POSITIONS_FILENAME;
$last_entries_filename = $base_folder . "/" .
self::LAST_ENTRIES_FILENAME;
$this->doc_map = "";
$this->doc_map_counter = 0;
$this->postings = [];
$this->posting_counts = [];
$this->positions = "";
$this->last_entries = [];
$this->last_rows = [];
for ($i = 0; $i < $num_ids; $i++) {
$doc_id = $doc_ids[$i];
$item = $this->getSummary($doc_id, $partition);
if (L\crawlTimeoutLog(
"----..have added %s items to new index.", $i) &&
$taking_too_long_touch) {
if (file_exists($taking_too_long_touch)) {
touch($taking_too_long_touch);
}
}
if (!isset($item[self::SOURCE_NAME])) {
continue;
}
$source_name = $item[self::SOURCE_NAME];
if (isset($feeds[$source_name])) {
$lang = $feeds[$source_name]['LANGUAGE'];
$media_category = $feeds[$source_name]['CATEGORY'];
$source_stop_regex =
$feeds[$source_name]['TRENDING_STOP_REGEX'];
} else {
$lang = C\p('DEFAULT_LOCALE');
$media_category = "news";
$source_stop_regex = "";
}
/* r6t was chosen as short enough not to be
changed by chargramming, but rare enough
that can be used as a useful splitter
*/
$phrase_string = $item[self::TITLE] . " r6t " .
$item[self::DESCRIPTION];
$word_and_qa_lists = PhraseParser::extractPhrasesInLists(
$phrase_string, $lang);
$meta_ids = $this->calculateMetas($lang, $item[self::PUBDATE],
$source_name, $item[self::HASH], $media_category);
$len = strlen($phrase_string);
$word_list = $word_and_qa_lists["WORD_LIST"];
if (PhraseParser::computeSafeSearchScore($phrase_string,
$item[self::URL]) < PhraseParser::SAFE_PHRASE_THRESHOLD) {
$meta_ids[] = "safe:true";
$meta_ids[] = "safe:all";
} else {
$meta_ids[] = "safe:false";
$meta_ids[] = "safe:all";
}
$title_length = 0;
if (!empty($word_list["r6t"][0])) {
$title_length = $word_list["r6t"][0] + 1;
unset($word_list["r6t"]);
}
$num_words = 0;
foreach($word_list as $word => $position_list)
{
$num_words += count($position_list);
}
$this->addScoresDocMap($doc_id, $num_words,
intval($item[self::PUBDATE]), 0, $title_length, $title_length,
[], []);
$this->addTermPostingLists(0, $word_list, $meta_ids,
$this->doc_map_counter);
$this->doc_map_counter++;
$this->updateTrendingTermCounts($term_counts, $phrase_string,
$word_list, $media_category, $source_name, $lang,
$item[self::PUBDATE], $source_stop_regex);
}
$this->joinPostings();
if ($just_stats) {
$term_stats = [];
foreach ($this->postings as $term => $postings) {
$posting_records = $postings_tools->unpack($postings);
$term_stats[$term] = count($posting_records);
}
$statistics = [
"NUM_DOCS" => count($this->doc_map),
"NUM_LINKS" => 0,
"TERM_STATISTICS" => $term_stats
];
return $statistics;
}
$undo_log = new UndoLog($this->dir_name);
$step_id = $undo_log->begin("build of feed partition $partition", [],
[$doc_map_filename, $postings_filename, $last_entries_filename,
$positions_filename]);
$doc_map_tools->save($doc_map_filename, $this->doc_map, false);
/* The dictionary takes a partition's terms in the order the
postings file holds them and finds a term by halving, so the
terms are saved in order, as IndexDocumentBundle saves them. */
ksort($this->postings);
$postings_tools->save($postings_filename, $this->postings, false);
$this->packLastEntries();
$last_entries_tools->save($last_entries_filename, $this->last_entries,
false);
L\filePutContentsWhole($positions_filename, $this->positions, false);
$undo_log->finish($step_id, [$doc_map_filename, $postings_filename,
$last_entries_filename, $positions_filename]);
unset($term_counts['seen']);
$this->addTermCountsTrendingTable($term_counts);
}
/**
* computeDocId works out the document id of a feed item from its url
* and its id, in the same form FeedsUpdateJob gives the items it adds:
* a hash of the url, the item id, a byte marking a feed item, and a
* hash of the url's host.
*
* @param array $site the feed item, with its URL and HASH fields
* @return string the item's document id
*/
public static function computeDocId($site)
{
$raw_guid = L\unbase64Hash($site[self::HASH]);
// chr(24) refers to 'feed' doc type
$doc_id = L\crawlHash($site[self::URL], true) .
$raw_guid . chr(24) . substr(L\crawlHash(
UrlParser::getHost($site[self::URL]) . "/", true), 1);
return $doc_id;
}
/**
* calculateMetas works out the meta words of a feed item, for example
* media:news and date:2026-10-05, which let a search pick items by
* kind, source, date and language. buildInvertedIndexPartition calls
* it for each item.
*
* @param string $lang the locale_tag of the feed item
* @param int $pubdate UNIX timestamp publication date of item
* @param string $source_name the name of the feed
* @param string $guid the guid of the item
* @param string $media_category determines what media: metas to inject.
* Default is news.
*
* @return array $meta_ids meta words found
*/
public function calculateMetas($lang, $pubdate, $source_name, $guid,
$media_category = "news")
{
$meta_ids = ["media:all", "media:$media_category",
"media:$media_category:" . urlencode( mb_strtolower($source_name)),
"guid:" . strtolower($guid)];
$meta_ids[] = 'date:all';
$meta_ids[] = 'date:' . date('Y', $pubdate);
$meta_ids[] = 'date:' . date('Y-m', $pubdate);
$meta_ids[] = 'date:' . date('Y-m-d', $pubdate);
$meta_ids[] = 'date:' . date('Y-m-d-H', $pubdate);
$meta_ids[] = 'date:' . date('Y-m-d-H-i', $pubdate);
$meta_ids[] = 'date:' . date('Y-m-d-H-i-s', $pubdate);
$meta_ids[] = 'lang:all';
if ($lang != "") {
$lang_parts = explode("-", $lang);
$meta_ids[] = 'lang:' . $lang_parts[0];
if (isset($lang_parts[1])) {
$meta_ids[] = 'lang:' . $lang;
}
}
return $meta_ids;
}
/**
* contains says if $filter_a holds the id of a feed item, that is if
* the item is in the bundle already. FeedsUpdateJob asks it before it
* adds an item.
*
* @param string $key the id of the feed item
* @return bool true if it is in the archive, false otherwise
*/
public function contains($key)
{
return $this->filter_a->contains($key);
}
/**
* forceSave builds the inverted index of the partition being filled
* and saves the Bloom filters, opening their files to every user.
* FeedsUpdateJob calls it when the dictionary has no partition to
* add.
*/
public function forceSave()
{
$this->buildInvertedIndexPartition();
$this->filter_a->save();
L\setWorldPermissions($this->dir_name . "/filter_a.ftr");
if ($this->filter_b) {
$this->filter_b->save();
L\setWorldPermissions($this->dir_name . "/filter_b.ftr");
}
}
/**
* updateTrendingTermCounts adds the words of one feed item to the
* counts of trending terms, by language and category, leaving out
* the source's name and the words its stop pattern matches.
* buildInvertedIndexPartition calls it for each item.
*
* @param array &$term_counts lang => [term => occurrences]
* @param string $source_phrase original non-stemmed phrase from feed
* item to adjust $term_counts with. Used to remember non-stemmed
* terms. We assume we have already extracted position lists from
* @param array $word_or_phrase_list associate array of
* stemmed_word_or_phrase => positions in feed item of where occurs
* @param string $media_category of feed source the item case from. We
* trending counts grouped by media category
* @param string $source_name of feed source the item case from. We exclude
* from counts the name of the feed source
* @param string $lang locale_tag for this feed item
* @param int $pubdate timestamp when string was published (used in
* weighting)
* @param string $source_stop_regex a regex to remove terms which occur
* frequently for this particular source
*/
public function updateTrendingTermCounts(&$term_counts, $source_phrase,
$word_or_phrase_list, $media_category, $source_name, $lang, $pubdate,
$source_stop_regex = "")
{
$time = time();
if ($time - $pubdate > C\ONE_DAY) {
return;
}
$time_weight = max(ceil(($time - $pubdate)/C\ONE_HOUR), 1);
$tokenizer = PhraseParser::getTokenizer($lang);
$stop_words = (empty($tokenizer::$stop_words)) ? [] :
$tokenizer::$stop_words;
$stop_words[] = "tztzlzngth";
$composites = [];
foreach ($word_or_phrase_list as $term => $positions) {
if (str_contains($term, "-")) {
$sub_terms = explode("-", $term);
$composites += $sub_terms;
}
}
foreach($word_or_phrase_list as $term => $positions) {
if ((!empty($source_stop_regex) &&
preg_match($source_stop_regex, $term)) ||
!is_string($term) || $term == "") {
continue;
}
$occurrences = count($positions);
$weight = substr_count($term, "-");
if (!empty($term_counts['seen'][$term])) {
$term = $term_counts['seen'][$term];
if (empty($term_counts[$lang][$media_category][$term])) {
if (empty($term_counts[$lang])) {
$term_counts[$lang] = [];
}
if (empty($term_counts[$lang][$media_category])) {
$term_counts[$lang][$media_category] = [];
}
$term_counts[$lang][$media_category][$term] = 0;
}
$term_counts[$lang][$media_category][$term] +=
($weight * $occurrences) / $time_weight;
} else if (!str_contains($term, " ") &&
!in_array($term, $composites) &&
!in_array($term[0], ["#", "_", "-"] ) &&
!in_array($term, $stop_words) &&
!str_starts_with($term, "http")) {
$original_term = $term;
if (preg_match("/\b" . preg_quote($term, "/") .
'[\w|\-]*\b/ui', $source_phrase, $match)) {
$term = $match[0];
if (!empty($source_stop_regex) &&
preg_match($source_stop_regex, $term)) {
continue;
}
} else {
$term = str_replace("-", " ", $term);
if( mb_stristr($source_name, $term) !== false) {
continue;
}
if (substr($term, -6) == "_pos_s") {
$term = substr($term, 0, -6);
$term_pattern = $term .
"\s*(\'|\&apos\;)\s*";
} else {
$term_pattern = preg_quote($term, "/");
}
if (preg_match("/\b$term_pattern".'[\w|\-]*\b/ui',
$source_phrase, $match)) {
$term = $match[0];
} else {
$trunc_term = mb_substr($term, 0, -1);
$trunc_original = mb_substr($original_term, 0, -1);
if (preg_match("/\b" . preg_quote($trunc_term, "/") .
'[\w|\-]*\b/ui', $source_phrase, $match)) {
$term = $match[0];
} else if (preg_match("/\b" .
preg_quote($trunc_original, "/") .
'[\w|\-]*\b/ui', $source_phrase, $match)) {
$term = $match[0];
} else {
continue;
}
}
if (!empty($source_stop_regex) &&
preg_match($source_stop_regex, $term)) {
continue;
}
}
$lower_term = mb_strtolower($term);
$lower_parts = preg_split("/\s+|\-/ui", $lower_term);
$uncommon = false;
foreach ($lower_parts as $lower_part) {
if (!empty($lower_part) &&
!in_array($lower_part, $stop_words)) {
$uncommon = true;
break;
}
}
if ($uncommon) {
$term = preg_replace("/\-/", " ", $lower_term);
$term = mb_convert_case($term, MB_CASE_TITLE, 'UTF-8');
if (empty($term_counts[$lang][$media_category][$term])) {
if (empty($term_counts[$lang])) {
$term_counts[$lang] = [];
}
if (empty($term_counts[$lang][$media_category])) {
$term_counts[$lang][$media_category] = [];
}
$term_counts[$lang][$media_category][$term] = 0;
}
$term_counts['seen'][$original_term] = $term;
if (!empty($term_counts['seen'][$lower_term])) {
$term = $term_counts['seen'][$lower_term];
if (empty($term_counts[$lang][$media_category][$term])){
$term_counts[$lang][$media_category][$term] = 0;
}
}
$term_counts['seen'][$lower_term] = $term;
$term_counts[$lang][$media_category][$term] +=
($weight * $occurrences) / $time_weight;
}
}
}
}
/**
* addTermCountsTrendingTable writes the trending term counts of a
* partition's build to the TRENDING_TERM table, then sums the hourly
* counts into daily, weekly, monthly and yearly ones and deletes the
* rows too old to keep. buildInvertedIndexPartition calls it once its
* files are saved.
*
* @param array $term_counts for the most recent update of the
* feed index, it should be an array [$lang => [$term => $occurrences]]
* for the top NUM_TRENDING terms per language
*/
public function addTermCountsTrendingTable($term_counts)
{
$db = $this->db;
$time = time();
$update_intervals = [
C\ONE_HOUR => C\ONE_DAY,
C\ONE_DAY => C\ONE_WEEK,
C\ONE_WEEK => C\ONE_MONTH,
C\ONE_MONTH => C\ONE_YEAR
];
$reset_interval_sql = "DELETE FROM TRENDING_TERM WHERE " .
"UPDATE_PERIOD = ? AND LANGUAGE = ? AND ".
"CATEGORY = ? AND TIMESTAMP >= ? ";
$interval_sql = "SELECT TERM, SUM(OCCURRENCES) AS OCCURRENCES ".
"FROM TRENDING_TERM WHERE UPDATE_PERIOD = ? AND " .
"TIMESTAMP >= ? AND LANGUAGE = ? AND CATEGORY = ? GROUP BY TERM ".
"ORDER BY OCCURRENCES DESC ".
$db->limitOffset(C\NUM_TRENDING);
$insert_sql = "INSERT INTO TRENDING_TERM (TERM, OCCURRENCES, " .
"UPDATE_PERIOD, TIMESTAMP, LANGUAGE, CATEGORY) VALUES ".
"(?, ?, ?, ?, ?, ?)";
$cull_sql = "DELETE FROM TRENDING_TERM WHERE UPDATE_PERIOD = ? " .
" AND TIMESTAMP < ?";
$have_culled = [];
foreach ($term_counts as $lang => $category_term_occurrences) {
L\crawlLog("Updating Language $lang trending terms");
foreach ($category_term_occurrences as $category =>
$term_occurrences) {
/* Sort the word occurrence list by occurrence
and insert into database.
*/
$num_inserted = 0;
arsort($term_occurrences);
foreach ($term_occurrences as $term => $occurrences) {
$db->execute($insert_sql, [$term, $occurrences, C\ONE_HOUR,
$time, $lang, $category]);
$num_inserted++;
if ($num_inserted >= C\NUM_TRENDING) {
break;
}
}
foreach ($update_intervals as $sub_interval => $interval) {
$interval_start = $time - $interval;
$recent_timestamp = floor($interval_start/$interval) *
$interval;
$db->execute($reset_interval_sql,
[$interval, $lang, $category, $recent_timestamp]);
$result = $db->execute($interval_sql,
[$sub_interval, $recent_timestamp, $lang, $category]);
while ($interval_info = $db->fetchArray($result)) {
$db->execute($insert_sql, [$interval_info['TERM'],
$interval_info['OCCURRENCES'], $interval,
$recent_timestamp, $lang, $category]);
}
if (empty($have_culled[$sub_interval])) {
$db->execute($cull_sql, [$sub_interval,
$recent_timestamp]);
$have_culled[$sub_interval] = true;
}
}
}
}
}
}
- Yioop Repo - Seekquarry