<?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\executables;
use seekquarry\yioop\configs as C;
use seekquarry\yioop\library as L;
use seekquarry\yioop\library\archive_bundle_iterators as A;
use seekquarry\yioop\library\data_structures\BloomFilterFile;
use seekquarry\yioop\library\Classifiers;
use seekquarry\yioop\library\ComputerVision;
use seekquarry\yioop\library\CrawlConstants;
use seekquarry\yioop\library\CrawlDaemon;
use seekquarry\yioop\library\FetchUrl;
use seekquarry\yioop\library\version_control\FetchGitRepositoryUrls;
use seekquarry\yioop\library\PageRuleParser;
use seekquarry\yioop\library\language_processing\PhraseParser;
use seekquarry\yioop\library\processors\PageProcessor;
use seekquarry\yioop\library\processors\ImageProcessor;
use seekquarry\yioop\library\UrlParser;
use seekquarry\yioop\library\storage_formats\IndexDocumentBundle;
use seekquarry\yioop\library\storage_formats\MessagesBundle;
use seekquarry\yioop\library\storage_formats\WebArchiveBundle;
if (!defined("seekquarry\\yioop\\configs\\UNIT_TEST_MODE")) {
if (php_sapi_name() != 'cli' ||
defined("seekquarry\\yioop\\configs\\IS_OWN_WEB_SERVER")) {
echo "BAD REQUEST"; exit();
}
}
/** for L\crawlHash and L\crawlLog and Yioop constants*/
require_once __DIR__ . "/../library/Utility.php";
if (!C\PROFILE) {
echo "Please configure the search engine instance by visiting" .
"its web interface on localhost.\n";
exit();
}
ini_set("memory_limit", C\FETCHER_MEMORY_LIMIT);
/** To guess language based on page encoding */
require_once __DIR__ . "/../library/LocaleFunctions.php";
/*
* We'll set up multi-byte string handling to use UTF-8
*/
mb_internal_encoding("UTF-8");
mb_regex_encoding("UTF-8");
/**
* This class is responsible for fetching web pages for the
* SeekQuarry/Yioop search engine
*
* Fetcher periodically queries the queue server asking for web pages to fetch.
* It gets at most MAX_FETCH_SIZE many web pages from the queue_server in one
* go. It then fetches these pages. Pages are fetched in batches of
* NUM_MULTI_CURL_PAGES many pages. Once downloaded the fetcher sends summaries
* back to the machine on which the queue_server lives. It does this by making a
* request of the web server on that machine and POSTs the data to the
* yioop web app. This data is handled by the FetchController class. The
* summary data can include up to four things: (1) robot.txt data, (2) summaries
* of each web page downloaded in the batch, and (3)a list of future urls to add
* to the to-crawl queue.
*
* @author Chris Pollett
*/
class Fetcher implements CrawlConstants
{
/**
* Most pages and links, counted together, the fetcher holds before it
* sends them to the queue server, so neither process ever has one very
* large upload in memory at once
*/
const SEEN_URLS_BEFORE_SEND = 1000;
/**
* Most pages and links a synthetic crawl's fetcher holds before it
* sends them. A synthetic crawl makes its pages itself at a rate no
* web crawl reaches, so it sends in far larger batches than
* SEEN_URLS_BEFORE_SEND.
*/
const SYNTHETIC_SEEN_URLS_BEFORE_SEND = 20000;
/**
* SLOW_PAGE_SECONDS is how many seconds a processor may spend on one
* page before processFetchPages logs the page's url, so the slowest
* pages of a crawl can be found and fetched again to test against.
*/
const SLOW_PAGE_SECONDS = 1;
/**
* SLOW_PAGES_BEFORE_FAST is how many slow pages, each taking a
* processor over SLOW_PAGE_SECONDS, one host may give a fetcher within
* an hour before the fetcher handles that host's pages with the fast
* variant of their processor for the rest of that hour.
*/
const SLOW_PAGES_BEFORE_FAST = 10;
/**
* SLOW_HOSTS_KEPT is how many hosts $slow_pages holds before
* noteSlowPage drops those whose hour has passed.
*/
const SLOW_HOSTS_KEPT = 1000;
/**
* ANSWER_SCHEDULE_AFTER is how many seconds a web crawl's fetcher may
* hold a schedule before it sends what it has found so far. The queue
* server takes back schedules in the order it sent them. It stops
* making schedules when FETCHER_NUM_WINDOW_SCALE times its number of
* fetchers are out or waiting behind an earlier one. So one fetcher
* slow on a batch of
* pages that take seconds each made every other fetcher wait. The
* first send of a schedule answers it, so the queue server goes on.
* Most schedules are sent back within two minutes anyway.
*/
const ANSWER_SCHEDULE_AFTER = 2 * C\ONE_MINUTE;
/**
* PAGE_IN_PROCESS_NAME begins the name of the file in SCHEDULES_DIR
* that holds the url of the page this fetcher's processor is reading.
* The file is gone once a batch is processed, so a restarted fetcher
* that finds it knows which page stopped the fetcher before.
*/
const PAGE_IN_PROCESS_NAME = "PageInProcess";
/**
* SCHEDULE_LEFT_NAME begins the name of the file in SCHEDULES_DIR that
* holds the urls of this fetcher's schedule not yet sent back, with
* the schedule's sequence number. A restarted fetcher goes on with
* them rather than leaving the queue server to wait for the schedule.
*/
const SCHEDULE_LEFT_NAME = "ScheduleLeft";
/**
* PREFETCH_WAIT_MICROSECONDS is how long finishPrefetch sleeps between
* passes of the next batch's download once nothing else is left to
* do, so that waiting does not keep a core busy.
*/
const PREFETCH_WAIT_MICROSECONDS = 10000;
/**
* SEEN_URL_COMPRESSION_LEVEL is the zlib level each site summary and
* link record sent to a queue server is packed at, from 1, fastest, to
* 9, smallest. Level 1 packs these records about three times as fast
* as the default level 6, for about an eighth more bytes. The queue
* server opens any level the same way.
*/
const SEEN_URL_COMPRESSION_LEVEL = 1;
/**
* RETRY_HTTP_CODES are the HTTP codes after which a page is worth
* fetching again later: a request timeout, Yioop's own code 418 for a
* page that did not download in MAX_DOWNLOAD_ATTEMPTS tries, too many
* requests, and the server errors that pass. A page that got one of
* these, or no answer at all, is sent back to the queue server in
* RETRY_URLS. A 403, a 404 or any other code is final.
*/
const RETRY_HTTP_CODES = [408, 418, 429, 500, 502, 503, 504];
/**
* CHALLENGE_RETRY_CODE is the HTTP code a page is sent back to the
* queue server with when the site answered with a bot-protection
* challenge in place of the page: 503, a server unable to serve the
* page for now, so the page is fetched again later.
*/
const CHALLENGE_RETRY_CODE = 503;
/**
* Reference to a database object. Used since has directory manipulation
* functions
* @var object
*/
public $db;
/**
* Urls or IP address of the web_server used to administer this instance
* of yioop. Used to figure out available queue_servers to contact
* for crawling data
*
* @var array
*/
public $name_server;
/**
* Array of Urls or IP addresses of the queue_servers to get sites to crawl
* from
* @var array
*/
public $queue_servers;
/**
* Index into $queue_servers of the server get schedule from (or last one
* we got the schedule from)
* @var int
*/
public $current_server;
/**
* An associative array of (mimetype => name of processor class to handle)
* pairs.
* @var array
*/
public $page_processors;
/**
* An associative array of (page processor => array of
* indexing plugin name associated with the page processor). It is used
* to determine after a page is processed which plugins'
* pageProcessing($page, $url) method should be called
* @var array
*/
public $plugin_processors;
/**
* Hash used to keep track of whether $plugin_processors info needs to be
* changed
* @var string
*/
public $plugin_hash;
/**
* Says whether the $allowed_sites array is being used or not
* @var bool
*/
public $restrict_sites_by_url;
/**
* List of file extensions supported for the crawl
* @var array
*/
public $indexed_file_types;
/**
* List of all known file extensions including those not used for crawl
* @var array
*/
public $all_file_types;
/**
* Web-sites that crawler can crawl. If used, ONLY these will be crawled
* @var array
*/
public $allowed_sites;
/**
* An array of Bloom filters which if non empty will be used to
* restrict discovered url links. Namely, if the domain of a url is not in
* any of the filters and is not of thhe form of a company level domain
* (cld) or www.cld then it will be pruned. Here a company level domain
* is of the form some_name.tld where tld is a top level domain
* or the form some_name.country_level_tld.tld if the the tld is that of a
* country. So for site.somewhere.jp, somewhere.jp would be the cld; for
* site.somewhere.co.jp, somewhere.co.jp would be the cld.
* @var array
*/
public $domain_filters;
/**
* Web-sites that the crawler must not crawl
* @var array
*/
public $disallowed_sites;
/**
* Lines of the crawl's Site Cookies option, each a domain, a colon and
* the cookies to send it, handed to FetchUrl::setSiteCookies before
* each batch of downloads.
* @var array
*/
public $site_cookies = [];
/**
* Microtime used to look up cache $allowed_sites and $disallowed_sites
* filtering data structures
* @var int
*/
public $allow_disallow_cache_time;
/**
* Holds the parsed page rules which will be applied to document summaries
* before finally storing and indexing them
* @var array
*/
public $page_rule_parser;
/**
* Name of file used to store fetcher statistics for the current crawl
* @var string
*/
public $crawl_stat_filename;
/**
* Fetcher statistics for the current crawl
* @var string
*/
public $crawl_stat_info;
/**
* Timestamp of the current crawl
* @var int
*/
public $crawl_time;
/**
* The last time the name server was checked for a crawl time
* @var int
*/
public $check_crawl_time;
/**
* For a web crawl only the number of web pages to download in one
* go.
* @var int
*/
public $num_multi_curl;
/**
* Contains the list of web pages to crawl from a queue_server
* @var array
*/
public $to_crawl;
/**
* Contains the list of web pages to crawl that failed on first attempt
* (we give them one more try before bailing on them)
* @var array
*/
public $to_crawl_again;
/**
* $last_try_urls holds, as keys with the value true, the urls of the
* current schedule that the Scheduler is sending out for the last
* time after failed fetches, as QueueServer::lastTryUrls lists them.
* A site's home page that answers one of these with a bot-protection
* challenge is indexed by the words of its url, since no later fetch
* will come. checkScheduler sets it from each schedule.
* @var array
*/
public $last_try_urls = [];
/**
* Number of attempts to download urls in current fetch batch
* @var int
*/
public $num_download_attempts;
/**
* $prefetch is the Fiber downloading the next batch of urls while the
* current batch is processed, or null when there is none. Its
* getPages call hands control back on each pass of its wait for
* transfers, so its downloads go on between pages. A batch taken
* from to_crawl this way is held schedule, as to_crawl is.
* @var \Fiber
*/
public $prefetch = null;
/**
* Summary information for visited sites that the fetcher hasn't sent to
* a queue_server yet
* @var array
*/
public $found_sites;
/**
* $pages_compressed is how many sites at the start of
* $found_sites[self::SEEN_URLS] have had their page gzip encoded by
* compressWaitingPage. compressAndUnsetSeenUrls sets it back to 0 when
* it sends the sites.
* @var int
*/
public $pages_compressed = 0;
/**
* Timestamp from a queue_server of the current schedule of sites to
* download. This is sent back to the server once this schedule is completed
* to help the queue server implement crawl-delay if needed.
* @var int
*/
public $schedule_time;
/**
* $schedule_received_time is when, in seconds since the epoch, this
* fetcher was given its current schedule. updateFoundSites compares
* it with ANSWER_SCHEDULE_AFTER.
* @var int
*/
public $schedule_received_time = 0;
/**
* $answered_sequence_number is the sequence number of the last
* schedule whose results this fetcher sent to the queue server that
* gave it. A schedule is answered once this equals $sequence_number.
* @var int
*/
public $answered_sequence_number = 0;
/**
* $schedule_resend_time is when, in seconds since the epoch, the queue
* server will send this fetcher's current schedule to another fetcher
* if it has not been answered, as MessagesBundle::scheduleTimeout
* reckons it. resumeScheduleLeft goes on with a schedule only before
* then, so its pages are not fetched twice.
* @var int
*/
public $schedule_resend_time = 0;
/**
* Seconds the processors spent, and how many pages they handled, for
* each kind of document, by the ending of its url, since the last
* time these were sent to a queue server. The queue server puts urls
* of a kind that is slow to process further back in its queue.
* @var array
*/
public $process_times = [];
/**
* Channel that queue server listens to messages for
* @var int
*/
public $channel;
/**
* Stores the name of the ordering used to crawl pages. This is used in a
* switch/case when computing weights of urls to be crawled before sending
* these new urls back to a queue_server.
* @var string
*/
public $crawl_order;
/**
* Maximum depth fetcher should extract need seed urls to
* @var int
*/
public $max_depth;
/**
* Stores the name of the summarizer used for crawling.
* Possible values are self::BASIC, self::GRAPH_BASED_SUMMARIZER,
* self::CENTROID_SUMMARIZER and self::CENTROID_WEIGHTED_SUMMARIZER
* @var string
*/
public $summarizer_option;
/**
* $slow_pages holds, for each host that gave this fetcher a slow
* page, the time its hour began and how many slow pages it gave in
* that hour. noteSlowPage adds to it and usesFastProcessor reads it.
* @var array
*/
public $slow_pages = [];
/**
* Indicates the kind of crawl being performed: self::WEB_CRAWL indicates
* a new crawl of the web; self::ARCHIVE_CRAWL indicates a crawl of an
* existing web archive
* @var string
*/
public $crawl_type;
/**
* The shape of the links on a synthetic crawl's pages, used only when
* crawl_type is a synthetic crawl.
* @var string
*/
public $graph_type;
/**
* How many links a synthetic crawl's pages carry.
* @var int
*/
public $out_degree;
/**
* The probability that a link on a synthetic page goes to another
* domain rather than a deeper path on the same one.
* @var float
*/
public $domain_link_probability;
/**
* The exponent of a synthetic power law web.
* @var float
*/
public $power_law_exponent;
/**
* The exponent of how often a synthetic word appears on a page.
* @var float
*/
public $term_exponent;
/**
* The average number of words on a synthetic page.
* @var int
*/
public $doc_length;
/**
* The spread of the number of words on a synthetic page.
* @var int
*/
public $length_spread;
/**
* The probability that a link on a synthetic page goes to an image
* rather than a page.
* @var float
*/
public $image_link_probability;
/**
* The number a synthetic web is built from, so a run repeats exactly.
* @var int
*/
public $synthetic_seed;
/**
* For an archive crawl, holds the name of the type of archive being
* iterated over (this is the class name of the iterator, without the word
* 'Iterator')
* @var string
*/
public $arc_type;
/**
* For a non-web archive crawl, holds the path to the directory that
* contains the archive files and their description (web archives have a
* different structure and are already distributed across machines and
* fetchers)
* @var string
*/
public $arc_dir;
/**
* If an web archive crawl (i.e. a re-crawl) is active then this field
* holds the iterator object used to iterate over the archive
* @var object
*/
public $archive_iterator;
/**
* Time, in seconds since the epoch, of the last time an archive crawl
* sent its pages to a queue server from checkArchiveScheduler, or 0
* when it has not sent since the crawl began
* @var int
*/
public $archive_send_time = 0;
/**
* $send_after is how many urls a web crawl's fetcher downloads before
* it sends what it found to a queue server. It starts at
* NUM_MULTI_CURL_PAGES when a crawl begins, so the first pages reach
* the index and the crawl's status after the first download, and it
* doubles after each send until it reaches SEEN_URLS_BEFORE_SEND, or
* SYNTHETIC_SEEN_URLS_BEFORE_SEND for a synthetic crawl.
* @var int
*/
public $send_after = C\NUM_MULTI_CURL_PAGES;
/**
* $fetched_since_send counts the urls downloaded since the last send
* to a queue server; updateFoundSites compares it with $send_after.
* @var int
*/
public $fetched_since_send = 0;
/**
* Keeps track of whether during the recrawl we should notify a
* queue_server scheduler about our progress in mini-indexing documents
* in the archive
* @var bool
*/
public $recrawl_check_scheduler;
/**
* If the crawl_type is self::ARCHIVE_CRAWL, then crawl_index is the
* timestamp of the existing archive to crawl
* @var string
*/
public $crawl_index;
/**
* Whether to cache pages or just the summaries
* @var bool
*/
public $cache_pages;
/**
* $schedules_folder is the folder the fetcher keeps its schedules,
* messages and the files it saves between runs in, the work
* directory's schedules folder. FetcherTest gives its fetchers one in
* the tests folder.
* @var string
*/
public $schedules_folder = C\SCHEDULES_DIR;
/**
* Which fetcher instance we are (if fetcher run as a job and more that one)
* @var string
*/
public $which_fetcher;
/**
* $temp_folder is the folder of temporary files this fetcher alone
* writes to, such as the records curl keeps of each download. It is
* named by the fetcher's number and channel, since two fetchers with
* one number may run on different channels at once. loop sets it.
* @var string
*/
public $temp_folder;
/**
* Maximum number of bytes to download of a webpage
* @var int
*/
public $page_range_request;
/**
* Max number of chars to extract for description from a page to index.
* Only words in the description are indexed.
* @var int
*/
public $max_description_len;
/**
* Maximum number of urls to extract from a single document
* @var int
*/
public $max_links_to_extract;
/**
* An array to keep track of hosts which have had a lot of http errors
* @var array
*/
public $hosts_with_errors;
/**
* When processing recrawl data this says to assume the data has
* already had its inks extracted into a field and so this doesn't
* have to be done in a separate step
*
* @var bool
*/
public $no_process_links;
/**
* $post_sizes holds, for each queue server url, the sizes in bytes
* this fetcher uses to split its uploads into parts. SIZE is the next
* part's size, LIMIT the largest post the server says it takes, GOOD
* the largest size that went through, and BAD the smallest size that
* failed twice in a row, or 0. The next size lies between GOOD and BAD,
* so the fetcher settles on the largest size that goes through.
*
* @var array
*/
public $post_sizes = [];
/**
* Fetcher must wait at least this long between multi-curl requests.
* The value below is dynamically determined but is at least as large
* as MINIMUM_FETCH_LOOP_TIME
* @var int
*/
public $minimum_fetch_loop_time;
/**
* $downloads_per_hour is how many pages this fetcher downloaded in the
* last hour or two, the number its "Downloads per hours" log line
* gives. updateScheduler sends it to a queue server with each upload of
* data, and checkCrawlTime sends it to the name server. Each queue
* server puts the rates of the fetchers it heard from lately in its
* crawl status, and the crawl speed history sums them over the
* fetchers.
* @var float
*/
public $downloads_per_hour = 0;
/**
* $rate_windows holds, for each per-hour figure ratePerHour gives, the
* running total two hours before, the running total one hour before,
* the time the window starts and the time it last moved, keyed by the
* figure's name.
* @var array
*/
public $rate_windows = [];
/**
* $rate_start is the time this fetcher was made. A per-hour figure's
* first window starts then, so the first rounds are divided by the
* time since the fetcher started.
* @var int
*/
public $rate_start;
/**
* Contains which classifiers to use for the current crawl
* Classifiers can be used to label web documents with a meta word
* if the classifiers threshold is met
* @var array
*/
public $active_classifiers;
/**
* Page prcoessors used by this fetcher
*
* @var array
*/
public $processors;
/**
* Contains an array of scrapers used to extract the import content
* from particular kind of HTML pages, for example, pages generated by
* a particular content management system.
* @var array
*/
public $scrapers;
/**
* Contains which classifiers to use for the current crawl
* that are being used to rank web documents. The score that the
* classifier gives to a document is used for this ranking purposes
* @var array
*/
public $active_rankers;
/**
* To keep track of total number of Git internal urls
*
* @var int
*/
public $total_git_urls;
/**
* To store all the internal git urls fetched
*
* @var array
*/
public $all_git_urls;
/**
* To map programming languages with their extensions
*
* @var array
*/
public $programming_language_extension;
/**
* If this is not null and a .onion url is detected then this url will
* used as a proxy server to download the .onion url
* @var string
*/
public $tor_proxy;
/**
* an array of proxy servers to use rather than to directly download web
* pages from the current machine. If is the empty array, then we just
* directly download from the current machine
*
* @var array
*/
public $proxy_servers;
/**
* If a crawl quiescent period is being used with the crawl, then
* this sproperty will be positive and indicate the number of seconds
* duration for the quiescent period.
* @var string
*/
public $sleep_duration;
/**
* If a crawl quiescent period is being used with the crawl, then
* this stores the time of day at which that period starts
* @var string
*/
public $sleep_start;
/**
* One of a fixed set of values which are used to control to what extent
* Yioop follows robots.txt files: ALWAYS_FOLLOW_ROBOTS,
* ALLOW_LANDING_ROBOTS, IGNORE_ROBOTS
* @var int
*/
public $robots_txt;
/**
* Holds the value of a debug message that might have been sent from
* the command line during the current execution of loop();
* @var string
*/
public $debug;
/**
* Holds the sequence number of the current schedule received from
* queue server
* @var int
*/
public $sequence_number;
/**
* DEFAULT_POST_MAX_SIZE is the size in bytes of each part of the first
* upload to a queue server. The server's answer then says the largest
* post it takes, and later uploads use that.
*/
const DEFAULT_POST_MAX_SIZE = 2000000;
/**
* MIN_POST_SIZE is the smallest upload part, in bytes, a fetcher tries
* when posts of a larger size keep failing.
*/
const MIN_POST_SIZE = 500000;
/**
* POST_SIZE_STEP is in bytes. When the largest part size that went
* through and the smallest that failed are closer than this, the fetcher
* keeps the size that went through and stops trying others.
*/
const POST_SIZE_STEP = 250000;
/**
* POST_OTHER_FIELDS_LEN is the number of bytes of each upload post kept
* for the fields sent beside the part, such as its hash and number.
*/
const POST_OTHER_FIELDS_LEN = 10240;
/**
* constant indicating Git repository
*/
const REPOSITORY_GIT = 'git';
/**
* constant indicating Git repository
*/
const GIT_URL_CONTINUE = '@@@@';
/**
* An indicator to tell no actions to be taken
*/
const INDICATOR_NONE = 'none';
/**
* URLS_PER_HOST_LOOKUP says that one host in this many urls of a
* download pass may be a host whose address is not yet known. A Domain
* Name System (DNS) lookup takes longer than fetching a page whose
* address is already known, so a pass made up of lookups would sit
* waiting on the resolver. The urls of one host wait on the same
* name, so hosts are counted, not urls.
*/
const URLS_PER_HOST_LOOKUP = 3;
/**
* LEAST_HOST_LOOKUPS_A_PASS is the fewest hosts needing an address
* lookup that one download pass will take. Without this floor a fetch
* batch holding only such hosts would never move.
*/
const LEAST_HOST_LOOKUPS_A_PASS = 2;
/**
* A indicator to represent next position after the access code in Git
* tree object
*/
const HEX_NULL_CHARACTER = "\x00";
/**
* Domain Filter Glob pattern for Bloom filters used to specify allowed
* domains. (Only specifies if some filter file of this type exists)
*/
const DOMAIN_FILTER_GLOB = C\DATA_DIR . "/domain_filters/*.ftr";
/**
* Sets up the field variables so that crawling can begin
*/
public function __construct()
{
$this->rate_start = time();
$this->processors = [];
PageProcessor::initializeIndexedFileTypes($this->processors);
$db_class = C\NS_DATASOURCES . ucfirst(C\p('DBMS')) . "Manager";
$this->db = new $db_class();
// initially same only one queueserver and is same as name server
$this->name_server = C\p('NAME_SERVER');
$this->queue_servers = [C\p('NAME_SERVER')];
$this->current_server = 0;
$this->page_processors = PageProcessor::$mime_processor;
$this->indexed_file_types = PageProcessor::$indexed_file_types;
$this->all_file_types = PageProcessor::$indexed_file_types;
$this->scrapers = [];
$this->plugin_hash = "";
$this->restrict_sites_by_url = false;
$this->allowed_sites = [];
$this->disallowed_sites = [];
$this->domain_filters = [];
$this->allow_disallow_cache_time = microtime(true);
$this->page_rule_parser = null;
$this->hosts_with_errors = [];
$this->crawl_stat_filename = "";
$this->crawl_stat_info = [];
$this->crawl_time = null;
$this->channel = 0;
$this->sleep_start = "00:00";
$this->sleep_duration = -1;
$this->check_crawl_time = null;
$this->schedule_time = null;
$this->crawl_type = self::WEB_CRAWL;
$this->crawl_index = null;
$this->recrawl_check_scheduler = false;
$this->to_crawl = [];
$this->to_crawl_again = [];
$this->num_download_attempts = 0;
$this->found_sites = [];
$this->found_sites[self::CACHE_PAGE_VALIDATION_DATA] = [];
$this->page_range_request = C\PAGE_RANGE_REQUEST;
$this->max_description_len = C\MAX_DESCRIPTION_LEN;
$this->max_links_to_extract = C\MAX_LINKS_TO_EXTRACT;
$this->which_fetcher = false;
$this->no_process_links = false;
$this->cache_pages = true;
$this->minimum_fetch_loop_time = C\MINIMUM_FETCH_LOOP_TIME;
$this->active_classifiers = [];
$this->active_rankers = [];
$this->tor_proxy = "";
$this->proxy_servers = [];
$ip_array = gethostbynamel(gethostname());
$this->total_git_urls = 0;
$this->all_git_urls = [];
$this->programming_language_extension = ['java' => 'java',
'py' => 'py'];
//we will get the correct crawl order from a queue_server
$this->crawl_order = self::HOST_BUDGETING;
$this->robots_txt = C\ALWAYS_FOLLOW_ROBOTS;
$this->max_depth = -1;
$this->summarizer_option = self::BASIC_SUMMARIZER;
$this->debug = "";
$this->num_multi_curl = C\NUM_MULTI_CURL_PAGES;
$this->sequence_number = 1;
}
/**
* Return the fetcher's copy of a page processor for the given
* mimetype.
*
* @param string $type mimetype want a processor for
* @param bool $fast whether to give the fast variant of the
* processor, whose summary is made by the ScrapeSummarizer, as
* for a host usesFastProcessor names
* @return object a page processor for that mime type of false if
* that mimetype can't be handled
*/
public function pageProcessor($type, $fast = false)
{
static $processor_cache = [];
static $plugin_hash = "";
if ($plugin_hash != $this->plugin_hash) {
$processor_cache = [];
$plugin_hash = $this->plugin_hash;
}
$cache_key = ($fast) ? "fast $type" : $type;
$summarizer_option = ($fast) ? self::BASIC_SUMMARIZER :
$this->summarizer_option;
if (isset($processor_cache[$cache_key])) {
return $processor_cache[$cache_key];
}
if (isset($this->page_processors[$type])) {
$page_processor = C\NS_PROCESSORS . $this->page_processors[$type];
$text_processor = C\NS_PROCESSORS . "TextProcessor";
} else {
$processor_cache[$cache_key] = false;
return false;
}
if (isset($this->plugin_processors[$page_processor])) {
$processor_cache[$cache_key] = new $page_processor(
$this->plugin_processors[$page_processor],
$this->max_description_len, -1, $summarizer_option);
} else {
$processor_cache[$cache_key] = new $page_processor([],
$this->max_description_len, -1, $summarizer_option);
}
if (L\generalIsA($page_processor, $text_processor)) {
$processor_cache[$cache_key]->text_data = true;
} else {
$processor_cache[$cache_key]->text_data = false;
}
return $processor_cache[$cache_key];
}
/**
* This is the function that should be called to get the fetcher to start
* fetching. Calls init to handle the command-line arguments then enters
* the fetcher's main loop
*/
public function start()
{
global $argv;
$this->channel = (empty($argv[3])) ? 0 : min(C\MAX_CHANNELS,
abs(intval($argv[3])));
if (isset($argv[2]) ) {
$arg2_parts = explode("-", $argv[2]);
if (count($arg2_parts) > 1) {
list($which_fetcher, $channel,) = $arg2_parts;
$this->channel = abs(intval($channel));
$this->which_fetcher = $which_fetcher;
unset($argv[3]);
} else {
$this->which_fetcher = intval($argv[2]);
$argv[2] = $argv[2] . "-" . $this->channel;
}
} else {
$this->which_fetcher = 0;
$argv[2] = "0-0";
}
CrawlDaemon::init($argv, "Fetcher");
$fetcher_name = $this->which_fetcher .
"-" . $this->channel . "-Fetcher";
L\crawlLog("\n\nInitialize logger..", $fetcher_name,
true);
L\crawlLog($this->which_fetcher . "-Fetcher using channel " .
$this->channel);
$this->loop();
}
/**
* Main loop for the fetcher.
*
* Checks for stop message, checks queue server if crawl has changed and
* for new pages to crawl. Loop gets a group of next pages to crawl if
* there are pages left to crawl (otherwise sleep 5 seconds). It downloads
* these pages, deduplicates them, and updates the found site info with the
* result before looping again.
*/
public function loop()
{
static $total_idle_time = 0;
L\crawlLog("In Fetch Loop");
/* A fatal error, such as running out of memory, ends the process
with nothing in this log, since PHP writes it elsewhere. This
puts it here, with the page being processed if there was one. */
register_shutdown_function(function () {
$error = error_get_last();
if (empty($error) || !in_array($error['type'], [E_ERROR,
E_CORE_ERROR, E_COMPILE_ERROR])) {
return;
}
$page = file_exists($this->pageInProcessName()) ?
file_get_contents($this->pageInProcessName()) : "none";
L\crawlLog("Fetcher stopped by a PHP error: " . $error['message'] .
" in " . $error['file'] . " line " . $error['line'] .
"; page being processed: $page; peak memory " .
memory_get_peak_usage());
});
L\crawlLog("PHP Version in use: " . phpversion());
$prefix = $this->which_fetcher . "-";
$this->temp_folder = C\TEMP_DIR . "/" . $this->which_fetcher . "-" .
$this->channel . "-Fetcher";
L\makePath($this->temp_folder);
PageProcessor::$temp_dir = $this->temp_folder;
$info[self::STATUS] = self::CONTINUE_STATE;
$local_archives = [""];
while (CrawlDaemon::processHandler()) {
$start_time = microtime(true);
$info = [];
$fetcher_message_file = $this->schedules_folder .
"/{$prefix}FetcherMessages.txt";
if (file_exists($fetcher_message_file)) {
$info = unserialize(file_get_contents($fetcher_message_file));
if (isset($info[self::DEBUG])) {
$this->debug = $info[self::DEBUG];
unlink($message_file);
continue;
}
unlink($fetcher_message_file);
if (isset($info[self::STATUS]) &&
$info[self::STATUS] == self::STOP_STATE) {
continue;
}
}
$switch_fetch_or_no_current = $this->checkCrawlTime();
if (($wake_up =
L\checkTimeInterval($this->sleep_start,
$this->sleep_duration)) > 0) {
L\crawlLog("MAIN LOOP CASE 0 --".
" FETCHER IS IN QUIESCENT/SLEEP MODE!!");
L\crawlLog("Current time is: " . date('r'));
L\crawlLog("Will wake up at " . date('r', $wake_up));
L\crawlLog("Current Crawl Timestamp: " .
$this->crawl_time);
L\interruptibleSleep($this->minimum_fetch_loop_time,
1, function () {
return !CrawlDaemon::processHandler(true);
});
continue;
} else if ($switch_fetch_or_no_current) { /* case (1) */
L\crawlLog("MAIN LOOP CASE 1 --".
" SWITCH CRAWL OR NO CURRENT CRAWL");
$info[self::CRAWL_TIME] = $this->crawl_time;
if ($info[self::CRAWL_TIME] == 0) {
$info[self::STATUS] = self::NO_DATA_STATE;
$this->to_crawl = [];
$this->sleep_start = "00:00";
$this->sleep_duration = "-1";
} else {
L\crawlLog("Crawl time is now " . $this->crawl_time);
}
} else if ($this->crawl_type == self::ARCHIVE_CRAWL &&
$this->arc_type != "WebArchiveBundle" &&
$this->arc_type != "") { /* case (2) */
// An archive crawl with data coming from the name server.
L\crawlLog(
"MAIN LOOP CASE 2 -- ARCHIVE SCHEDULER (NOT RECRAWL)");
$info = $this->checkArchiveScheduler();
if ($info === false) {
L\crawlLog("No Archive Schedule Data...".
" will try again in " . C\FETCH_SLEEP_TIME." seconds.");
L\interruptibleSleep(C\FETCH_SLEEP_TIME, 1,
function () {
return !CrawlDaemon::processHandler(true);
});
$total_idle_time += C\FETCH_SLEEP_TIME/C\ONE_HOUR;
continue;
}
} else if ($this->crawl_time > 0) { /* case (3) */
// Either a web crawl or a recrawl of a previous web crawl.
if ($this->crawl_type == self::ARCHIVE_CRAWL) {
L\crawlLog("MAIN LOOP CASE 3 -- RECRAWL SCHEDULER");
} else {
L\crawlLog("MAIN LOOP CASE 4 -- WEB SCHEDULER");
}
/* A fetcher on the queue server's machine sees the crawl's
index bundle on its own disk. Between batches it builds
one closed partition that nobody else has taken, then
merges the next built partition into the dictionary if
no other process is merging, so indexing keeps up with
crawling without a separate DictionaryUpdater process.
A fetcher on another
machine finds no bundle and only fetches.
It builds only when it holds no schedule: its results go
back only once to_crawl and to_crawl_again are empty,
and the queue server drains results in schedule order,
so a schedule held through a build of several minutes
stops the scheduler from sending batches to every
fetcher. */
$holds_schedule = count($this->to_crawl) > 0 ||
count($this->to_crawl_again) > 0 ||
$this->prefetch !== null;
foreach (($holds_schedule) ? [] :
[self::index_data_base_name,
self::double_index_base_name] as $base_name) {
$bundle_path = C\CACHE_DIR . "/" . $base_name .
$this->crawl_time;
if (!is_dir($bundle_path)) {
continue;
}
$build_start = microtime(true);
$built = IndexDocumentBundle::buildClosedPartition(
$bundle_path);
L\crawlLog("Looked for a partition to build in " .
L\changeInMicrotime($build_start));
if ($built >= 0) {
L\crawlLog("Built partition $built of $bundle_path");
}
$merged = IndexDocumentBundle::mergeNextPartition(
$bundle_path);
if ($merged >= 0) {
L\crawlLog("Merged partition $merged of " .
"$bundle_path into its dictionary");
}
}
$info = $this->checkScheduler();
if ($info === false) {
L\crawlLog("Cannot connect to name server...".
" will try again in ".C\FETCH_SLEEP_TIME." seconds.");
L\interruptibleSleep(C\FETCH_SLEEP_TIME, 1,
function () {
return !CrawlDaemon::processHandler(true);
});
$total_idle_time += C\FETCH_SLEEP_TIME/C\ONE_HOUR;
continue;
}
$this->updateDomainFilters();
} else {
L\crawlLog("MAIN LOOP CASE 5 -- NO CURRENT CRAWL");
$info[self::STATUS] = self::NO_DATA_STATE;
}
$idle_per_hour = $this->ratePerHour("IDLE", $total_idle_time,
time());
L\crawlLog("Percent idle time per hour: " .
number_format($idle_per_hour * 100, 2) ."%");
/* case (2), case (3) might have set info without
$info[self::STATUS] being set
*/
if (!isset($info[self::STATUS])) {
if ($info === true) {
$info = [];
}
$info[self::STATUS] = self::CONTINUE_STATE;
}
if ($info[self::STATUS] == self::NO_DATA_STATE) {
L\crawlLog("No data. Sleeping...");
L\interruptibleSleep(C\FETCH_SLEEP_TIME, 1,
function () {
return !CrawlDaemon::processHandler(true);
});
$total_idle_time += C\FETCH_SLEEP_TIME/C\ONE_HOUR;
continue;
}
$tmp_base_name = (isset($info[self::CRAWL_TIME])) ?
C\FETCHER_CACHE_DIR . "/{$prefix}" . self::archive_base_name .
$info[self::CRAWL_TIME] . ".txt" : "";
if (isset($info[self::CRAWL_TIME]) &&
(empty($this->crawl_stat_filename) ||
$this->crawl_stat_filename != $tmp_base_name)) {
if (!empty($this->crawl_stat_filename)) {
L\crawlLog("Old name: " . $this->crawl_stat_filename);
}
$this->to_crawl_again = [];
$this->found_sites = [];
L\garbageCollect();
$this->crawl_stat_filename = $tmp_base_name;
if (file_exists($this->crawl_stat_filename)) {
$this->crawl_stat_info = unserialize(file_get_contents(
$this->crawl_stat_filename));
} else {
$this->crawl_stat_info = ["NUM_SUMMARIZED_PAGES" => 0,
"NUM_CACHED_PAGES" => 0, "NUM_SEEN_SITES" => 0];
}
$this->crawl_time = $info[self::CRAWL_TIME];
$this->archive_send_time = 0;
$this->send_after = C\NUM_MULTI_CURL_PAGES;
$this->fetched_since_send = 0;
L\crawlLog("New name: ". $this->crawl_stat_filename);
L\crawlLog("Switching archive...");
if (!isset($info[self::ARC_DATA])) {
continue;
}
}
if (empty($this->crawl_stat_filename)) {
L\crawlLog("Fetcher not crawling empty skipping page download");
} else {
switch ($this->crawl_type) {
case self::WEB_CRAWL:
if ($this->prefetch === null) {
$downloaded_pages =
$this->downloadPagesWebCrawl();
} else {
$this->finishPrefetch();
$downloaded_pages = $this->prefetch->getReturn();
$this->prefetch = null;
}
$this->startPrefetch();
break;
case self::ARCHIVE_CRAWL:
if (isset($info[self::ARC_DATA])) {
$downloaded_pages = $info[self::ARC_DATA];
} else {
$downloaded_pages =
$this->downloadPagesArchiveCrawl();
}
break;
}
if (isset($downloaded_pages["NO_PROCESS"])) {
unset($downloaded_pages["NO_PROCESS"]);
$summarized_site_pages = array_values($downloaded_pages);
$this->no_process_links = true;
} else {
$summarized_site_pages =
$this->processFetchPages($downloaded_pages);
$this->no_process_links = false;
}
L\crawlLog("Number of summarized pages ".
count($summarized_site_pages));
$force_send = (isset($info[self::END_ITERATOR]) &&
$info[self::END_ITERATOR]) ? true : false;
$this->updateFoundSites($summarized_site_pages, $force_send);
}
$sleep_time = max(0, ceil($this->minimum_fetch_loop_time
- L\changeInMicrotime($start_time)));
if ($sleep_time > 0) {
L\crawlLog(
"Ensure minimum loop time by sleeping..." . $sleep_time);
$total_idle_time += $sleep_time/C\ONE_HOUR;
L\interruptibleSleep($sleep_time, 1, function () {
return !CrawlDaemon::processHandler(true);
});
}
if (!empty($this->debug)) {
L\crawlLog("Debug Message: {$this->debug} has been processed");
$this->debug = "";
}
}//end while
L\crawlLog("Fetcher shutting down!!");
}
/**
* Updates the array of domain filters currently loaded into memory based
* on which BloomFilterFiles are present in
* WORK_DIRECTORY/data/domain_filters and if they have changed since the
* current in-memory filters were loaded
*/
public function updateDomainFilters()
{
L\crawlLog("Checking for Updates to Domain Filters");
$filter_names = glob(self::DOMAIN_FILTER_GLOB);
$current_filters = array_keys($this->domain_filters);
foreach ($current_filters as $filter) {
if (!in_array($filter, $filter_names)) {
unset($this->domain_filters[$filter_name]);
L\crawlLog("Filter: " . $filter_name . " no longer exists, ".
"so removing it from active filters");
}
}
$updates = [];
foreach ($filter_names as $filter_name) {
if (empty($this->domain_filters[$filter_name])) {
L\crawlLog("New Filter: " . $filter_name .
" detected, so adding to filter list!");
$updates[] = $filter_name;
continue;
}
if ($this->domain_filters[$filter_name]["MODIFIED_TIME"] !=
filemtime($filter_name)) {
$updates[] = $filter_name;
L\crawlLog("Filter: " . $filter_name ." changed, so updating!");
}
}
foreach ($updates as $filter_name) {
$this->domain_filters[$filter_name]["MODIFIED_TIME"] =
filemtime($filter_name);
$this->domain_filters[$filter_name]["FILTER"] =
BloomFilterFile::load($filter_name);
if (empty($this->domain_filters[$filter_name]["FILTER"])) {
L\crawlLog("Problem Loading Filter: " . $filter_name .
", so disabling!");
unset($this->domain_filters[$filter_name]);
}
}
}
/**
* Get a list of urls from the current fetch batch provided by the queue
* server. Then downloads these pages. Finally, reschedules, if
* possible, pages that did not successfully get downloaded.
*
* @return array an associative array of web pages and meta data
* fetched from the internet
*/
public function downloadPagesWebCrawl()
{
static $total_downloads_succeeded = 0;
$start_time = microtime(true);
$can_schedule_again = false;
if (count($this->to_crawl) > 0) {
$can_schedule_again = true;
}
$sites = $this->getFetchSites();
L\crawlLog("Done getting list of ".count($sites)." to download...");
if (!$sites) {
L\crawlLog("No seeds to fetch...");
L\interruptibleSleep(max(0, ceil($this->minimum_fetch_loop_time
- L\changeInMicrotime($start_time))), 1, function () {
return !CrawlDaemon::processHandler(true);
});
return [];
}
$filtered_sites = [];
$site_pages = [];
foreach ($sites as $site) {
$hard_coded_parts = explode("###!", $site[self::URL]);
if (count($hard_coded_parts) > 1) {
if (!isset($hard_coded_parts[2])) {
$hard_coded_parts[2] = "";
}
$site[self::URL] = $hard_coded_parts[0];
$title = urldecode($hard_coded_parts[1]);
$description = urldecode($hard_coded_parts[2]);
$site[self::PAGE] = "<html><head><title>{$title}".
"</title></head><body><h1>{$title}</h1>".
"<p>{$description}</p></body></html>";
$site[self::HTTP_CODE] = 200;
$site[self::TYPE] = "text/html";
$site[self::ENCODING] = "UTF-8";
$site[self::IP_ADDRESSES] = ["0.0.0.0"];
$site[self::TIMESTAMP] = time();
$site_pages[] = $site;
} else {
$filtered_sites[] = $site;
}
}
if (!empty($this->graph_type)) {
$settings = ['graph_type' => $this->graph_type,
'out_degree' => $this->out_degree,
'seed' => $this->synthetic_seed,
'power_law_exponent' => $this->power_law_exponent,
'term_exponent' => $this->term_exponent,
'doc_length' => $this->doc_length,
'length_spread' => $this->length_spread,
'domain_link_probability' => $this->domain_link_probability,
'image_link_probability' => $this->image_link_probability];
$got_pages = FetchUrl::syntheticGetPages($filtered_sites,
$settings, self::URL, self::PAGE);
} else {
FetchUrl::setSiteCookies($this->site_cookies);
$got_pages = FetchUrl::getPages($filtered_sites, true,
$this->page_range_request, $this->temp_folder, self::URL,
self::PAGE,
false, null, false, $this->tor_proxy,
$this->proxy_servers, [$this, "compressWaitingPage"], true);
}
$site_pages = array_merge($site_pages, $got_pages);
L\crawlLog("..getPages call complete..");
for ($j = 0; $j < count($site_pages); $j++) {
if (isset($site_pages[$j][self::REPOSITORY_TYPE])) {
$git_repository_url = $site_pages[$j][self::URL];
$git_compressed_content = FetchGitRepositoryUrls::getGitData(
$git_repository_url);
$git_uncompressed_content = gzuncompress(
$git_compressed_content);
$length = strlen ($git_uncompressed_content);
$git_hash_end = strpos($git_uncompressed_content,
self::HEX_NULL_CHARACTER);
$git_uncompressed_content = substr($git_uncompressed_content,
$git_hash_end + 1, $length);
$site_pages[$j][self::PAGE] = $git_uncompressed_content;
$mime_type = UrlParser::guessMimeTypeFromFileName(
$site_pages[$j][self::FILE_NAME]);
if ($mime_type == "text/xml") {
$mime_type = "text/gitxml";
}
$site_pages[$j][self::TYPE] = $mime_type;
}
}
list($downloaded_pages, $schedule_again_pages) =
$this->reschedulePages($site_pages);
$total_downloads_succeeded += count($downloaded_pages);
$this->scheduleFailedPagesAgain($schedule_again_pages,
$can_schedule_again);
$this->downloads_per_hour = $this->ratePerHour("DOWNLOADS",
$total_downloads_succeeded, time());
L\crawlLog("...Downloads per hours: " . $this->downloads_per_hour);
L\crawlLog("Downloading complete");
return $downloaded_pages;
}
/**
* startPrefetch starts downloading the next batch of urls in a Fiber,
* so the download goes on while the batch just downloaded is
* processed. The Fiber runs downloadPagesWebCrawl, whose getPages call
* hands control back on each pass of its wait for transfers. It starts
* only in a web crawl that is not synthetic and only while to_crawl
* holds urls, so the batches that try failed urls again run as
* before.
*/
public function startPrefetch()
{
if ($this->crawl_type != self::WEB_CRAWL ||
!empty($this->graph_type) || count($this->to_crawl) == 0) {
return;
}
$this->prefetch = new \Fiber(function () {
return $this->downloadPagesWebCrawl();
});
$this->prefetch->start();
}
/**
* finishPrefetch runs the next batch's download until it ends, as the
* main loop does before processing that batch and updateFoundSites
* does before an upload. Between passes it sleeps
* PREFETCH_WAIT_MICROSECONDS.
*/
public function finishPrefetch()
{
while ($this->prefetch->isSuspended()) {
$this->prefetch->resume();
if ($this->prefetch->isSuspended()) {
usleep(self::PREFETCH_WAIT_MICROSECONDS);
}
}
}
/**
* Extracts NUM_MULTI_CURL_PAGES from the current Archive Bundle that is
* being recrawled.
*
* @return array an associative array of web pages and meta data from
* the archive bundle being iterated over
*/
public function downloadPagesArchiveCrawl()
{
$prefix = $this->which_fetcher . "-";
$arc_name = "$prefix" . self::archive_base_name . $this->crawl_index;
$base_name = C\FETCHER_CACHE_DIR . "/$arc_name";
$pages = [];
if (!isset($this->archive_iterator->iterate_timestamp) ||
$this->archive_iterator->iterate_timestamp != $this->crawl_index ||
$this->archive_iterator->result_timestamp != $this->crawl_time) {
if (!file_exists($base_name)){
L\crawlLog("!!Fetcher web archive $arc_name does not exist.");
L\crawlLog(" Only fetchers involved in original crawl will ");
L\crawlLog(" participate in a web archive recrawl!!");
return $pages;
} else {
L\crawlLog("Initializing Web Archive Bundle Iterator.");
$this->archive_iterator =
new A\WebArchiveBundleIterator($prefix, $this->crawl_index,
$this->crawl_time);
if ($this->archive_iterator == null) {
L\crawlLog("Error creating archive iterator!!");
return $pages;
}
}
}
if (!$this->archive_iterator->end_of_iterator) {
L\crawlLog("Getting pages from archive iterator...");
$pages = $this->archive_iterator->nextPages(C\NUM_MULTI_CURL_PAGES);
L\crawlLog("...pages get complete.");
}
return $pages;
}
/**
* Deletes any crawl web archive bundles not in the provided array of crawls
*
* @param array $still_active_crawls those crawls which should not
* be deleted, so all others will be deleted
* @see loop()
*/
public function deleteOldCrawls(&$still_active_crawls)
{
$prefix = $this->which_fetcher . "-";
$archives = glob(C\FETCHER_CACHE_DIR .
'/*'.self::archive_base_name.'*');
$full_base_name = $prefix . self::archive_base_name;
foreach ($archives as $archive) {
if (preg_match('/'.preg_quote(self::archive_base_name) . '(\d+)/',
$archive, $matches) !== false) {
$time = $matches[1];
if (!in_array($time, $still_active_crawls) ){
$this->db->unlinkRecursive($archive);
}
}
}
$files = glob($this->schedules_folder . '/*');
$names = [self::fetch_crawl_info, self::fetch_closed_name,
self::schedule_name, self::fetch_archive_iterator,
self::save_point, self::PAGE_IN_PROCESS_NAME,
self::SCHEDULE_LEFT_NAME];
foreach ($files as $file) {
$timestamp = "";
foreach ($names as $name) {
$full_name = $prefix . $name;
if (strlen(
$pre_timestamp = strstr($file, $full_name)) > 0) {
$timestamp = substr($pre_timestamp,
strlen($full_name), 10);
break;
}
}
if ($timestamp !== "" &&
!in_array($timestamp, $still_active_crawls)) {
if (is_dir($file)) {
$this->db->unlinkRecursive($file);
} else {
unlink($file);
}
}
}
}
/**
* Makes a request of the name server machine to get the timestamp of the
* currently running crawl to see if it changed
*
* If the timestamp has changed save the rest of the current fetch batch,
* then load any existing fetch from the new crawl; otherwise, set the crawl
* to empty. Also, handles deleting old crawls on this fetcher machine
* based on a list of current crawls on the name server.
*
* @return bool true if loaded a fetch batch due to time change
*/
public function checkCrawlTime()
{
static $saved_crawl_times = [];
$name_server = $this->name_server;
$start_time = microtime(true);
$time = time();
$this->check_crawl_time = $time;
$session = md5($time . C\p('AUTH_KEY'));
$prefix = $this->which_fetcher . "-" . $this->channel . "-";
$robot_instance = $prefix . C\p('ROBOT_INSTANCE');
$time_change = false;
$crawl_time = !is_null($this->crawl_time) ? $this->crawl_time : 0;
if ($crawl_time > 0) {
L\crawlLog("Checking name server:");
L\crawlLog(
" $name_server to see if active crawl time has changed.");
} else {
L\crawlLog("Checking name server:");
L\crawlLog(" $name_server to see if should start crawling");
}
$request =
$name_server . "?c=fetch&a=crawlTime&time=$time&session=$session" .
"&robot_instance=" . $robot_instance . "&machine_uri=" .
C\p('WEB_URI') .
"&crawl_time=$crawl_time&downloads_per_hour=" .
intval($this->downloads_per_hour);
L\crawlLog("Request was:" . $request);
/* Until it has one, a fetcher asks the name server for its Web Bot
Auth key pair, to sign its https requests with. */
$ask_key = (FetchUrl::$web_bot_auth_key_pair === null) ?
"&web_bot_auth=1" : "";
$info_string = FetchUrl::getPage($request . $ask_key, null, true);
L\crawlLog("Start of response was:" . substr($info_string, 0, 256));
$info = @unserialize(trim($info_string));
$key_pair = base64_decode($info[self::WEB_BOT_AUTH_KEY] ?? "", true);
if (function_exists("sodium_crypto_sign_detached") &&
$key_pair !== false &&
strlen($key_pair) == SODIUM_CRYPTO_SIGN_KEYPAIRBYTES) {
FetchUrl::$web_bot_auth_key_pair = $key_pair;
L\crawlLog("Received the name server's Web Bot Auth key.");
}
if (isset($info[self::SAVED_CRAWL_TIMES])) {
if (array_diff($info[self::SAVED_CRAWL_TIMES], $saved_crawl_times)
!= [] ||
array_diff($saved_crawl_times, $info[self::SAVED_CRAWL_TIMES])
!= []) {
$saved_crawl_times = $info[self::SAVED_CRAWL_TIMES];
$this->deleteOldCrawls($saved_crawl_times);
}
}
$check_cull_fields = [
self::RESTRICT_SITES_BY_URL => "restrict_sites_by_url",
self::ALLOWED_SITES => "allowed_sites",
self::DISALLOWED_SITES => "disallowed_sites"];
$cull_now_non_crawlable = false;
foreach ($check_cull_fields as $info_field => $field) {
if (isset($info[$info_field])) {
if (!isset($this->$field) || $this->$field !=
$info[$info_field]) {
$cull_now_non_crawlable = true;
}
$this->$field = $info[$info_field];
}
}
if ($cull_now_non_crawlable) {
L\crawlLog("Allowed/Disallowed Urls have changed");
L\crawlLog("Checking if urls in to crawl lists need to be culled");
$this->cullNoncrawlableSites();
}
if (isset($info[self::CRAWL_TIME])
&& ($info[self::CRAWL_TIME] != $this->crawl_time)) {
if ($info[self::CRAWL_TIME] > 0) {
L\crawlLog("New Crawl Time Found: {$info[self::CRAWL_TIME]}");
} else {
L\crawlLog("Crawl Time Changing to 0 -- No Crawl");
}
$dir = $this->schedules_folder;
$time_change = true;
/* Zero out the crawl. If haven't done crawl before, then scheduler
will be called */
$this->to_crawl = [];
$this->to_crawl_again = [];
$this->found_sites = [];
$this->prefetch = null;
if (isset($info[self::QUEUE_SERVERS])) {
$count_servers = count($info[self::QUEUE_SERVERS]);
if (!isset($this->queue_servers) ||
count($this->queue_servers) != $count_servers) {
L\crawlLog("New Queue Server List:");
$which_server = 0;
foreach ($info[self::QUEUE_SERVERS] as $server) {
$which_server++;
L\crawlLog("($which_server) $server");
}
}
$this->queue_servers = $info[self::QUEUE_SERVERS];
if (!isset($this->current_server) ||
$this->current_server > $count_servers) {
/*
prevent all fetchers from initially contacting same
queue servers
*/
$this->current_server = rand(0, $count_servers - 1);
}
}
if ($this->crawl_time > 0) {
file_put_contents("$dir/$prefix" . self::fetch_closed_name .
"{$this->crawl_time}.txt", "1");
}
/* Update the basic crawl info, so that we can decide between going
to a queue server for a schedule or to the name server for
archive data. */
$this->crawl_time = $info[self::CRAWL_TIME];
if ($this->crawl_time > 0 && isset($info[self::ARC_DIR]) ) {
$this->crawl_type = $info[self::CRAWL_TYPE];
$this->arc_dir = $info[self::ARC_DIR];
$this->arc_type = $info[self::ARC_TYPE];
} else {
$this->crawl_type = self::WEB_CRAWL;
$this->arc_dir = '';
$this->arc_type = '';
}
$this->setCrawlParamsFromArray($info);
// Load any batch that might exist for changed-to crawl
if (file_exists("$dir/$prefix" . self::fetch_crawl_info.
"{$this->crawl_time}.txt")) {
L\crawlLog("Loading old info for ".
"{$this->crawl_time}.");
$info = unserialize(file_get_contents(
"$dir/$prefix" . self::fetch_crawl_info.
"{$this->crawl_time}.txt"));
$this->setCrawlParamsFromArray($info);
unlink("$dir/$prefix" . self::fetch_crawl_info .
"{$this->crawl_time}.txt");
if (file_exists("$dir/$prefix".self::fetch_closed_name .
"{$this->crawl_time}.txt")) {
unlink("$dir/$prefix".self::fetch_closed_name .
"{$this->crawl_time}.txt");
}
$this->to_crawl = [];
$this->to_crawl_again = [];
$this->found_sites = [];
if ($this->crawl_type == self::WEB_CRAWL) {
$this->resumeScheduleLeft();
}
}
if (L\generalIsA(C\NS_ARCHIVE . $this->arc_type . "Iterator",
C\NS_ARCHIVE . "TextArchiveBundleIterator")) {
$result_dir = $this->schedules_folder . "/" .
$prefix . self::fetch_archive_iterator . $this->crawl_time;
$iterator_name = C\NS_ARCHIVE . $this->arc_type . "Iterator";
$this->archive_iterator = new $iterator_name(
$info[self::CRAWL_INDEX],
false, $this->crawl_time, $result_dir);
$this->db->setWorldPermissionsRecursive($result_dir);
}
} else if (!isset($info[self::CRAWL_TIME])) {
L\crawlLog("!!!!Server did not respond with a crawl time!!!!");
}
L\crawlLog("End Name Server Check");
if ($time_change) {
L\crawlLog("Crawl time has changed!");
} else {
L\crawlLog("Crawl time stayed same!");
}
return $time_change;
}
/**
* Get status, current crawl, crawl order, and new site information from
* the queue_server.
*
* @return mixed array or bool. If we are doing
* a web crawl and we still have pages to crawl then true, if the
* scheduler page fails to download then false, otherwise, returns
* an array of info from the scheduler.
*/
public function checkScheduler()
{
$prefix = $this->which_fetcher . "-" . $this->channel . "-";
$info = [];
$to_crawl_count = count($this->to_crawl);
$to_crawl_again_count = count($this->to_crawl_again);
if ($this->recrawl_check_scheduler) {
L\crawlLog("Archive Crawl checking ... Recrawl.");
}
if ((count($this->to_crawl) > 0 || count($this->to_crawl_again) > 0 ||
$this->prefetch !== null) && (!$this->recrawl_check_scheduler)) {
L\crawlLog(" Current to crawl count:" . $to_crawl_count);
L\crawlLog(" Current to crawl try again count:".
$to_crawl_again_count);
L\crawlLog("So not checking scheduler.");
return true;
}
$this->current_server = ($this->current_server + 1)
% count($this->queue_servers);
$this->recrawl_check_scheduler = false;
$queue_server = $this->queue_servers[$this->current_server];
L\crawlLog("Checking $queue_server for a new schedule.");
// hosts with error counts cleared with each schedule
$this->hosts_with_errors = [];
$start_time = microtime(true);
$time = time();
$session = md5($time . C\p('AUTH_KEY'));
$request =
$queue_server . "?c=fetch&a=schedule&time=$time&session=$session".
"&robot_instance=" . $prefix . C\p('ROBOT_INSTANCE').
"&machine_uri=" . C\p('WEB_URI') . "&crawl_time=" .
$this->crawl_time .
"&check_crawl_time=" . $this->check_crawl_time;
$info_string = FetchUrl::getPage($request, null, true);
L\crawlLog("Making schedule request: " . $request);
if ($info_string === false) {
L\crawlLog("The request failed!!!!");
return false;
}
$info_string = trim($info_string);
$tok = strtok($info_string, "\n");
$decode_token = base64_decode($tok);
L\crawlLog("First 256 bytes of first token in response:");
L\crawlLog(substr($decode_token, 0, 256));
if ((!isset($decode_token[0]) || $decode_token[0] != 'a') ||
(!isset($decode_token[1]) || $decode_token[1] != ':')) {
L\crawlLog("Error in decoding response, request failed!!!!");
return false;
}
$info = unserialize($decode_token);
$this->setCrawlParamsFromArray($info);
if (isset($info[self::SEQUENCE_NUMBER])) {
L\crawlLog("Received schedule number: " .
$info[self::SEQUENCE_NUMBER]);
$this->schedule_received_time = time();
}
if (isset($info[self::SITES])) {
$tok = strtok("\n"); //skip meta info
$this->last_try_urls = array_fill_keys(
$info[self::LAST_TRY_URLS] ?? [], true);
$this->to_crawl = [];
while($tok !== false) {
$string = base64_decode($tok);
$read = unpack("G", substr($string, 0 , 4));
$weight = $read[1];
$depth = L\unpackInt(substr($string, 4 , 4));
$delay = L\unpackInt(substr($string, 8, 4));
$url = substr($string, 12);
$this->to_crawl[] = [$url, $weight, $depth, $delay];
$tok = strtok("\n");
}
file_put_contents($this->schedules_folder . "/$prefix".
self::fetch_crawl_info . "{$this->crawl_time}.txt",
serialize($info));
$this->schedule_resend_time = time() +
MessagesBundle::scheduleTimeout(count($this->to_crawl));
$this->saveScheduleLeft();
}
L\crawlLog("Time to check Scheduler " .
L\changeInMicrotime($start_time));
return $info;
}
/**
* During an archive crawl this method is used to get from the name server
* a collection of pages to process. The fetcher will later process these
* and send summaries to various queue_servers.
*
* @return array containing archive page data
*/
public function checkArchiveScheduler()
{
$start_time = microtime(true);
/*
It's still important to switch queue servers, so that we send new
data to each server each time we fetch
new data from the name server. The pages are sent the first time
there are any, and then whenever a minute has passed since the
last send, so a user sees results within a minute of the start.
*/
$send_now = !empty($this->found_sites[self::SEEN_URLS]) &&
time() - $this->archive_send_time >= C\ONE_MINUTE;
$this->selectCurrentServerAndUpdateIfNeeded($send_now);
if ($send_now) {
$this->archive_send_time = time();
}
$chunk = false;
if (L\generalIsA(C\NS_ARCHIVE . $this->arc_type . "Iterator",
C\NS_ARCHIVE . "TextArchiveBundleIterator")) {
$archive_iterator = $this->archive_iterator;
$chunk = true;
$info = [];
$max_offset = A\TextArchiveBundleIterator::BUFFER_SIZE +
A\TextArchiveBundleIterator::MAX_RECORD_SIZE;
if ($archive_iterator->buffer_fh &&
$archive_iterator->current_offset < $max_offset) {
L\crawlLog("Local Iterator Offset: ".
$archive_iterator->current_offset);
L\crawlLog("Local Max Offset: ". $max_offset);
$info[self::ARC_DATA] =
$archive_iterator->nextPages(C\ARCHIVE_BATCH_SIZE);
L\crawlLog("Time to get archive data from local buffer ".
L\changeInMicrotime($start_time));
}
if ($archive_iterator->buffer_fh
&& $archive_iterator->current_offset < $max_offset ) {
return $info;
}
if (isset($info[self::ARC_DATA]) && count($info[self::ARC_DATA])>0){
$arc_data = $info[self::ARC_DATA];
}
L\crawlLog("Done processing Local Buffer, requesting more data...");
}
L\crawlLog("Fetching Archive data from name server with request:");
$name_server = $this->name_server;
$time = time();
$session = md5($time . C\p('AUTH_KEY'));
$prefix = $this->which_fetcher . "-" . $this->channel . "-";
$request =
$name_server."?c=fetch&a=archiveSchedule&time=$time".
"&session=$session&robot_instance=" . $prefix .
C\p('ROBOT_INSTANCE').
"&machine_uri=" . C\p('WEB_URI') . "&crawl_time=" .
$this->crawl_time .
"&check_crawl_time=" . $this->check_crawl_time;
L\crawlLog($request);
$response_string = FetchUrl::getPage($request, null, true);
if ($response_string === false) {
L\crawlLog("Request failed!");
return false;
}
if ($response_string) {
$info = @unserialize($response_string);
} else {
$info = [];
$info[self::STATUS] = self::NO_DATA_STATE;
}
$this->setCrawlParamsFromArray($info);
if (isset($info[self::DATA])) {
/* Unpack the archive data and return it in the $info array; also
write a copy to disk in case something goes wrong. */
$pages = unserialize(gzuncompress(L\webdecode($info[self::DATA])));
if ($chunk) {
if (isset($pages[self::ARC_DATA]) ) {
if (isset($pages[self::INI])) {
$archive_iterator->setIniInfo($pages[self::INI]);
}
if ($pages[self::ARC_DATA]) {
$archive_iterator->makeBuffer($pages[self::ARC_DATA]);
}
if (isset($pages[self::HEADER]) &&
is_array($pages[self::HEADER]) &&
$pages[self::HEADER] != []) {
$archive_iterator->header = $pages[self::HEADER];
}
if (!$pages[self::START_PARTITION]) {
$archive_iterator->nextPages(1);
}
if (isset($pages[self::PARTITION_NUM])) {
L\crawlLog(" Done get data".
" from file {$pages[self::PARTITION_NUM]}");
}
if (isset($pages[self::NUM_PARTITIONS])) {
L\crawlLog(
" of {$pages[self::NUM_PARTITIONS]} files.");
}
}
if (isset($arc_data)) {
$info[self::ARC_DATA] = $arc_data;
}
} else {
$info[self::ARC_DATA] = $pages;
}
} else if (isset($info['ARCHIVE_BUNDLE_ERROR'])) {
L\crawlLog(" ".$info['ARCHIVE_BUNDLE_ERROR']);
}
L\crawlLog("Time to fetch archive data from name server ".
L\changeInMicrotime($start_time));
return $info;
}
/**
* At least once, and while memory is low selects next server and send
* any fetcher data we have to it.
*
* @param bool $at_least_current_server whether to send to the site info
* to at least one queue server or to send only if memory is above
* threshold. Only in later casee is next server advanced.
*/
public function selectCurrentServerAndUpdateIfNeeded(
$at_least_current_server)
{
$i = 0;
$num_servers = count($this->queue_servers);
/* Make sure no queue server starves if to crawl data available.
Try to keep memory foot print smaller.
*/
$cs = $this->current_server;
do {
if ($at_least_current_server ||
(isset($this->found_sites[self::TO_CRAWL][$cs]) &&
count($this->found_sites[self::TO_CRAWL][$cs]) > 0)) {
$this->updateScheduler($cs, $at_least_current_server);
$at_least_current_server = false;
}
$cs = ($cs + 1) % $num_servers;
$i++;
} while(memory_get_usage() >
L\memoryCeiling(C\MEMORY_FILL_FACTOR) &&
$i < $num_servers);
}
/**
* Sets parameters for fetching based on provided info struct
* ($info typically would come from the queue server)
*
* @param array &$info struct with info about the kind of crawl, timestamp
* of index, crawl order, etc.
*/
public function setCrawlParamsFromArray(&$info)
{
if (empty($info)) {
return;
}
/* QUEUE_SERVERS and CURRENT_SERVER might not be set if info came
from a queue_server rather than from name server
*/
if (isset($info[self::QUEUE_SERVERS])) {
$this->queue_servers = $info[self::QUEUE_SERVERS];
} else {
$info[self::QUEUE_SERVERS] = $this->queue_servers;
}
if (!isset($info[self::CURRENT_SERVER])) {
$info[self::CURRENT_SERVER] = $this->current_server;
}
$update_fields = [
self::ALLOWED_SITES => 'allowed_sites',
self::CACHE_PAGES => 'cache_pages',
self::CRAWL_INDEX => "crawl_index",
self::CRAWL_ORDER => 'crawl_order',
self::MAX_DEPTH => 'max_depth',
self::SLEEP_START => 'sleep_start',
self::SLEEP_DURATION => 'sleep_duration',
self::CRAWL_TYPE => "crawl_type",
self::GRAPH_TYPE => 'graph_type',
self::OUT_DEGREE => 'out_degree',
self::DOMAIN_LINK_PROBABILITY => 'domain_link_probability',
self::POWER_LAW_EXPONENT => 'power_law_exponent',
self::TERM_EXPONENT => 'term_exponent',
self::DOC_LENGTH => 'doc_length',
self::LENGTH_SPREAD => 'length_spread',
self::IMAGE_LINK_PROBABILITY => 'image_link_probability',
self::SYNTHETIC_SEED => 'synthetic_seed',
self::DISALLOWED_SITES => 'disallowed_sites',
self::SITE_COOKIES => 'site_cookies',
self::INDEXED_FILE_TYPES => 'indexed_file_types',
self::MINIMUM_FETCH_LOOP_TIME => 'minimum_fetch_loop_time',
self::PROXY_SERVERS => 'proxy_servers',
self::RESTRICT_SITES_BY_URL => 'restrict_sites_by_url',
self::ROBOTS_TXT => 'robots_txt',
self::SUMMARIZER_OPTION => "summarizer_option",
self::TOR_PROXY => 'tor_proxy'];
$check_cull_fields = ["restrict_sites_by_url", "allowed_sites",
"disallowed_sites"];
$cull_now_non_crawlable = false;
foreach ($update_fields as $info_field => $field) {
if (isset($info[$info_field])) {
if (in_array($info_field, $check_cull_fields) &&
(!isset($this->$field) || $this->$field !=
$info[$info_field]) ) {
$cull_now_non_crawlable = true;
}
if ($this->$field != $info[$info_field]) {
$this->$field = $info[$info_field];
$print_field = var_export($info[$info_field], true);
L\crawlLog(
"Crawl Field: $field updated to $print_field");
}
}
}
L\crawlLog("Minimum fetch loop time is: " .
$this->minimum_fetch_loop_time . " seconds");
if ($cull_now_non_crawlable) {
L\crawlLog("Allowed/Disallowed Urls have changed");
L\crawlLog("Checking if urls in to crawl lists need to be culled");
$this->cullNoncrawlableSites();
}
if (!empty($info[self::ACTIVE_CLASSIFIERS_DATA])) {
$this->active_classifiers = isset($info[self::ACTIVE_CLASSIFIERS])
&& is_array( $info[self::ACTIVE_CLASSIFIERS]) ?
$info[self::ACTIVE_CLASSIFIERS] : [];
$this->active_rankers = isset($info[self::ACTIVE_RANKERS])
&& is_array($info[self::ACTIVE_RANKERS]) ?
$info[self::ACTIVE_RANKERS] : [];
/*
The classifier data is set by the fetch controller for each
active classifier, and is a compressed, serialized structure
containing all of the objects needed for classification.
*/
$classifiers_data = $info[self::ACTIVE_CLASSIFIERS_DATA];
$this->classifiers = [];
foreach ($classifiers_data as $label => $classifier_data) {
if ($classifier_data) {
$classifier = Classifier::newClassifierFromData(
$classifier_data);
$this->classifiers[] = $classifier;
L\crawlLog("Loading '{$label}' classifier/ranker.");
if (in_array($label, $this->active_classifiers)) {
L\crawlLog(" Using '{$label}' as a classifier.");
}
if (in_array($label, $this->active_rankers)) {
L\crawlLog(" Using '{$label}' as a ranker.");
}
} else {
L\crawlLog("Skipping classifier '{$label}'; missing ".
"finalized data.");
}
}
}
if (isset($info[self::PAGE_RULES]) ){
$rule_string = implode("\n", $info[self::PAGE_RULES]);
$rule_string = html_entity_decode($rule_string, ENT_QUOTES);
$this->page_rule_parser =
new PageRuleParser($rule_string);
}
if (isset($info[self::INDEXING_PLUGINS]) &&
$this->plugin_hash !=
L\crawlHash(serialize($info[self::INDEXING_PLUGINS]) )) {
$this->plugin_hash =
L\crawlHash(serialize($info[self::INDEXING_PLUGINS]));
$this->plugin_processors = [];
foreach ($info[self::INDEXING_PLUGINS] as $plugin) {
if (empty($plugin)) {
continue;
}
$plugin_name = C\NS_PLUGINS . $plugin . "Plugin";
$processors = $plugin_name::getProcessors();
$plugin_object = new $plugin_name();
if (method_exists($plugin_name, "setConfiguration") &&
isset($info[self::INDEXING_PLUGINS_DATA][$plugin])) {
$plugin_object->setConfiguration(
$info[self::INDEXING_PLUGINS_DATA][$plugin]);
}
foreach ($processors as $processor) {
$this->plugin_processors[C\NS_PROCESSORS .
$processor][$plugin_name] = $plugin_object;
}
}
foreach ($this->indexed_file_types as $file_type) {
$processor = C\NS_PROCESSORS . ucfirst($file_type)."Processor";
$processor_path = C\BASE_DIR . "/library/processors/".
ucfirst($file_type)."Processor.php";
if (!class_exists($processor)) {
continue;
}
if (!isset($this->plugin_processors[$processor])) {
$this->plugin_processors[$processor] = [];
}
$parent_processor = $processor;
while(($parent_processor =
get_parent_class($parent_processor)) &&
$parent_processor != C\NS_PROCESSORS . "PageProcessor") {
if (isset($this->plugin_processors[$parent_processor])) {
$this->plugin_processors[$processor] +=
$this->plugin_processors[$parent_processor];
}
}
}
foreach ($this->plugin_processors as $processor => $plugins) {
$this->plugin_processors[$processor] = array_values($plugins);
}
}
if (!empty($info[self::SCRAPERS]) ) {
$this->scrapers = unserialize(base64_decode($info[self::SCRAPERS]));
}
if (isset($info[self::SCHEDULE_TIME])) {
$this->schedule_time = $info[self::SCHEDULE_TIME];
}
if (isset($info[self::PAGE_RANGE_REQUEST])) {
$this->page_range_request = $info[self::PAGE_RANGE_REQUEST];
}
if (isset($info[self::MAX_DESCRIPTION_LEN])) {
$this->max_description_len = $info[self::MAX_DESCRIPTION_LEN];
}
if (isset($info[self::MAX_LINKS_TO_EXTRACT])) {
$this->max_links_to_extract = $info[self::MAX_LINKS_TO_EXTRACT];
}
if (isset($info[self::SEQUENCE_NUMBER])) {
$this->sequence_number = $info[self::SEQUENCE_NUMBER];
}
}
/**
* Prepare an array of up to NUM_MULTI_CURL_PAGES' worth of sites to be
* downloaded in one go using the to_crawl array. Delete these sites
* from the to_crawl array.
*
* @return array sites which are ready to be downloaded
*/
public function getFetchSites()
{
$start_time = microtime(true);
$seeds = [];
$delete_indices = [];
$num_items = count($this->to_crawl);
if ($num_items > 0) {
$crawl_source = & $this->to_crawl;
$this->num_download_attempts = 1;
$to_crawl_flag = true;
} else {
L\crawlLog("...Trying to crawl sites which failed the first time");
$num_items = count($this->to_crawl_again);
$crawl_source = & $this->to_crawl_again;
$this->num_download_attempts++;
$to_crawl_flag = false;
}
reset($crawl_source);
if ($num_items > $this->num_multi_curl) {
$num_items = $this->num_multi_curl;
}
/* The allowance for hosts needing an address lookup comes from
how wide a whole pass may be, not from how many urls are left in
the batch. Reckoning it from what is left shrank the allowance as
the batch drained, so the last of a batch came down in twos. */
$num_ip_lookups = max($this->num_multi_curl /
self::URLS_PER_HOST_LOOKUP, self::LEAST_HOST_LOOKUPS_A_PASS);
$i = 0;
$lookup_hosts = [];
foreach ($crawl_source as $site_key => $site_value) {
/* A scheduled url reads url###ip###referer, the ip filled in
where the queue server already knows the host address. A url
with that part empty has a host that still needs a lookup.
The urls of one host wait on the same name, so a host is
counted against the allowance once, on its first url. */
$host = ($site_value[0] == self::DUMMY) ? "" :
UrlParser::getHost($site_value[0]);
$needs_lookup = $host != "" &&
empty(explode("###", $site_value[0])[1]) &&
!isset($lookup_hosts[$host]);
if ($i >= $num_items || ($needs_lookup &&
count($lookup_hosts) >= $num_ip_lookups)) {
break;
}
$delete_indices[] = $site_key;
if ($site_value[0] != self::DUMMY) {
if ($needs_lookup) {
$lookup_hosts[$host] = true;
}
// only download if host doesn't seem congested
if (!isset($this->hosts_with_errors[$host]) ||
$this->hosts_with_errors[$host] <
C\DOWNLOAD_ERROR_THRESHOLD) {
$url_to_check = $site_value[0];
$extension = UrlParser::getDocumentType($url_to_check);
$repository_indicator = FetchGitRepositoryUrls::
checkForRepository($extension);
if ($repository_indicator == self::REPOSITORY_GIT) {
$git_internal_urls = FetchGitRepositoryUrls::
setGitRepositoryUrl($url_to_check, $i, $seeds,
$repository_indicator, $site_value,
$this->total_git_urls, $this->all_git_urls);
$i = $git_internal_urls['position'];
$git_url_index = $git_internal_urls['index'];
$seeds = $git_internal_urls['seeds'];
$repository_indicator = $git_internal_urls['indicator'];
$this->total_git_urls = $git_internal_urls['count'];
$this->all_git_urls = $git_internal_urls['all'];
} else {
$seeds[$i][self::URL] = $site_value[0];
$seeds[$i][self::WEIGHT] = $site_value[1];
$seeds[$i][self::DOC_DEPTH] = $site_value[2];
$seeds[$i][self::CRAWL_DELAY] = $site_value[3];
if (!empty($site_value[self::NO_RANGE])) {
$seeds[$i][self::NO_RANGE] = true;
}
}
/*
Crawl delay is only used in scheduling on the QueueServer.
On the Fetcher, we only use crawl-delay to determine
if we will give a page a second try if it doesn't
download the first time
*/
if (UrlParser::getDocumentFilename($seeds[$i][self::URL]) .
"." . UrlParser::getDocumentType($seeds[$i][self::URL])
== "robots.txt") {
$seeds[$i][self::ROBOT_PATHS] = [
self::ALLOWED_SITES => [],
self::DISALLOWED_SITES => []];
}
$i++;
}
} else {
break;
}
} //end for
foreach ($delete_indices as $delete_index) {
$git_set = false;
if ($to_crawl_flag == true) {
$extension = UrlParser::getDocumentType(
$this->to_crawl[$delete_index][0]);
$repository_type = FetchGitRepositoryUrls::checkForRepository(
$extension);
if ($repository_type != self::REPOSITORY_GIT) {
unset($this->to_crawl[$delete_index]);
}
} else {
$extension = UrlParser::getDocumentType(
$this->to_crawl_again[$delete_index][0]);
$repository_type = FetchGitRepositoryUrls::checkForRepository(
$extension);
unset($this->to_crawl_again[$delete_index]);
}
if ($repository_type == self::REPOSITORY_GIT) {
if (!$git_set) {
$next_url_start = $url_to_check . self::GIT_URL_CONTINUE .
$git_url_index;
$git_set = true;
$this->to_crawl[$delete_index][0] = $next_url_start;
}
if ($repository_indicator == self::INDICATOR_NONE) {
unset($this->to_crawl[$delete_index]);
}
}
}
L\crawlLog("Fetch url list to download time ".
L\changeInMicrotime($start_time));
return $seeds;
}
/**
* Sorts out pages for which no content was downloaded so that they can be
* scheduled to be crawled again.
*
* @param array &$site_pages pages to sort
* @return an array conisting of two array downloaded pages and
* not downloaded pages.
*/
public function reschedulePages(&$site_pages)
{
$start_time = microtime(true);
$downloaded = [];
$not_downloaded = [];
$time = time();
foreach ($site_pages as $site) {
if ( (isset($site[self::ROBOT_PATHS]) || isset($site[self::PAGE]))
&& (is_numeric($site[self::HTTP_CODE] ) &&
$site[self::HTTP_CODE] > 0 && $site[self::HTTP_CODE] != 416) ) {
$downloaded[] = $site;
} else if (substr($site[self::URL], -strlen("robots.txt"))
== "robots.txt" && $time - $this->crawl_time > C\ONE_DAY) {
/*
Assume slow to respond robots sites after the first day
of crawling are probably spammy (slow to respond and
low value content, sucking up lots of reschedule
resources), so don't reschedule or bother trying to crawl
the rest of the site. If have allowed recrawl after x time
set, then after x time if the site is legit we will
probably find and try it again.
*/
L\crawlLog("Slow to respond robots.txt at: ". $site[self::URL] .
" so blocking rather than rescheduling.");
$site[self::HTTP_CODE] = 408;
$site[self::PAGE] = "User-agent: *\nDisallow: /\n";
$site[self::TYPE] = "text/robot";
$downloaded[] = $site;
} else if ($this->num_download_attempts >=
C\MAX_DOWNLOAD_ATTEMPTS) {
$site[self::HTTP_CODE] = 418; /* teapot code for multiple failed
downloads */
$site[self::PAGE] = $site[self::URL];
$site[self::TYPE] = "text/plain";
$downloaded[] = $site;
} else {
if ($site[self::HTTP_CODE] == 416) {
L\crawlLog("Range request not satisfiable...");
$site[self::NO_RANGE] = true;
}
L\crawlLog("Rescheduling ". $site[self::URL]);
$not_downloaded[] = $site;
}
}
L\crawlLog(" Sort downloaded/not downloaded ".
L\changeInMicrotime($start_time));
return [$downloaded, $not_downloaded];
}
/**
* scheduleFailedPagesAgain decides what happens to the pages of a
* batch that got no answer. downloadPagesWebCrawl calls it with the
* pages reschedulePages did not count as downloaded. While the batch
* still has pages it has not tried, a page of a site with no crawl
* delay is put in to_crawl_again, to be tried once more in this batch.
* Every other such page, one of a site with a crawl delay or one that
* failed in a later try of the batch, is put in RETRY_URLS with code
* 0, so the queue server can queue it to be fetched again.
*
* @param array $failed_pages the pages that got no answer
* @param bool $can_schedule_again whether this pass was over pages of
* the batch not tried before, so a page may get one more try in
* the batch
*/
public function scheduleFailedPagesAgain($failed_pages,
$can_schedule_again)
{
$num_queue_servers = count($this->queue_servers);
foreach ($failed_pages as $failed_page) {
if ($can_schedule_again &&
($failed_page[self::CRAWL_DELAY] ?? -1) == 0) {
$again = [$failed_page[self::URL],
$failed_page[self::WEIGHT],
$failed_page[self::DOC_DEPTH],
$failed_page[self::CRAWL_DELAY]];
if (!empty($failed_page[self::NO_RANGE])) {
$again[self::NO_RANGE] = true;
}
$this->to_crawl_again[] = $again;
continue;
}
$part = L\calculatePartition($failed_page[self::URL],
$num_queue_servers, C\NS_LIB . "UrlParser::getHost");
$this->found_sites[self::RETRY_URLS][$part][] =
[$failed_page[self::URL], $failed_page[self::WEIGHT] ?? 0,
$failed_page[self::DOC_DEPTH] ?? 0, "", 0.0, 0];
L\crawlLog($failed_page[self::URL] . " got no answer and will " .
"be offered to the queue server to be fetched again");
}
L\crawlLog(" Pages to try again in this batch: " .
count($this->to_crawl_again));
}
/**
* mayRetryRefusal says if a page a site refused with 403 Forbidden is
* worth fetching again later. A site may refuse a crawler by the TLS
* fingerprint of its connection, which FetchUrl::varyTlsHello changes
* from one connection to the next, so a later try may pass. Only a
* crawl whose robots.txt mode is Allow Landing Page or Ignore tries
* again, and only for a seed page, at depth 0, or a site's landing
* page, the root of its host; the Scheduler then sends it out at most
* QueueServer::FETCH_TRIES_MAX times. processFetchPages calls it for
* each page that failed.
*
* @param array $site the page as downloaded
* @param int $code the HTTP code the page got
* @return bool whether to send the page back to be fetched again
*/
public function mayRetryRefusal($site, $code)
{
if ($code != 403 || !in_array($this->robots_txt,
[C\ALLOW_LANDING_ROBOTS, C\IGNORE_ROBOTS])) {
return false;
}
$url_parts = @parse_url($site[self::URL]);
return ($site[self::DOC_DEPTH] ?? -1) == 0 ||
(in_array($url_parts['path'] ?? "", ["", "/"]) &&
empty($url_parts['query']));
}
/**
* indexesUrlWords says if a page that gave no text of its own is
* indexed by the words of its url. That is so for a site's home page,
* the root of its host with no query, in a crawl whose robots.txt
* mode is Allow Landing Page, since such a crawl means to have every
* site's home page in its index. processFetchPages calls it for a
* page that gave no description, and for a bot-protection challenge
* on the Scheduler's last try at the page, as $last_try_urls lists.
*
* @param string $url the page's url
* @return bool whether the page is indexed by the words of its url
*/
public function indexesUrlWords($url)
{
if ($this->robots_txt != C\ALLOW_LANDING_ROBOTS) {
return false;
}
return trim(UrlParser::getWordsInHostUrl($url)) != "";
}
/**
* urlWordsDocInfo gives the summary of a home page indexed by the words
* of its url. Its description is the words of its host, as pepsi for
* https://www.pepsi.com/, followed by the host, so the page's hash
* differs from that of every other host's. Its title is the host,
* unless the page gave a title of its own. Its other fields, such as
* its links, are kept.
*
* @param string $url the home page's url
* @param array $doc_info what processing the page gave, if anything
* @return array the page's summary
*/
public function urlWordsDocInfo($url, $doc_info = [])
{
$host = @parse_url($url, PHP_URL_HOST) ?: $url;
$description = trim(UrlParser::getWordsInHostUrl($url)) . " " .
$host;
$doc_info[self::DESCRIPTION] = $description;
$doc_info[self::PAGE] = $description;
if (trim($doc_info[self::TITLE] ?? "") == "") {
$doc_info[self::TITLE] = $host;
}
L\crawlLog(" $url gave no text, so it is indexed by the words " .
"of its url: $description");
return $doc_info;
}
/**
* Processes an array of downloaded web pages with the appropriate page
* processor.
*
* Summary data is extracted from each non robots.txt file in the array.
* Disallowed paths and crawl-delays are extracted from robots.txt files.
*
* @param array $site_pages a collection of web pages to process
* @return array summary data extracted from these pages
*/
public function processFetchPages($site_pages)
{
$page_processors = $this->page_processors;
L\crawlLog("Start process pages... Current Memory:" .
memory_get_usage());
$start_time = microtime(true);
$processor_times = [];
$prefix = $this->which_fetcher . "-";
$stored_site_pages = [];
$summarized_site_pages = [];
$num_items = $this->crawl_stat_info['NUM_SUMMARIZED_PAGES'];
$num_cached_pages = 0;
$found_thumb_urls = false;
$i = 0;
/* The parts of processing no processor takes, timed so the
processing time line can say what its rest is made of. */
$download_pass_time = 0;
$link_time = 0;
foreach ($site_pages as $site) {
/* The next batch's download gets a pass between pages, so its
transfers go on while this batch is processed. */
if ($this->prefetch?->isSuspended()) {
$pass_start = microtime(true);
$this->prefetch->resume();
$download_pass_time += microtime(true) - $pass_start;
}
$url_words_only = false;
$sent_back = false;
$last_try = isset($this->last_try_urls[$site[self::URL]]);
if (FetchUrl::isBotChallenge($site[self::PAGE] ?? "")) {
/* The site sent a bot check in place of its page. The
page is offered to the queue server to be fetched
again, with the code for a server that is busy. On
the Scheduler's last try, a site's home page in a
crawl that allows landing pages is indexed by the words
of its url. Otherwise the page is not indexed, and an
earlier copy in the index is kept. */
$part = L\calculatePartition($site[self::URL],
count($this->queue_servers),
C\NS_LIB . "UrlParser::getHost");
$this->found_sites[self::RETRY_URLS][$part][] =
[$site[self::URL], $site[self::WEIGHT] ?? 0,
$site[self::DOC_DEPTH] ?? 0, "", 0.0,
self::CHALLENGE_RETRY_CODE];
$sent_back = true;
$url_words_only = $last_try &&
$this->indexesUrlWords($site[self::URL]);
L\crawlLog($site[self::URL] . " answered with a " .
"bot-protection challenge, so it " . ($url_words_only ?
"is indexed by the words of its url" : "is not indexed") .
" and will be offered to the queue server to be " .
"fetched again");
if (!$url_words_only) {
continue;
}
}
$response_code = $site[self::HTTP_CODE];
$was_error = false;
if ($response_code < 200 || $response_code >= 300) {
L\crawlLog($site[self::URL] . " response code $response_code");
$host = UrlParser::getHost($site[self::URL]);
if (!isset($this->hosts_with_errors[$host])) {
$this->hosts_with_errors[$host] = 0;
}
if (($response_code >= 400 && $response_code != 404) ||
$response_code < 100) {
/* < 100 will capture failures to connect which are
returned as strings */
$was_error = true;
$this->hosts_with_errors[$host]++;
}
/* A page that failed in a way that may pass is also sent
back, so the Scheduler can queue it again; the queue
server it goes to is picked as for its links. */
$retry_code = is_numeric($response_code) ?
intval($response_code) : 0;
if ($was_error && !$sent_back &&
!isset($site[self::ROBOT_PATHS]) &&
($retry_code < 100 ||
in_array($retry_code, self::RETRY_HTTP_CODES) ||
$this->mayRetryRefusal($site, $retry_code))) {
$part = L\calculatePartition($site[self::URL],
count($this->queue_servers),
C\NS_LIB . "UrlParser::getHost");
$this->found_sites[self::RETRY_URLS][$part][] =
[$site[self::URL], $site[self::WEIGHT] ?? 0,
$site[self::DOC_DEPTH] ?? 0, "", 0.0, $retry_code];
$sent_back = true;
L\crawlLog($site[self::URL] . " will be offered to the " .
"queue server to be fetched again");
}
/* we print out errors to std output. We still go ahead and
process the page. Maybe it is a cool error page, also
this makes sure we don't crawl it again
*/
}
// text/robot is my made up mimetype for robots.txt files
$was_robot_error = false;
if (isset($site[self::ROBOT_PATHS])) {
if (!$was_error) {
$type = "text/robot";
} else {
$type = $site[self::TYPE];
if ($response_code != 404) {
/*
disallow crawling if robots.txt has any error other
than not found
*/
$was_robot_error = true;
$site[self::ROBOT_PATHS][
self::DISALLOWED_SITES][] = "/";
}
}
} else if (isset($site[self::FILE_NAME])) {
$extension = UrlParser::getDocumentType($site[self::FILE_NAME]);
if ($extension ==
$this->programming_language_extension['java']) {
$type = "text/java";
} else if ($extension ==
$this->programming_language_extension['py']) {
$type = "text/py";
} else {
$type = $site[self::TYPE];
}
} else {
$type = $site[self::TYPE];
}
$handled = false;
/*deals with short URLs and directs them to the original link
for robots.txt don't want to introduce stuff that can be
mis-parsed (we follow redirects in this case anyway) */
if (isset($site[self::LOCATION]) &&
count($site[self::LOCATION]) > 0
&& strcmp($type, "text/robot") != 0) {
array_unshift($site[self::LOCATION], $site[self::URL]);
$tmp_loc = array_pop($site[self::LOCATION]);
$tmp_loc = UrlParser::canonicalLink(
$tmp_loc, $site[self::URL]);
$site[self::LOCATION][] = $tmp_loc;
$doc_info = [];
foreach (UrlParser::redirectTargets($site[self::URL],
$tmp_loc) as $target) {
$doc_info[self::LINKS][$target] =
"location:" . $site[self::URL];
}
$doc_info[self::LOCATION] = true;
$doc_info[self::DESCRIPTION] = $site[self::URL] ." => ".
$tmp_loc;
$doc_info[self::PAGE] = $doc_info[self::DESCRIPTION];
$doc_info[self::TITLE] = $site[self::URL];
$text_data = true;
if (!isset($site[self::ENCODING])) {
$site[self::ENCODING] = "UTF-8";
}
$handled = true;
} else if ($url_words_only) {
$doc_info = $this->urlWordsDocInfo($site[self::URL]);
$text_data = true;
$site[self::ENCODING] = "UTF-8";
$handled = true;
}
$host = UrlParser::getHost($site[self::URL]);
$fast = $this->usesFastProcessor($host, time());
if (!$handled) {
$processor = $this->pageProcessor($type, $fast);
if (!$processor) {
L\crawlLog("No page processor for mime type: ".$type);
L\crawlLog("Not processing: " . $site[self::URL]);
continue;
}
$text_data = $processor->text_data;
}
$original_links = [];
if (isset($site[self::PAGE]) && !$handled) {
if ($text_data) {
L\convertUtf8IfNeeded($site, self::PAGE, self::ENCODING,
C\NS_LIB . "crawlLog");
}
$page_processor = get_class($processor);
L\crawlLog(" Using Processor..." . substr($page_processor,
strlen(C\NS_PROCESSORS)) . ($fast ? " (fast)" : "") .
" for " . $site[self::URL]);
if (isset($site[self::REPOSITORY_TYPE]) &&
$site[self::REPOSITORY_TYPE] == self::REPOSITORY_GIT) {
$tmp_url_store = $site[self::URL];
$site[self::URL] = $site[self::FILE_NAME];
}
if (L\generalIsA($page_processor, C\NS_PROCESSORS.
"HtmlProcessor")) {
$processor->scrapers = $this->scrapers;
}
if (in_array($site[self::TYPE], ['text/html', 'text/plain',
'application/xhtml+xml', 'application/xml', 'text/xml'] )) {
if (!mb_check_encoding($site[self::PAGE], "UTF-8")) {
$site[self::PAGE] = L\convertToUtf8(
$site[self::PAGE], $site[self::ENCODING] ?? "");
}
}
$page = $site[self::PAGE];
$empty_image = false;
if (L\generalIsA($page_processor, C\NS_PROCESSORS.
"ImageProcessor")) {
if (!empty($site[self::CONTENT_SIZE]) &&
!empty($site[self::SIZE]) && $site[self::CONTENT_SIZE] >
$site[self::SIZE]) {
$page = "";
$empty_image = true;
}
}
if ($empty_image) {
$doc_info = null;
} else {
file_put_contents($this->pageInProcessName(),
$site[self::URL]);
$handle_start = microtime(true);
$doc_info = $processor->handle($page,
$site[self::URL]);
$processor_name = substr(strrchr($page_processor, "\\"),
1);
$processor_times[$processor_name][0] =
($processor_times[$processor_name][0] ?? 0) +
microtime(true) - $handle_start;
$processor_times[$processor_name][1] =
($processor_times[$processor_name][1] ?? 0) + 1;
$handle_time = microtime(true) - $handle_start;
$document_type = UrlParser::getDocumentType(
$site[self::URL]);
$this->process_times[$document_type][0] =
($this->process_times[$document_type][0] ?? 0) +
$handle_time;
$this->process_times[$document_type][1] =
($this->process_times[$document_type][1] ?? 0) + 1;
if ($handle_time > self::SLOW_PAGE_SECONDS) {
if ($this->noteSlowPage($host, time()) ==
self::SLOW_PAGES_BEFORE_FAST) {
L\crawlLog(" $host gave " .
self::SLOW_PAGES_BEFORE_FAST . " slow " .
"pages this hour, so its pages get the " .
"fast processor for the rest of it.");
}
$steps = "";
foreach ($processor->step_times ?? [] as
$step => $seconds) {
$steps .= (($steps == "") ? "; steps " : ", ") .
"$step " . round($seconds, 2) . "s";
}
/* The language and length say whether a slow
summary came from the page's size or from how
its language is split into terms. */
L\crawlLog(" Slow page: $processor_name took " .
round($handle_time, 2) . "s on " .
$site[self::URL] . "; language " .
((is_array($doc_info) ? $doc_info[self::LANG] ??
"" : "") ?: "none") . ", " .
mb_strlen((string)$page) . " characters" .
$steps);
}
}
if (C\FETCHER_PROCESS_DELAY > 0 ) {
usleep(C\FETCHER_PROCESS_DELAY);
}
if (isset($site[self::REPOSITORY_TYPE]) &&
$site[self::REPOSITORY_TYPE] == self::REPOSITORY_GIT) {
$site[self::URL] = $tmp_url_store;
}
if (!$doc_info) {
L\crawlLog(" Processing Yielded No Data For: ".
$site[self::URL]);
}
if ($page_processor != C\NS_PROCESSORS . "RobotProcessor" &&
!isset($doc_info[self::JUST_METAS])) {
if (!empty($doc_info[self::LINKS])) {
$links_start = microtime(true);
$site[self::CLD_IN_COMMON] = UrlParser::
countCompanyLevelDomainsInCommonDetectFarm(
$site[self::URL], array_keys(
$doc_info[self::LINKS]));
if ($site[self::CLD_IN_COMMON] < 0 &&
!C\nsdefined("KEEP_SPAM_DOMAINS") ) {
$doc_info[self::LINKS] = [];
} else {
$this->pruneLinks($doc_info, self::LINKS,
$start_time);
}
$link_time += microtime(true) - $links_start;
}
}
} else if (!$handled) {
$doc_info = false;
}
if (empty($site[self::CLD_IN_COMMON])) {
$site[self::CLD_IN_COMMON] = 0;
}
/* A home page whose server answered but gave no text, in a
crawl that allows landing pages, is still indexed, by the
words of its url. A host that never answered is not, and a
page sent back to be fetched again waits for the
Scheduler's last try. */
if (is_numeric($response_code) && $response_code >= 100 &&
(!$sent_back || $last_try) &&
empty($doc_info[self::LOCATION]) &&
trim($doc_info[self::DESCRIPTION] ?? "") == "" &&
$this->indexesUrlWords($site[self::URL])) {
$doc_info = $this->urlWordsDocInfo($site[self::URL],
$doc_info ?: []);
}
$not_loc = true;
if ($doc_info) {
$site[self::DOC_INFO] = $doc_info;
if (isset($doc_info[self::LOCATION])) {
$site[self::HASH] = L\crawlHash(
L\crawlHash($site[self::URL], true) . "LOCATION", true);
$not_loc = false;
}
$site[self::ROBOT_INSTANCE] = $prefix . C\p('ROBOT_INSTANCE');
if (!is_dir(C\FETCHER_CACHE_DIR)) {
mkdir(C\FETCHER_CACHE_DIR);
$htaccess = "Options None\nphp_flag engine off\n";
file_put_contents(C\FETCHER_CACHE_DIR . "/.htaccess",
$htaccess);
}
if ($type == "text/robot" &&
isset($doc_info[self::PAGE])) {
$site[self::PAGE] = $doc_info[self::PAGE];
}
if ($text_data) {
if (isset($doc_info[self::PAGE])) {
$site[self::PAGE] = $doc_info[self::PAGE];
} else {
$site[self::PAGE] = null;
}
if ($not_loc) {
$content = $doc_info[self::DESCRIPTION] ?? "";
$site[self::HASH] = FetchUrl::computePageHash(
$content, "text", $site[self::URL]);
}
} else {
$site[self::HASH] = FetchUrl::computePageHash(
$site[self::PAGE]);
}
if (isset($doc_info[self::WORD_CLOUD])) {
$site[self::WORD_CLOUD] = $doc_info[self::WORD_CLOUD];
} else {
$site[self::WORD_CLOUD] = null;
}
if (isset($doc_info[self::CRAWL_DELAY])) {
$site[self::CRAWL_DELAY] = $doc_info[self::CRAWL_DELAY];
}
if (isset($doc_info[self::ROBOT_PATHS]) && !$was_error) {
$site[self::ROBOT_PATHS] = $doc_info[self::ROBOT_PATHS];
}
if (!isset($site[self::ROBOT_METAS])) {
$site[self::ROBOT_METAS] = [];
}
if (isset($doc_info[self::ROBOT_METAS])) {
$site[self::ROBOT_METAS] = array_merge(
$site[self::ROBOT_METAS], $doc_info[self::ROBOT_METAS]);
}
if ($this->robots_txt == C\IGNORE_ROBOTS) {
$site[self::ROBOT_METAS] = [];
}
if ($this->robots_txt == C\ALLOW_LANDING_ROBOTS) {
$site[self::ROBOT_METAS] = array_diff(
$site[self::ROBOT_METAS], ["NOINDEX", "NONE"]);
}
//here's where we enforce NOFOLLOW
if (in_array("NOFOLLOW", $site[self::ROBOT_METAS]) ||
in_array("NONE", $site[self::ROBOT_METAS])) {
$site[self::DOC_INFO][self::LINKS] = [];
}
if (isset($doc_info[self::AGENT_LIST])) {
$site[self::AGENT_LIST] = $doc_info[self::AGENT_LIST];
}
if (isset($doc_info[self::IS_FEED])) {
$site[self::IS_FEED] = $doc_info[self::IS_FEED];
}
if (!empty($doc_info["ALT_DURATION"]) && empty(
$site[self::DURATION])) {
$duration = $doc_info["ALT_DURATION"];
if (preg_match("/PT(\d+)M(\d+)S/", $duration, $matches)) {
$minutes = intval($matches[1]) ?? 0;
$seconds = intval($matches[2]) ?? 0;
$duration = $minutes * C\ONE_MINUTE +
$seconds;
}
$site[self::DURATION] = $duration;
}
$pub_date = $site[self::PUBDATE] ??
($doc_info["ALT_PUBDATE"] ?? 0);
if (!is_numeric($pub_date)) {
$pub_date = strtotime($pub_date);
}
$site[self::PUBDATE] = intval($pub_date);
$this->copySiteFields($i, $site, $summarized_site_pages,
[self::SUBDOCS, 0]);
$summarized_site_pages[$i][self::URL] =
strip_tags($site[self::URL]);
if (isset($site[self::REPOSITORY_TYPE]) &&
$site[self::REPOSITORY_TYPE] == self::REPOSITORY_GIT) {
$summarized_site_pages[$i][self::TITLE] =
$site[self::FILE_NAME];
} else if (isset($site[self::DOC_INFO][self::TITLE])) {
$summarized_site_pages[$i][self::TITLE] = strip_tags(
$site[self::DOC_INFO][self::TITLE]);
// stripping html to be on the safe side
} else {
$summarized_site_pages[$i][self::TITLE] =
strip_tags($site[self::URL]);
}
if (!isset($site[self::REPOSITORY_TYPE])) {
if ($was_robot_error) {
$site[self::DOC_INFO][self::DESCRIPTION] =
"There was an HTTP error in trying to download ".
"this robots.txt file, so all paths to this site ".
"were disallowed by Yioop.\n".
$site[self::DOC_INFO][self::DESCRIPTION];
}
$summarized_site_pages[$i][self::DESCRIPTION] = strip_tags(
$site[self::DOC_INFO][self::DESCRIPTION] ?? "");
} else {
$summarized_site_pages[$i][self::DESCRIPTION] =
$site[self::DOC_INFO][self::DESCRIPTION] ?? "";
}
if (isset($site[self::DOC_INFO][self::JUST_METAS]) ||
isset($site[self::ROBOT_PATHS])) {
$summarized_site_pages[$i][self::JUST_METAS] = true;
}
if (isset($site[self::DOC_INFO][self::META_WORDS])) {
if (!isset($summarized_site_pages[$i][self::META_WORDS])) {
$summarized_site_pages[$i][self::META_WORDS] =
$site[self::DOC_INFO][self::META_WORDS];
} else {
$summarized_site_pages[$i][self::META_WORDS] =
array_merge(
$summarized_site_pages[$i][self::META_WORDS],
$site[self::DOC_INFO][self::META_WORDS]);
}
}
if (isset($site[self::DOC_INFO][self::LANG])) {
if ($site[self::DOC_INFO][self::LANG] == 'en' &&
$site[self::ENCODING] != "UTF-8") {
$site[self::DOC_INFO][self::LANG] =
L\guessLangEncoding($site[self::ENCODING]);
}
$summarized_site_pages[$i][self::LANG] =
$site[self::DOC_INFO][self::LANG];
}
if (isset($site[self::DOC_INFO][self::LINKS])) {
$summarized_site_pages[$i][self::LINKS] =
$site[self::DOC_INFO][self::LINKS];
}
if (isset($site[self::DOC_INFO][self::SUBDOCS])) {
$this->processSubdocs($i, $site, $summarized_site_pages);
}
if (isset($summarized_site_pages[$i][self::LINKS])) {
$summarized_site_pages[$i][self::LINKS] =
UrlParser::cleanRedundantLinks(
$summarized_site_pages[$i][self::LINKS],
$summarized_site_pages[$i][self::URL]);
}
if (!empty($this->classifiers)) {
Classifier::labelPage($summarized_site_pages[$i],
$this->classifiers, $this->active_classifiers,
$this->active_rankers);
}
if ($this->page_rule_parser != null) {
$this->page_rule_parser->executeRuleTrees(
$summarized_site_pages[$i]);
}
if (!empty($summarized_site_pages[$i][self::THUMB_URL]) &&
empty($summarized_site_pages[$i][self::THUMB])) {
$found_thumb_urls = true;
}
$metas = isset($summarized_site_pages[$i][self::ROBOT_METAS]) ?
$summarized_site_pages[$i][self::ROBOT_METAS] : [];
if (array_intersect($metas,
["NOARCHIVE", "NOINDEX", "JUSTFOLLOW", "NONE"]) != []) {
$summarized_site_pages[$i][self::PAGE] = "";
} else if (!empty($summarized_site_pages[$i][self::PAGE])) {
$num_cached_pages++;
}
$i++;
}
} // end for
$num_pages = count($summarized_site_pages);
if ($num_pages > 0) {
$this->crawl_stat_info["NUM_SUMMARIZED_PAGES"] += $num_pages;
$this->crawl_stat_info["NUM_CACHED_PAGES"] += $num_cached_pages;
file_put_contents($this->crawl_stat_filename,
serialize($this->crawl_stat_info));
}
for ($i = 0; $i < $num_pages; $i++) {
$summarized_site_pages[$i][self::INDEX] = $num_items + $i;
}
$thumbs_start = microtime(true);
if (file_exists($this->pageInProcessName())) {
unlink($this->pageInProcessName());
}
if ($this->crawl_type == self::WEB_CRAWL && $found_thumb_urls) {
$this->getPageThumbs($summarized_site_pages);
}
/* The machine's load average says whether the processing time
rose because other processes on the machine took the cores.
Windows gives no load average, so there it is left out. */
$load = function_exists("sys_getloadavg") ? sys_getloadavg() : false;
L\crawlLog(" Process pages time: ".L\changeInMicrotime($start_time).
" Current Memory: " . memory_get_usage() .
(is_array($load) ? " Load average: " . implode(" ", $load) : ""));
/* How the processing time divides, so a slow crawl's log says
which processor, plugin or thumbnail step to look at. */
$split = " Processing time by part:";
$rest_time = microtime(true) - $start_time;
foreach ($processor_times ?? [] as $name => $time_and_count) {
$split .= " $name " . round($time_and_count[0], 2) . "s for " .
$time_and_count[1] . " pages;";
$rest_time -= $time_and_count[0];
}
foreach (PageProcessor::$plugin_times as $name => $time) {
$split .= " " . substr(strrchr($name, "\\"), 1) . " " .
round($time, 2) . "s (inside its processor's time);";
}
$thumbs_time = microtime(true) - $thumbs_start;
/* Text read from pictures is part of its processor's time. The
next batch's download passes and the link checks are named
apart; the rest is what nothing named took, such as hashing
pages, running classifiers and page rules. */
L\crawlLog($split . " ocr " . round(ComputerVision::$ocr_time, 2) .
"s (inside its processor's time); thumbnails " .
round($thumbs_time, 2) . "s; next batch's download " .
round($download_pass_time, 2) . "s; links " .
round($link_time, 2) . "s; rest " . round($rest_time -
$thumbs_time - $download_pass_time - $link_time, 2) . "s");
PageProcessor::$plugin_times = [];
ComputerVision::$ocr_time = 0;
return $summarized_site_pages;
}
/**
* Adds thumbs for websites with a self::THUMB_URL field by downloading the
* linked to images and making a thumb from it.
*
* @param array &$sites associative array of web sites information to add
* thumbs for. At least one site in the array should have a
* self::THUMB_URL field that we want have the thumb of
*/
public function getPageThumbs(&$sites)
{
L\crawlLog("Getting Thumbs for Downloaded Pages...");
$num_sites = count($sites);
$thumb_sites = [];
for ($i = 0; $i < $num_sites; $i++) {
$thumb_sites[$i] = [];
if (!empty($sites[$i][self::THUMB_URL])
&& empty($sites[$i][self::THUMB])) {
$thumb_sites[$i][self::THUMB_URL] = $sites[$i][self::THUMB_URL];
}
}
$thumb_sites = FetchUrl::getPages($thumb_sites, false,
$this->page_range_request, null, self::THUMB_URL, self::DATA, true);
for ($i = 0; $i < $num_sites; $i++) {
if (!empty($thumb_sites[$i][self::DATA])) {
set_error_handler(null);
$image = @imagecreatefromstring($thumb_sites[$i][self::DATA]);
$thumb_string = ImageProcessor::createThumb($image);
restore_error_handler();
if ($thumb_string) {
$sites[$i][self::THUMB] = 'data:image/webp;base64,' .
base64_encode($thumb_string);
}
}
}
}
/**
* Used to remove from the to_crawl urls those that are no longer crawlable
* because the allowed and disallowed sites have changed.
*/
public function cullNoncrawlableSites()
{
$count = count($this->to_crawl);
$count_again = count($this->to_crawl_again);
L\crawlLog("Culling noncrawlable urls after change in crawl ".
"parameters; To Crawl Count: $count, To Crawl Again: ".
"$count_again");
$start_time = microtime(true);
$k = 0;
for ($i = 0; $i < $count; $i++) {
L\crawlTimeoutLog("..still culling to crawl urls. Examining ".
"location %s in queue of %s.", $i, $count);
if (empty($this->to_crawl[$i][0])) {
unset($this->to_crawl[$i]);
$k++;
} else {
list($url, ) = $this->to_crawl[$i];
if (!$this->allowedToCrawlSite($url) ||
$this->disallowedToCrawlSite($url)) {
unset($this->to_crawl[$i]);
$k++;
}
}
}
$this->allow_disallow_cache_time = microtime(true);
for ($i = 0; $i < $count_again; $i++) {
L\crawlTimeoutLog("..still culling to crawl again urls. Examining ".
"location %s in queue of %s.", $i, $count);
if (empty($this->to_crawl_again[$i][0])) {
unset($this->to_crawl_again[$i]);
$k++;
} else {
list($url, ) = $this->to_crawl_again[$i];
if (!$this->allowedToCrawlSite($url) ||
$this->disallowedToCrawlSite($url)) {
unset($this->to_crawl_again[$i]);
$k++;
}
}
}
L\crawlLog("...Removed $k cullable URLS from to crawl lists in time: ".
L\changeInMicrotime($start_time));
}
/**
* Checks if url belongs to a list of sites that are allowed to be
* crawled and that the file type is crawlable
*
* @param string $url url to check
* @return bool whether is allowed to be crawled or not
*/
public function allowedToCrawlSite($url)
{
$doc_type = UrlParser::getDocumentType($url);
if (!in_array($doc_type, $this->all_file_types)) {
$doc_type = "unknown";
}
if (!in_array($doc_type, $this->indexed_file_types)) {
return false;
}
if ($this->restrict_sites_by_url) {
return UrlParser::urlMemberSiteArray($url, $this->allowed_sites,
"a" . $this->allow_disallow_cache_time);
}
return true;
}
/**
* Checks if url belongs to a list of sites that aren't supposed to be
* crawled
*
* @param string $url url to check
* @return bool whether is shouldn't be crawled
*/
public function disallowedToCrawlSite($url)
{
return UrlParser::urlMemberSiteArray($url, $this->disallowed_sites,
"d" . $this->allow_disallow_cache_time);
}
/**
* This method attempts to cull from the doc_info struct the
* best $this->max_links_to_extract. Currently, this is done by first
* removing links of filetype or sites the crawler is forbidden from crawl.
* Then a crude estimate of the information contained in the links test:
* strlen(gzip(text)) is used to extract the best remaining links.
*
* @param array &$doc_info a string with a CrawlConstants::LINKS subarray
* This subarray in turn contains url => text pairs.
* @param string $field field for links default is CrawlConstants::LINKS
* @param int $member_cache_time says how long allowed and disallowed url
* info should be caches by urlMemberSiteArray
*/
public function pruneLinks(&$doc_info, $field = CrawlConstants::LINKS,
$member_cache_time = 0)
{
if (!isset($doc_info[$field])) {
return;
}
$links = [];
$allowed_name = "a" . $member_cache_time;
$disallowed_name = "d" . $member_cache_time;
if (!empty($this->domain_filters)) {
$filters = array_column($this->domain_filters, "FILTER");
}
foreach ($doc_info[$field] as $url => $url_info) {
$filename = UrlParser::getDocumentFilename($url);
if (substr(UrlParser::guessMimeTypeFromFileName($filename), 0, 5)
!= "image" && !empty($filters) &&
UrlParser::cullByDomainFilter($url, $filters)) {
L\crawlLog("$url not in domain filter, removing link!");
continue;
}
$doc_type = UrlParser::getDocumentType($url);
if (!in_array($doc_type, $this->all_file_types)) {
$doc_type = "unknown";
$url_parts = explode("#", $url);
$url = $url_parts[0];
$fragment = $url_parts[1] ?? "";
if ($fragment == "sitemap") {
$doc_type = "xml";
}
}
if (!in_array($doc_type, $this->indexed_file_types)) {
continue;
}
if ($this->restrict_sites_by_url) {
if (!UrlParser::urlMemberSiteArray($url, $this->allowed_sites,
$allowed_name)) {
continue;
}
}
if (UrlParser::urlMemberSiteArray($url, $this->disallowed_sites,
$disallowed_name)) {
continue;
}
$links[$url] = $url_info;
}
$doc_info[$field] = UrlParser::pruneLinks($links,
$this->max_links_to_extract);
}
/**
* Copies fields from the array of site data to the $i indexed
* element of the $summarized_site_pages. This will flatten info
* in a DOC_INFO subarray. This method is both useful to prepare downloaded
* site info and to prepare meta info for sub docs that might be produced
* by an indexing puglin.
*
* @param int $i index to copy to
* @param array $site web page info to copy
* @param array $exclude_fields an array of fields not to copy
* @param array &$summarized_site_pages array of summaries of web pages
*/
public function copySiteFields($i, $site, &$summarized_site_pages,
$exclude_fields = [])
{
$summarized_site_pages[$i] ??= [];
foreach ($site as $field => $value) {
if ($field == self::DOC_INFO && is_array($value)) {
foreach ($value as $sub_field => $sub_value) {
if (!isset($summarized_site_pages[$i][$sub_field])
&& !in_array($sub_field, $exclude_fields)) {
$summarized_site_pages[$i][$sub_field] =
$sub_value;
}
}
} else {
if (!in_array($field, $exclude_fields)) {
$summarized_site_pages[$i][$field] = $site[$field];
}
}
}
}
/**
* The pageProcessing method of an IndexingPlugin generates
* a self::SUBDOCS array of additional "micro-documents" that
* might have been in the page. This methods adds these
* documents to the summaried_size_pages and stored_site_pages
* arrays constructed during the execution of processFetchPages()
*
* @param int &$i index to begin adding subdocs at
* @param array $site web page that subdocs were from and from
* which some subdoc summary info is copied
* @param array &$summarized_site_pages array of summaries of web pages
*/
public function processSubdocs(&$i, $site, &$summarized_site_pages)
{
$subdocs = $site[self::DOC_INFO][self::SUBDOCS];
foreach ($subdocs as $subdoc) {
$i++;
$this->copySiteFields($i, $site, $summarized_site_pages,
[self::SUBDOCS, 0]);
$summarized_site_pages[$i][self::URL] =
strip_tags($site[self::URL]);
$summarized_site_pages[$i][self::TITLE] =
strip_tags($subdoc[self::TITLE]);
$summarized_site_pages[$i][self::DESCRIPTION] =
strip_tags($subdoc[self::DESCRIPTION]);
if (isset($site[self::JUST_METAS])) {
$summarized_site_pages[$i][self::JUST_METAS] = true;
}
if (isset($subdoc[self::LANG])) {
$summarized_site_pages[$i][self::LANG] =
$subdoc[self::LANG];
}
if (isset($subdoc[self::LINKS])) {
$summarized_site_pages[$i][self::LINKS] =
$subdoc[self::LINKS];
}
if (isset($subdoc[self::SUBDOCTYPE])) {
$summarized_site_pages[$i][self::SUBDOCTYPE] =
$subdoc[self::SUBDOCTYPE];
}
}
}
/**
* Updates the $this->found_sites array with data from the most recently
* downloaded sites. This means updating the following sub arrays:
* the self::ROBOT_PATHS, self::TO_CRAWL. It checks if there are still
* more urls to crawl. If so, a mini index is built and,
* the queue server is called with the data.
*
* @param array $sites site data to use for the update
* @param bool $force_send whether to force send data back to queue_server
* or rely on usual thresholds before sending
*/
public function updateFoundSites($sites, $force_send = false)
{
$start_time = microtime(true);
L\crawlLog(" Updating Found Sites Array...");
$this->fetched_since_send += count($sites);
for ($i = 0; $i < count($sites); $i++) {
$site = $sites[$i];
if (!isset($site[self::URL])) {
continue;
}
$host = UrlParser::getHost($site[self::URL]);
if (isset($site[self::ROBOT_PATHS])) {
if ($site[self::IP_ADDRESSES] == ["0.0.0.0"]
&& !str_contains($host, "localhost")) {
/* probably couldn't find site so this will block
from crawl
*/
$site[self::ROBOT_PATHS][self::DISALLOWED_SITES] =
["/"];
}
/* set same robots.txt for all redirects went through
FetchURL::getPages always follows redirects for
robots.txt files
*/
$locations = [$host];
if (!empty($site[self::LOCATION]) &&
is_array($site[self::LOCATION])) {
$locations = array_merge($locations, $site[self::LOCATION]);
}
foreach ($locations as $location) {
$h = UrlParser::getHost($location);
$this->found_sites[self::ROBOT_TXT][$h][self::IP_ADDRESSES]=
$site[self::IP_ADDRESSES];
$this->found_sites[self::ROBOT_TXT][$h][self::ROBOT_PATHS] =
$site[self::ROBOT_PATHS];
if (isset($site[self::CRAWL_DELAY])) {
$this->found_sites[self::ROBOT_TXT][$h][
self::CRAWL_DELAY] = $site[self::CRAWL_DELAY];
}
}
if (isset($site[self::LINKS])
&& $this->crawl_type == self::WEB_CRAWL) {
$num_links = count($site[self::LINKS]);
//robots pages might have sitemaps links on them
//which we want to crawl
$link_urls = array_values($site[self::LINKS]);
$this->addToCrawlSites($link_urls, $site[self::URL],
$site[self::WEIGHT], $site[self::DOC_DEPTH],
$site[self::CLD_IN_COMMON]);
}
$this->found_sites[self::SEEN_URLS][] = $site;
} else {
$this->found_sites[self::SEEN_URLS][] = $site;
if (isset($site[self::LINKS])
&& $this->crawl_type == self::WEB_CRAWL) {
$link_urls = array_keys($site[self::LINKS]);
$this->addToCrawlSites($link_urls, $site[self::URL],
$site[self::WEIGHT], $site[self::DOC_DEPTH] ?? 0,
$site[self::CLD_IN_COMMON] ?? 0);
}
} //end else
//Add cache page validation data
if (isset($site[self::CACHE_PAGE_VALIDATORS])) {
$this->found_sites[self::CACHE_PAGE_VALIDATION_DATA][] =
[$site[self::URL], $site[self::CACHE_PAGE_VALIDATORS]];
}
if (isset($this->hosts_with_errors[$host]) &&
$this->hosts_with_errors[$host] > C\DOWNLOAD_ERROR_THRESHOLD) {
$this->found_sites[self::ROBOT_TXT][$host][
self::CRAWL_DELAY] = C\ERROR_CRAWL_DELAY;
L\crawlLog("setting crawl delay $host");
}
if (isset($this->found_sites[self::TO_CRAWL])) {
$this->found_sites[self::TO_CRAWL] =
array_filter($this->found_sites[self::TO_CRAWL]);
}
if (isset($site[self::INDEX])) {
$site_index = $site[self::INDEX];
} else {
$site_index = "[LINK]";
}
$subdoc_info = "";
if (isset($site[self::SUBDOCTYPE])) {
$subdoc_info = "(Subdoc: {$site[self::SUBDOCTYPE]})";
}
/* for log file get rid of non-utf-8 characters
that latter make it hard to view the log
*/
L\crawlLog($site_index . ". $subdoc_info ".
iconv("UTF-8", "ISO-8859-1//IGNORE", $site[self::URL]));
} // end for
L\crawlLog(" Done Update Found Sites Array Time ".
L\changeInMicrotime($start_time));
$most_before_send = ($this->crawl_type == self::SYNTHETIC_CRAWL) ?
self::SYNTHETIC_SEEN_URLS_BEFORE_SEND : self::SEEN_URLS_BEFORE_SEND;
if ($force_send || ($this->crawl_type == self::WEB_CRAWL &&
count($this->to_crawl) <= 0 && count($this->to_crawl_again) <= 0 &&
$this->prefetch === null) ||
($this->archive_iterator &&
$this->archive_iterator->end_of_iterator) ||
count($this->found_sites[self::SEEN_URLS] ?? []) >=
$most_before_send || ($this->crawl_type ==
self::WEB_CRAWL && $this->fetched_since_send >= $this->send_after &&
!empty($this->found_sites[self::SEEN_URLS])) ||
($this->crawl_type == self::WEB_CRAWL &&
$this->schedule_received_time > 0 &&
$this->answered_sequence_number != $this->sequence_number &&
time() - $this->schedule_received_time >=
self::ANSWER_SCHEDULE_AFTER) ||
memory_get_usage() > L\memoryCeiling( C\MEMORY_FILL_FACTOR) ) {
$start_time = microtime(true);
/* An upload can take many seconds, and the next batch's
transfers would wait unattended through it until they
timed out, so that download is finished first. */
if ($this->prefetch !== null) {
$this->finishPrefetch();
}
L\crawlLog(" Start Update Server ");
$this->selectCurrentServerAndUpdateIfNeeded(true);
if ($this->crawl_type == self::WEB_CRAWL) {
$this->saveScheduleLeft();
}
$this->send_after = min(2 * $this->send_after,
$most_before_send);
$this->fetched_since_send = 0;
L\crawlLog(" Update Server Time " .
L\changeInMicrotime($start_time));
}
}
/**
* Used to add a set of links from a web page to the array of sites which
* need to be crawled.
*
* @param array $link_urls an array of urls to be crawled
* @param string $old_url url of page where links came from
* @param int $old_weight the weight on the page the link came from
* (order of importance among links on page)
* @param int $old_depth of the web page the links came from
* @param int $num_common number of company level domains in common between
* $link_urls and $old_url
*/
public function addToCrawlSites($link_urls, $old_url, $old_weight,
$old_depth, $num_common)
{
$num_links = count($link_urls);
$old_cld = UrlParser::getCompanyLevelDomain($old_url);
$old_url_scheme = UrlParser::getScheme($old_url);
if ($this->crawl_order != self::BREADTH_FIRST &&
$num_common < 0 && !C\nsdefined("KEEP_SPAM_DOMAINS")) {
/* see QueueServer crawl delay code. If a host has no crawl
delay then, the cld host is checked for a crawl delay and
that is taken as the crawl delay value
*/
$num_links = 0;
$this->found_sites[self::ROBOT_TXT][$old_url_scheme .
"://" . $old_cld][self::CRAWL_DELAY] = C\ERROR_CRAWL_DELAY;
L\crawlLog(
"$old_cld seems to be a link farm, crawl delaying!!!!");
}
if ($num_links == 0) {
return;
}
if ($this->max_depth >= 0 && $old_depth + 1 > $this->max_depth) {
return; // reached max depth so don't add any urls
}
$num_queue_servers = count($this->queue_servers);
for ($i = 0; $i < $num_links; $i++) {
$url = $link_urls[$i];
if (strlen($url) > 0) {
// explicitly get rid of any fragment component of url
$fragment_parts = explode("#", $url);
if (($fragment_parts[1] ?? "") != "sitemap") {
$url = $fragment_parts[0];
}
if (strlen($url) == 0) {
continue;
}
$part = L\calculatePartition($url, $num_queue_servers,
C\NS_LIB . "UrlParser::getHost");
$new_depth = $old_depth + 1;
if ($this->crawl_order == self::BREADTH_FIRST) {
$new_weight = 0;
} else { //host budgeting and default case
$new_weight = $i;
}
$this->found_sites[self::TO_CRAWL][$part][] =
[$url, $new_weight, $new_depth, $old_url, $old_weight];
}
}
}
/**
* Updates the queue_server about sites that have been crawled.
*
* This method is called if there are currently no more sites to crawl. It
* compresses and does a post request to send the page summary data, robot
* data, and to crawl url data back to the server. In the
* event that the server doesn't acknowledge it loops and tries again after
* a delay until the post is successful. At this point, memory for this data
* is freed.
*
* @param string $server index of queue server to update
* @param bool $send_robots whether to send robots.txt data if present
*/
public function updateScheduler($server, $send_robots = false)
{
$queue_server = $this->queue_servers[$server];
L\crawlLog("Updating machine: " . $queue_server);
$prefix = $this->which_fetcher . "-" . $this->channel . "-";
if (count($this->to_crawl) <= 0) {
$schedule_time = $this->schedule_time;
}
/*
In what follows as we generate post data we delete stuff
from $this->found_sites, to try to minimize our memory
footprint.
*/
$byte_counts = ["TOTAL" => 0, "ROBOT" => 0, "SCHEDULE" => 0,
"INDEX" => 0, "CACHE_PAGE_VALIDATION" => 0];
$post_data = ['c'=>'fetch', 'a'=>'update',
'crawl_time' => $this->crawl_time, 'machine_uri' => C\p('WEB_URI'),
'robot_instance' => $prefix . C\p('ROBOT_INSTANCE'), 'data' => '',
'check_crawl_time' => $this->check_crawl_time,
'crawl_type' => $this->crawl_type,
'downloads_per_hour' => intval($this->downloads_per_hour),
'diff_origin' => ($server != $this->current_server),
'sequence_number' => $this->sequence_number];
if ($this->crawl_type == self::ARCHIVE_CRAWL &&
!empty($this->arc_type)) {
$post_data['arc_type'] = $this->arc_type;
}
//handle robots.txt data
if (isset($this->found_sites[self::ROBOT_TXT]) && $send_robots) {
$data = L\webencode(
gzcompress(serialize($this->found_sites[self::ROBOT_TXT])));
unset($this->found_sites[self::ROBOT_TXT]);
$bytes_robot = strlen($data);
$post_data['data'] .= $data;
L\crawlLog("...$bytes_robot bytes of robot data");
$byte_counts["TOTAL"] += $bytes_robot;
$byte_counts["ROBOT"] = $bytes_robot;
}
//handle cache validation data
if (isset($this->found_sites[self::CACHE_PAGE_VALIDATION_DATA])) {
$cache_page_validation_data = L\webencode(
gzcompress(serialize(
$this->found_sites[self::CACHE_PAGE_VALIDATION_DATA])));
unset($this->found_sites[self::CACHE_PAGE_VALIDATION_DATA]);
$bytes_cache_page_validation = strlen($cache_page_validation_data);
$post_data['data'] .= $cache_page_validation_data;
L\crawlLog("..." . $bytes_cache_page_validation .
" bytes of cache page validation data");
$byte_counts["TOTAL"] += $bytes_cache_page_validation;
$byte_counts["CACHE_PAGE_VALIDATION"] =
$bytes_cache_page_validation;
}
//handle schedule data
$schedule_data = [];
if (isset($this->found_sites[self::TO_CRAWL][$server])) {
$schedule_data[self::TO_CRAWL] =
$this->found_sites[self::TO_CRAWL][$server];
}
unset($this->found_sites[self::TO_CRAWL][$server]);
if (!empty($this->found_sites[self::RETRY_URLS][$server])) {
$schedule_data[self::RETRY_URLS] =
$this->found_sites[self::RETRY_URLS][$server];
}
unset($this->found_sites[self::RETRY_URLS][$server]);
$seen_count = 0;
if (isset($this->found_sites[self::SEEN_URLS]) &&
($seen_count = count($this->found_sites[self::SEEN_URLS])) > 0 ) {
$hash_seen_urls = [];
foreach ($this->found_sites[self::SEEN_URLS] as $site) {
$hash_seen_urls[] =
L\crawlHash($site[self::URL], true);
}
$schedule_data[self::HASH_SEEN_URLS] = $hash_seen_urls;
unset($hash_seen_urls);
}
if (!empty($schedule_data) && !empty($this->process_times)) {
$schedule_data[self::PROCESS_TIMES] = $this->process_times;
$this->process_times = [];
}
if (!empty($schedule_data)) {
if (isset($schedule_time)) {
$schedule_data[self::SCHEDULE_TIME] = $schedule_time;
}
$data = L\webencode(gzcompress(serialize($schedule_data)));
$post_data['data'] .= $data;
$bytes_schedule = strlen($data);
L\crawlLog("..." . $bytes_schedule . " bytes of schedule data");
$byte_counts["TOTAL"] += $bytes_schedule;
$byte_counts["SCHEDULE"] = $bytes_schedule;
}
$byte_counts["SCHEDULE_TIME"] = $this->schedule_time;
unset($schedule_data);
if ($seen_count > 0 || !empty(
$this->found_sites[self::LINK_SEEN_URLS][$server])) {
$compress_urls = $this->compressAndUnsetSeenUrls($server);
$len_urls = strlen($compress_urls);
L\crawlLog("...Finish Compressing seen URLs.");
$out_string = L\packInt($len_urls) . $compress_urls;
unset($compress_urls);
L\garbageCollect();
$data = L\webencode($out_string);
L\crawlLog(".....web encode result.");
// don't compress index data
$post_data['data'] .= $data;
unset($out_string);
$bytes_index = strlen($data);
L\crawlLog("..." . $bytes_index . " bytes of index data");
$byte_counts["TOTAL"] += $bytes_index;
$byte_counts["INDEX"] = $bytes_index;
if ($this->crawl_type == self::ARCHIVE_CRAWL) {
$this->recrawl_check_scheduler = true;
}
}
if ($byte_counts["TOTAL"] <= 0) {
L\crawlLog("No data to send, sending an empty update so the " .
"queue server can close this batch...");
}
L\crawlLog("...");
//try to send to queue server
$this->uploadCrawlData($queue_server, $byte_counts, $post_data);
unset($post_data);
if ($server == $this->current_server) {
$this->answered_sequence_number = $this->sequence_number;
}
L\crawlLog("... Current Memory:" . memory_get_usage());
}
/**
* pageInProcessName gives the path of the file, named from
* PAGE_IN_PROCESS_NAME, this fetcher, its channel and its crawl, that
* holds the url of the page its processor is reading
*
* @return string the path of the file
*/
public function pageInProcessName()
{
return $this->schedules_folder . "/" . $this->which_fetcher . "-" .
$this->channel . "-" . self::PAGE_IN_PROCESS_NAME .
$this->crawl_time . ".txt";
}
/**
* scheduleLeftName gives the path of the file, named from
* SCHEDULE_LEFT_NAME, this fetcher, its channel and its crawl, that
* holds the urls of its schedule not yet sent back
*
* @return string the path of the file
*/
public function scheduleLeftName()
{
return $this->schedules_folder . "/" . $this->which_fetcher . "-" .
$this->channel . "-" . self::SCHEDULE_LEFT_NAME .
$this->crawl_time . ".txt";
}
/**
* saveScheduleLeft writes the urls of the current schedule whose pages
* have not been sent back to the queue server: those in to_crawl and
* to_crawl_again, and those the next batch's download (the prefetch)
* has already taken. checkScheduler calls it when a schedule arrives
* and updateFoundSites after each send, the times when nothing is
* held that a restart would lose, so a fetcher that stops while
* processing a page can go on from there. When none are left the file
* is removed.
*/
public function saveScheduleLeft()
{
$left = array_merge($this->to_crawl, $this->to_crawl_again);
if ($this->prefetch?->isTerminated()) {
foreach ($this->prefetch->getReturn() ?? [] as $site) {
if (is_array($site) && isset($site[self::URL])) {
$left[] = [$site[self::URL], $site[self::WEIGHT] ?? 0,
$site[self::DOC_DEPTH] ?? 0,
$site[self::CRAWL_DELAY] ?? 0];
}
}
}
$file_name = $this->scheduleLeftName();
if (empty($left)) {
if (file_exists($file_name)) {
unlink($file_name);
}
return;
}
file_put_contents($file_name, serialize([
self::SEQUENCE_NUMBER => $this->sequence_number,
self::TO_CRAWL => $left,
"RESEND_TIME" => ($this->answered_sequence_number ==
$this->sequence_number) ? PHP_INT_MAX :
$this->schedule_resend_time]));
}
/**
* resumeScheduleLeft takes back, when a fetcher starts, the urls of
* the schedule it had not sent back when it stopped, as
* saveScheduleLeft wrote them, and its sequence number. A page whose
* url PAGE_IN_PROCESS_NAME's file holds was being read when the
* fetcher stopped, so it is left out and logged: a page that ends the
* process, as two lobbying spreadsheets did in the crawl of 18
* October, would otherwise stop each fetcher the schedule went to.
* The fetcher then goes on with the schedule at once, where the queue
* server would have waited out the schedule's timeout before sending
* it to another fetcher. A schedule the queue server has sent to
* another fetcher by now, one past its resend time and not yet
* answered, is left to that fetcher.
*/
public function resumeScheduleLeft()
{
$stopping_url = "";
if (file_exists($this->pageInProcessName())) {
$stopping_url = file_get_contents($this->pageInProcessName());
unlink($this->pageInProcessName());
L\crawlLog("This fetcher stopped while processing " .
"$stopping_url, so that page is left out");
}
$file_name = $this->scheduleLeftName();
if (!file_exists($file_name)) {
return;
}
$saved = unserialize(file_get_contents($file_name));
unlink($file_name);
if (empty($saved[self::TO_CRAWL])) {
return;
}
if (time() >= ($saved["RESEND_TIME"] ?? 0)) {
L\crawlLog("Schedule " . ($saved[self::SEQUENCE_NUMBER] ?? "") .
" has been sent to another fetcher by now, so it is left");
return;
}
$this->to_crawl = [];
foreach ($saved[self::TO_CRAWL] as $url_info) {
if (explode("###", $url_info[0])[0] !== $stopping_url) {
$this->to_crawl[] = $url_info;
}
}
$this->to_crawl_again = [];
$this->sequence_number = $saved[self::SEQUENCE_NUMBER];
$this->schedule_received_time = time();
L\crawlLog("Going on with " . count($this->to_crawl) .
" urls of schedule {$this->sequence_number} not yet sent back");
}
/**
* compressWaitingPage gzip encodes the page of the next site in
* $this->found_sites[self::SEEN_URLS] whose page is not yet encoded, at
* ARCHIVE_COMPRESSION_LEVEL, and keeps the bytes in the site under
* PRECOMPRESSED_PAGE, where packSeenSite takes them when the site is
* sent. FetchUrl::getPages calls it while it waits on downloads, so the
* encoding a send would do is done in time the fetcher would spend
* waiting.
*
* @return bool whether a page was encoded
*/
public function compressWaitingPage()
{
if (empty($this->found_sites[self::SEEN_URLS])) {
return false;
}
$seen = &$this->found_sites[self::SEEN_URLS];
$num_seen = count($seen);
while ($this->pages_compressed < $num_seen) {
$index = $this->pages_compressed++;
if (isset($seen[$index][self::PAGE]) &&
is_string($seen[$index][self::PAGE]) &&
!isset($seen[$index][self::PRECOMPRESSED_PAGE])) {
$seen[$index][self::PRECOMPRESSED_PAGE] = gzencode(
serialize($seen[$index][self::PAGE]),
C\ARCHIVE_COMPRESSION_LEVEL);
return true;
}
}
return false;
}
/**
* Computes a string of compressed urls from the seen urls and extracted
* links destined for the current queue server. Then unsets these
* values from $this->found_sites
*
* @param int $server index of queue server to compress and unset urls for
* @return string of compressed urls
*/
public function compressAndUnsetSeenUrls($server)
{
$num_queue_servers = count($this->queue_servers);
$compress_urls = "";
if (!isset($this->found_sites[self::LINK_SEEN_URLS][$server])) {
$this->found_sites[self::LINK_SEEN_URLS][$server] = [];
}
if (isset($this->found_sites[self::SEEN_URLS]) &&
is_array($this->found_sites[self::SEEN_URLS])) {
$this->found_sites[self::SEEN_URLS] =
array_merge($this->found_sites[self::SEEN_URLS],
$this->found_sites[self::LINK_SEEN_URLS][$server]);
} else {
$this->found_sites[self::SEEN_URLS] =
$this->found_sites[self::LINK_SEEN_URLS][$server];
}
$this->found_sites[self::LINK_SEEN_URLS][$server] = [];
if (isset($this->found_sites[self::SEEN_URLS])) {
$num_seen = count($this->found_sites[self::SEEN_URLS]);
$i = 0;
/* Sites are taken off the end of a reversed list, so each is
freed as it is packed and the rest are not moved. Taking
each off the front moves all the others every time, which
costs seconds on a list of tens of thousands of links. */
$this->found_sites[self::SEEN_URLS] =
array_reverse($this->found_sites[self::SEEN_URLS]);
while($this->found_sites[self::SEEN_URLS] != []) {
L\crawlTimeoutLog("..compressing seen url %s of %s",
$i, $num_seen);
$i++;
$site = array_pop($this->found_sites[self::SEEN_URLS]);
if (empty($site[self::LANG])) {
$lang = L\guessLocaleFromString($site[self::DESCRIPTION]);
} else {
$lang = $site[self::LANG];
}
$not_link = (empty($site[self::TYPE]) ||
$site[self::TYPE] != 'link');
/* Only the question answers of a page's title and summary
are sent; the queue server's index works out the
page's terms itself. A link gets no answers. */
$question_answers = ($not_link) ?
(PhraseParser::extractPhrasesInLists(
($site[self::TITLE] ?? "") . " " .
($site[self::DESCRIPTION] ?? ""), $lang, true)[
'QUESTION_ANSWER_LIST'] ?? []) : [];
if (!empty($question_answers)) {
$site[self::QUESTION_ANSWERS] = $question_answers;
}
if ($this->crawl_type == self::ARCHIVE_CRAWL &&
isset($this->archive_iterator)) {
$site[self::DOC_RANK] =
$this->archive_iterator->weight($site);
}
if (!isset($site[self::ROBOT_METAS]) ||
!in_array("JUSTFOLLOW", $site[self::ROBOT_METAS]) ||
isset($site[self::JUST_METAS])) {
/* The page is compressed here at the level the
index's archive keeps it on disk, so the queue
server stores the bytes as they come. */
$site_string = L\packSeenSite($site,
C\ARCHIVE_COMPRESSION_LEVEL,
self::SEEN_URL_COMPRESSION_LEVEL);
$compress_urls .= L\packInt(strlen($site_string)) .
$site_string;
} else {
L\crawlLog("..filtering " . $site[self::URL] .
" because of JUSTFOLLOW.");
continue;
}
$site_url = $site[self::URL];
$host = UrlParser::getHost($site_url);
// extract link seen urls for this if not a link
if (!$this->no_process_links && (empty($site[self::TYPE]) ||
$site[self::TYPE] != 'link') &&
!isset($site[self::JUST_METAS]) &&
!isset($this->programming_language_extension[$lang])) {
$tokenizer = PhraseParser::getTokenizer($lang);
$has_stopwords_remover = !empty($tokenizer) &&
method_exists($tokenizer, "stopwordsRemover");
$is_feed_link = (!empty($site[self::IS_FEED])) ?
true : false;
foreach ($site[self::LINKS] as $url => $url_info) {
/* this mysterious check means won't index links from
robots.txt. Sitemap will still be in TO_CRAWL,
but that's done elsewhere
*/
L\crawlTimeoutLog("..storing link info for seen url".
" %s of %s", $i, $num_seen);
if (strlen($url) == 0 || is_numeric($url)) {
continue;
}
$link_host = UrlParser::getHost($url);
if (strlen($link_host) == 0) {
continue;
}
if (is_array($url_info)) {
$link_title = (empty($url_info[self::TITLE])) ?
$url : $url_info[self::TITLE];
$link_text = (empty($url_info[self::DESCRIPTION])) ?
UrlParser::extractTextFromUrl($url) :
$url_info[self::DESCRIPTION];
} else {
$link_text = $url_info;
$link_title = $url;
$url_info = [
self::TITLE => $link_title,
self::DESCRIPTION => $link_text,
];
}
$part_num = L\calculatePartition($link_host,
$num_queue_servers);
$link_type = UrlParser::getDocumentType($url);
$is_image_link = (in_array($link_type,
PageProcessor::$image_types));
$summary = [];
$this->found_sites[self::LINK_SEEN_URLS][$part_num] ??=
[];
$elink_flag = ($link_host != $host) ? true : false;
$link_text = strip_tags($link_text);
if ($has_stopwords_remover) {
$useful_text = $tokenizer->stopwordsRemover(
$link_text);
} else {
$useful_text = $link_text;
}
$useful_text = preg_replace("/\.\.|\s/u", "",
$useful_text ?? "");
if (mb_strlen($useful_text) < C\MIN_LINKS_TEXT_CHARS) {
continue;
}
$ref = ($elink_flag) ? "eref" : "iref";
$url = str_replace('|', "%7C", $url);
$link_id = "url|" . $url . "|text|" .
urlencode($link_text) . "|$ref|" . $site_url;
$elink_flag_string = ($elink_flag) ? "e" : "i";
$link_keys = L\crawlHash($url, true) .
L\crawlHash($link_id, true) .
$elink_flag_string .
substr(L\crawlHash($host . "/", true), 1);
$summary[self::URL] = $link_id;
$summary[self::TITLE] = $link_title;
// stripping html to be on the safe side
$summary[self::DESCRIPTION] = $link_text;
$summary[self::TIMESTAMP] = $site[self::TIMESTAMP];
$summary[self::ENCODING] =
(empty($site[self::ENCODING])) ?
'UTF-8' : $site[self::ENCODING];
$summary[self::HASH] = $link_id;
$summary[self::TYPE] = "link";
$summary[self::HTTP_CODE] = $link_keys;
$summary[self::LANG] = $lang;
$this->found_sites[self::LINK_SEEN_URLS][$part_num][] =
$summary;
}
}
}
unset($this->found_sites[self::SEEN_URLS]);
$this->pages_compressed = 0;
$this->crawl_stat_info['NUM_SEEN_SITES'] += $num_seen;
}
return $compress_urls;
}
/**
* Sends to crawl, robot, and index data to the current queue server.
* If this data is more than post_max_size, it splits it into chunks
* which are then reassembled by the queue server web app before being
* put into the appropriate schedule sub-directory.
*
* @param string $queue_server url of the current queue server
* @param array $byte_counts has four fields: TOTAL, ROBOT, SCHEDULE,
* INDEX. These give the number of bytes overall for the
* 'data' field of $post_data and for each of these components.
* @param array $post_data data to be uploaded to the queue server web app
* @return mixed return value of the recursive uploadCrawlData call
* on the rate-limiting retry path; void on the normal path
* (upload completes via execMachines)
*/
public function uploadCrawlData($queue_server, $byte_counts, &$post_data)
{
static $total_uploaded = 0;
$post_data['fetcher_peak_memory'] = memory_get_peak_usage();
$post_data['byte_counts'] = L\webencode(serialize($byte_counts));
$data = $post_data['data'];
unset($post_data['data']);
$post_data['hash_data'] = L\crawlHash($data);
$len = strlen($data);
do {
$part_size = $this->postSize($queue_server);
$max_len = $part_size - self::POST_OTHER_FIELDS_LEN;
$num_parts = max(1, (int)ceil($len / $max_len));
$post_data['num_parts'] = $num_parts;
$restart = false;
for ($i = 1; $i <= $num_parts && !$restart; $i++) {
$post_data['part'] = substr($data, ($i - 1) * $max_len,
$max_len);
$post_data['hash_part'] = L\crawlHash($post_data['part']);
$post_data['current_part'] = $i;
L\crawlLog("Sending Queue Server Part $i of $num_parts...");
L\crawlLog("...sending about " . strlen($post_data['part']) .
" bytes.");
$failures = 0;
do {
if ($failures > 0) {
$this->logUploadTrouble($queue_server, $info_string,
$post_data);
sleep(C\FETCH_SLEEP_TIME);
}
$time = time();
$post_data['time'] = $time;
$post_data['session'] = md5($time . C\p('AUTH_KEY'));
$info_string = FetchUrl::getPage($queue_server,
$post_data, true);
$info = @unserialize(trim($info_string));
if (isset($info[self::LOGGING])) {
L\crawlLog("Messages from Fetch Controller:");
L\crawlLog($info[self::LOGGING]);
}
if (isset($info[self::POST_MAX_SIZE])) {
$this->noteServerPostLimit($queue_server,
$info[self::POST_MAX_SIZE]);
}
$worked = isset($info[self::STATUS]) &&
$info[self::STATUS] == self::CONTINUE_STATE;
if (!$worked) {
$failures++;
/* One failure can be the server or the network
being busy, so a size counts as too big only
when it fails twice in a row. */
if ($failures % 2 == 0) {
$this->notePostResult($queue_server, $part_size,
false);
}
}
if ($this->postSize($queue_server) < $part_size) {
$restart = true;
L\crawlLog("Restarting upload with parts of " .
$this->postSize($queue_server) . " bytes.");
}
} while (!$worked && !$restart);
}
} while ($restart);
$this->notePostResult($queue_server, $part_size, true);
if (!empty($info['REPEAT_TIME']) && !empty($this->schedule_time)
&& $info['REPEAT_TIME'] > $this->schedule_time) {
/* we are crawling using a double index archive and it has
swapped.
*/
L\crawlLog("Queue Server Double Index Archive Swapped...");
L\crawlLog("...deleting rest of to crawl batch");
$this->to_crawl = [];
$this->to_crawl_again = [];
$this->found_sites = [];
}
L\crawlLog("Queue Server info response code: ".$info[self::STATUS]);
L\crawlLog("Queue Server's crawl time is: " .
($info[self::CRAWL_TIME] ?? ""));
L\crawlLog("Web Server peak memory usage: ".
($info[self::MEMORY_USAGE] ?? ""));
L\crawlLog("This fetcher peak memory usage: ".
memory_get_peak_usage());
L\crawlLog(
"Updated Queue Server, sent approximately" .
" {$byte_counts['TOTAL']} bytes:");
$total_uploaded += $byte_counts['TOTAL'];
$uploads_per_hour = $this->ratePerHour("UPLOADS", $total_uploaded,
time());
L\crawlLog("... Upload bytes per hours: " . $uploads_per_hour);
}
/**
* Logs why an upload part to a queue server did not go through: the
* server's answer, or what was posted when there was no answer.
*
* @param string $queue_server url of the queue server
* @param string $info_string the server's answer to the post
* @param array $post_data the fields that were posted
*/
public function logUploadTrouble($queue_server, $info_string, $post_data)
{
L\crawlLog("Trouble sending to the scheduler at url:");
L\crawlLog($queue_server);
if (empty($info_string)) {
L\crawlLog("Response from server was empty");
unset($post_data['part']);
L\crawlLog("Here is what was posted, apart from the data:");
L\crawlLog(print_r($post_data, true));
} else {
L\crawlLog("Response was:");
L\crawlLog("$info_string");
}
L\crawlLog("Trying again in " . C\FETCH_SLEEP_TIME . " seconds...");
}
/**
* Gives the size in bytes of the next upload part to a queue server.
* A server not yet posted to gets DEFAULT_POST_MAX_SIZE.
*
* @param string $queue_server url of the queue server
* @return int the part size in bytes
*/
public function postSize($queue_server)
{
return $this->post_sizes[$queue_server]['SIZE'] ??
self::DEFAULT_POST_MAX_SIZE;
}
/**
* Records the largest post a queue server says it takes. The next
* part size is kept at or under it, and, when no size has failed, is
* raised to it, so uploads use the largest parts the server allows.
*
* @param string $queue_server url of the queue server
* @param int $limit the largest post in bytes the server takes
*/
public function noteServerPostLimit($queue_server, $limit)
{
$limit = intval($limit);
if ($limit <= self::POST_OTHER_FIELDS_LEN) {
return;
}
$sizes = $this->post_sizes[$queue_server] ?? ['SIZE' =>
self::DEFAULT_POST_MAX_SIZE, 'GOOD' => 0, 'BAD' => 0];
$sizes['LIMIT'] = $limit;
if ($sizes['BAD'] > $limit) {
$sizes['BAD'] = 0;
}
$sizes['SIZE'] = (empty($sizes['BAD'])) ? $limit :
min($sizes['SIZE'], $limit);
$this->post_sizes[$queue_server] = $sizes;
}
/**
* Records whether an upload to a queue server went through at a part
* size and picks the next part size. A size that went through raises
* GOOD and a size that failed lowers BAD. The next size is halfway
* between them, or half the failed size when none has gone through,
* and never under MIN_POST_SIZE. When GOOD and BAD are within
* POST_SIZE_STEP of each other, the next size is GOOD.
*
* @param string $queue_server url of the queue server
* @param int $size the part size in bytes that was used
* @param bool $worked whether the upload went through at that size
*/
public function notePostResult($queue_server, $size, $worked)
{
$sizes = $this->post_sizes[$queue_server] ?? ['SIZE' => $size,
'LIMIT' => $size, 'GOOD' => 0, 'BAD' => 0];
if ($worked) {
$sizes['GOOD'] = max($sizes['GOOD'], $size);
} else {
$sizes['BAD'] = (empty($sizes['BAD'])) ? $size :
min($sizes['BAD'], $size);
if ($sizes['GOOD'] >= $sizes['BAD']) {
$sizes['GOOD'] = 0;
}
}
if (empty($sizes['BAD'])) {
$sizes['SIZE'] = max($sizes['LIMIT'], $sizes['GOOD']);
} else if (empty($sizes['GOOD'])) {
$sizes['SIZE'] = max(self::MIN_POST_SIZE,
intdiv($sizes['BAD'], 2));
} else if ($sizes['BAD'] - $sizes['GOOD'] < self::POST_SIZE_STEP) {
$sizes['SIZE'] = $sizes['GOOD'];
} else {
$sizes['SIZE'] = intdiv($sizes['GOOD'] + $sizes['BAD'], 2);
}
$this->post_sizes[$queue_server] = $sizes;
}
/**
* ratePerHour gives how much of a running total, such as the pages
* downloaded, came per hour over the last one to two hours. The window
* of a figure moves once an hour: its start becomes the time it last
* moved, and the total two hours before becomes the total one hour
* before. The first window starts at $rate_start, when the fetcher was
* made, so the first figure is the total over the time since then. A
* window of no length counts as one hour.
*
* @param string $name the figure, which names its window in
* $rate_windows
* @param float $total the running total of the figure
* @param int $now the time of the total
* @return float the amount per hour
*/
public function ratePerHour($name, $total, $now)
{
$window = $this->rate_windows[$name] ??
[0, 0, $this->rate_start, 0];
[$total_two_hours, $total_previous_hour, $window_start,
$last_move] = $window;
$hours = ($now - $window_start) / C\ONE_HOUR;
$hours = ($hours <= 0) ? 1 : $hours;
$per_hour = ($total - $total_two_hours) / $hours;
if ($now > $last_move + C\ONE_HOUR) {
$window = [$total_previous_hour, $total,
($last_move > 0) ? $last_move : $window_start, $now];
}
$this->rate_windows[$name] = $window;
return $per_hour;
}
/**
* noteSlowPage counts a slow page from a host in $slow_pages. A
* host's count starts again when its hour has passed. When more
* than SLOW_HOSTS_KEPT hosts are held, those whose hour has passed
* are dropped.
*
* @param string $host the host the slow page came from
* @param int $now the time of the slow page
* @return int how many slow pages the host has given this hour
*/
public function noteSlowPage($host, $now)
{
if (count($this->slow_pages) > self::SLOW_HOSTS_KEPT) {
$this->slow_pages = array_filter($this->slow_pages,
fn($held) => $now - $held[0] < C\ONE_HOUR);
}
[$hour_start, $count] = $this->slow_pages[$host] ?? [$now, 0];
if ($now - $hour_start >= C\ONE_HOUR) {
[$hour_start, $count] = [$now, 0];
}
$this->slow_pages[$host] = [$hour_start, $count + 1];
return $count + 1;
}
/**
* usesFastProcessor says whether a host's pages get the fast variant
* of their processor: true once the host has given
* SLOW_PAGES_BEFORE_FAST slow pages in the hour that has not yet
* passed. A fetcher then summarizes the host's pages with the
* ScrapeSummarizer, which reads the page's description and first
* blocks of text rather than scoring every sentence.
*
* @param string $host the host of the page about to be processed
* @param int $now the time now
* @return bool whether to use the fast processor
*/
public function usesFastProcessor($host, $now)
{
[$hour_start, $count] = $this->slow_pages[$host] ?? [$now, 0];
return $now - $hour_start < C\ONE_HOUR &&
$count >= self::SLOW_PAGES_BEFORE_FAST;
}
/**
* buildSlots says how many fetchers on this machine and channel may
* build closed partitions at once: BUILD_FETCHER_FRACTION of the
* fetchers running, rounded up, and at least one. A fetcher counts as
* running while its lock file in the schedules folder is younger than
* STALE_LOCK_TIMEOUT, as CrawlDaemon keeps it. The fetcher's loop
* passes the result to IndexDocumentBundle::buildClosedPartition.
*
* @return int the number of build slots
*/
public function buildSlots()
{
$running = 0;
$now = time();
foreach (glob($this->schedules_folder .
"/*-{$this->channel}-FetcherLock.txt")
as $lock_file) {
if ($now - @filemtime($lock_file) < C\STALE_LOCK_TIMEOUT) {
$running++;
}
}
return max(1, (int)ceil(C\BUILD_FETCHER_FRACTION * $running));
}
}
if (!C\nsdefined("UNIT_TEST_MODE")) {
/*
* Instantiate and runs the Fetcher
*/
$fetcher = new Fetcher();
$fetcher->start();
}
- Yioop Repo - Seekquarry