<commit_before># coding=utf-8
from spamhandling import handle_spam, check_if_spam
from datahandling import (add_or_update_api_data, clear_api_data, schedule_store_bodyfetcher_queue,
                          schedule_store_bodyfetcher_max_ids, add_queue_timing_data)
from chatcommunicate import tell_rooms_with
from globalvars import GlobalVars
from operator import itemgetter
from datetime import datetime
import json
import time
import threading
import requests
import copy
from classes import Post, PostParseError
from helpers import (log, log_current_thread, append_to_current_thread_name,
                     convert_new_scan_to_spam_result_if_new_reasons, add_to_global_bodyfetcher_queue_in_new_thread)
import recently_scanned_posts as rsp
from itertools import chain
from tasks import Tasks


# noinspection PyClassHasNoInit,PyBroadException
class BodyFetcher:
    queue = {}
    previous_max_ids = {}
    posts_in_process = {}

    # special_cases are the minimum number of posts, for each of the specified sites, which
    # need to be in the queue prior to feching posts.
    # The number of questions we fetch each day per site is the total of
    #   new questions + new answers + edits.
    # Stack Overflow is handled specially. It's know that some SO edits/new posts don't
    #   appear on the WebSocket.
    # Queue depths were last comprehensively adjusted on 2015-12-30.
    special_cases = {
        # 2020-11-02:
        # Request numbers pre 2020-11-02 are very low due to a now fixed bug.
        #
        #                                                                pre                   sum of requests
        #                                               questions    2020-11-02   2020-02-19    2020-10-28 to
        #                                                per day       setting     requests      2020-11-02
        # "stackoverflow.com": 3,                   # _  6,816            3          360            4,365
        # "math.stackexchange.com": 2,              # _    596            1          473            6,346
        # "ru.stackoverflow.com": 2,                # _    230           10           13              145
        # "askubuntu.com": ,                        # _    140            1           88            1,199
        # "es.stackoverflow.com": 2,                # _    138            5           25              225
        # "superuser.com": ,                        # _    122            1           87            1,038
        # "physics.stackexchange.com": ,            # _     90            1           76            1,161
        # "stats.stackexchange.com": 2,             # _     82            5           16              151
        # "pt.stackoverflow.com": 2,                # _     73           10            7               75
        # "unix.stackexchange.com": ,               # _     72            1           76              772
        # "electronics.stackexchange.com": ,        # _     69            1           46              723
        # "serverfault.com": ,                      # _     62            1           43              582
        # "tex.stackexchange.com": 2,               # _     60            5            8               98
        # "blender.stackexchange.com": 2,           # _     59            5            8               85
        # "salesforce.stackexchange.com": ,         # _     49            1           47              472
        # "gis.stackexchange.com": 2,               # _     46            3           15              166
        # "mathoverflow.net" (time_sensitive)       # _     37            -           33              511
        # "english.stackexchange.com": ,            # _     36            1           34              382
        # "magento.stackexchange.com": 2,           # _     34            3            5               93
        # "ell.stackexchange.com": ,                # _     33            1           24              365
        # "wordpress.stackexchange.com": ,          # _     29            1           30              283
        # "apple.stackexchange.com": ,              # _     29            1           46              294
        # "diy.stackexchange.com": ,                # _     26            1           24              306
        # "mathematica.stackexchange.com": ,        # _     25            1           21              384
        # "dba.stackexchange.com": ,                # _     23            1           31              343
        # "datascience.stackexchange.com": ,        # _     21            1           17              220
        # "chemistry.stackexchange.com": ,          # _     20            1           20              140
        # "security.stackexchange.com": ,           # _     18            1           15              238
        # "codereview.stackexchange.com": ,         # _     18            5            2               39
        #  The only reason this is the cut-off is that it was the last in the existing list
        #    as of 2020-11-01.
    }

    time_sensitive = ["security.stackexchange.com", "movies.stackexchange.com",
                      "mathoverflow.net", "gaming.stackexchange.com", "webmasters.stackexchange.com",
                      "arduino.stackexchange.com", "workplace.stackexchange.com"]

    threshold = 1

    last_activity_date = 0
    last_activity_date_lock = threading.Lock()
    ACTIVITY_DATE_EXTRA_EARLIER_MS_TO_FETCH = 6 * 60 * 1000  # 6 minutes in milliseconds; is beyond edit grace period

    api_data_lock = threading.Lock()
    queue_lock = threading.Lock()
    max_ids_modify_lock = threading.Lock()
    check_queue_lock = threading.Lock()
    posts_in_process_lock = threading.Lock()

    def add_to_queue(self, hostname, question_id, should_check_site=False):
        # For the Sandbox questions on MSE, we choose to ignore the entire question and all answers.
        ignored_mse_questions = [
            3122,    # Formatting Sandbox
            51812,   # The API sandbox
            296077,  # Sandbox archive
        ]
        if question_id in ignored_mse_questions and hostname == "meta.stackexchange.com":
            return  # don't check meta sandbox, it's full of weird posts

        with self.queue_lock:
            if hostname not in self.queue:
                self.queue[hostname] = {}

            # Something about how the queue is being filled is storing Post IDs in a list.
            # So, if we get here we need to make sure that the correct types are paseed.
            #
            # If the item in self.queue[hostname] is a dict, do nothing.
            # If the item in self.queue[hostname] is not a dict but is a list or a tuple, then convert to dict and
            # then replace the list or tuple with the dict.
            # If the item in self.queue[hostname] is neither a dict or a list, then explode.
            if type(self.queue[hostname]) is dict:
                pass
            elif type(self.queue[hostname]) in [list, tuple]:
                post_list_dict = {}
                for post_list_id in self.queue[hostname]:
                    post_list_dict[str(post_list_id)] = None
                self.queue[hostname] = post_list_dict
            else:
                raise TypeError("A non-iterable is in the queue item for a given site, this will cause errors!")

            # This line only works if we are using a dict in the self.queue[hostname] object, which we should be with
            # the previous conversion code.
            self.queue[hostname][str(question_id)] = datetime.utcnow()
            flovis_dict = None
            if GlobalVars.flovis is not None:
                flovis_dict = {sk: list(sq.keys()) for sk, sq in self.queue.items()}

        if flovis_dict is not None:
            GlobalVars.flovis.stage('bodyfetcher/enqueued', hostname, question_id, flovis_dict)

        if should_check_site:
            # The call to add_to_queue indicated that the site should be immediately processed.
            with self.queue_lock:
                new_posts = self.queue.pop(hostname, None)
            if new_posts:
                schedule_store_bodyfetcher_queue()
                self.make_api_call_for_site(hostname, new_posts)

        site_and_posts = self.get_fist_queue_item_to_process()
        if site_and_posts:
            schedule_store_bodyfetcher_queue()
            self.make_api_call_for_site(*site_and_posts)

    def get_fist_queue_item_to_process(self):
        # We use a copy of the queue keys (sites) and lengths in order to allow
        # the queue to be changed in other threads.
        # Overall this results in a FIFO for sites which have reached their threshold, because
        # dicts are guaranteed to be iterated in insertion order in Python >= 3.6.
        # We use self.check_queue_lock here to fully dispatch one queued site at a time and allow
        # consolidation of multiple WebSocket events for the same real-world event.
        with self.check_queue_lock:
            time.sleep(.25)  # Some time for multiple potential  WebSocket events to queue the same post
            special_sites = []
            site_to_handle = None
            is_time_sensitive_time = datetime.utcnow().hour in range(4, 12)
            with self.queue_lock:
                sites_in_queue = {site: len(values) for site, values in self.queue.items()}
            # Get sites listed in special cases and as time_sensitive
            for site, length in sites_in_queue.items():
                if site in self.special_cases:
                    special_sites.append(site)
                    if length >= self.special_cases[site]:
                        site_to_handle = site
                        break
                if is_time_sensitive_time and site in self.time_sensitive:
                    special_sites.append(site)
                    if length >= 1:
                        site_to_handle = site
                        break
            else:
                # We didn't find a special site which met the applicable threshold.
                # Remove the sites which we've already considered from our copy of the queue's keys.
                for site in special_sites:
                    sites_in_queue.pop(site, None)<commit_msg>Prevent additional scan threads if CPU use above threshold.<commit_after># coding=utf-8
from spamhandling import handle_spam, check_if_spam
from datahandling import (add_or_update_api_data, clear_api_data, schedule_store_bodyfetcher_queue,
                          schedule_store_bodyfetcher_max_ids, add_queue_timing_data)
from chatcommunicate import tell_rooms_with
from globalvars import GlobalVars
from operator import itemgetter
from datetime import datetime
import json
import time
import threading
import requests
import copy
from classes import Post, PostParseError
from helpers import (log, log_current_thread, append_to_current_thread_name,
                     convert_new_scan_to_spam_result_if_new_reasons, add_to_global_bodyfetcher_queue_in_new_thread)
import recently_scanned_posts as rsp
from itertools import chain
from tasks import Tasks
import psutil


# noinspection PyClassHasNoInit,PyBroadException
class BodyFetcher:
    queue = {}
    previous_max_ids = {}
    posts_in_process = {}

    LAUNCH_PROCESSING_THREAD_MAXIMUM_CPU_USE_THRESHOLD = 98

    # special_cases are the minimum number of posts, for each of the specified sites, which
    # need to be in the queue prior to feching posts.
    # The number of questions we fetch each day per site is the total of
    #   new questions + new answers + edits.
    # Stack Overflow is handled specially. It's know that some SO edits/new posts don't
    #   appear on the WebSocket.
    # Queue depths were last comprehensively adjusted on 2015-12-30.
    special_cases = {
        # 2020-11-02:
        # Request numbers pre 2020-11-02 are very low due to a now fixed bug.
        #
        #                                                                pre                   sum of requests
        #                                               questions    2020-11-02   2020-02-19    2020-10-28 to
        #                                                per day       setting     requests      2020-11-02
        # "stackoverflow.com": 3,                   # _  6,816            3          360            4,365
        # "math.stackexchange.com": 2,              # _    596            1          473            6,346
        # "ru.stackoverflow.com": 2,                # _    230           10           13              145
        # "askubuntu.com": ,                        # _    140            1           88            1,199
        # "es.stackoverflow.com": 2,                # _    138            5           25              225
        # "superuser.com": ,                        # _    122            1           87            1,038
        # "physics.stackexchange.com": ,            # _     90            1           76            1,161
        # "stats.stackexchange.com": 2,             # _     82            5           16              151
        # "pt.stackoverflow.com": 2,                # _     73           10            7               75
        # "unix.stackexchange.com": ,               # _     72            1           76              772
        # "electronics.stackexchange.com": ,        # _     69            1           46              723
        # "serverfault.com": ,                      # _     62            1           43              582
        # "tex.stackexchange.com": 2,               # _     60            5            8               98
        # "blender.stackexchange.com": 2,           # _     59            5            8               85
        # "salesforce.stackexchange.com": ,         # _     49            1           47              472
        # "gis.stackexchange.com": 2,               # _     46            3           15              166
        # "mathoverflow.net" (time_sensitive)       # _     37            -           33              511
        # "english.stackexchange.com": ,            # _     36            1           34              382
        # "magento.stackexchange.com": 2,           # _     34            3            5               93
        # "ell.stackexchange.com": ,                # _     33            1           24              365
        # "wordpress.stackexchange.com": ,          # _     29            1           30              283
        # "apple.stackexchange.com": ,              # _     29            1           46              294
        # "diy.stackexchange.com": ,                # _     26            1           24              306
        # "mathematica.stackexchange.com": ,        # _     25            1           21              384
        # "dba.stackexchange.com": ,                # _     23            1           31              343
        # "datascience.stackexchange.com": ,        # _     21            1           17              220
        # "chemistry.stackexchange.com": ,          # _     20            1           20              140
        # "security.stackexchange.com": ,           # _     18            1           15              238
        # "codereview.stackexchange.com": ,         # _     18            5            2               39
        #  The only reason this is the cut-off is that it was the last in the existing list
        #    as of 2020-11-01.
    }

    time_sensitive = ["security.stackexchange.com", "movies.stackexchange.com",
                      "mathoverflow.net", "gaming.stackexchange.com", "webmasters.stackexchange.com",
                      "arduino.stackexchange.com", "workplace.stackexchange.com"]

    threshold = 1

    last_activity_date = 0
    last_activity_date_lock = threading.Lock()
    ACTIVITY_DATE_EXTRA_EARLIER_MS_TO_FETCH = 6 * 60 * 1000  # 6 minutes in milliseconds; is beyond edit grace period

    api_data_lock = threading.Lock()
    queue_lock = threading.Lock()
    max_ids_modify_lock = threading.Lock()
    check_queue_lock = threading.Lock()
    posts_in_process_lock = threading.Lock()

    def add_to_queue(self, hostname, question_id, should_check_site=False):
        # For the Sandbox questions on MSE, we choose to ignore the entire question and all answers.
        ignored_mse_questions = [
            3122,    # Formatting Sandbox
            51812,   # The API sandbox
            296077,  # Sandbox archive
        ]
        if question_id in ignored_mse_questions and hostname == "meta.stackexchange.com":
            return  # don't check meta sandbox, it's full of weird posts

        with self.queue_lock:
            if hostname not in self.queue:
                self.queue[hostname] = {}

            # Something about how the queue is being filled is storing Post IDs in a list.
            # So, if we get here we need to make sure that the correct types are paseed.
            #
            # If the item in self.queue[hostname] is a dict, do nothing.
            # If the item in self.queue[hostname] is not a dict but is a list or a tuple, then convert to dict and
            # then replace the list or tuple with the dict.
            # If the item in self.queue[hostname] is neither a dict or a list, then explode.
            if type(self.queue[hostname]) is dict:
                pass
            elif type(self.queue[hostname]) in [list, tuple]:
                post_list_dict = {}
                for post_list_id in self.queue[hostname]:
                    post_list_dict[str(post_list_id)] = None
                self.queue[hostname] = post_list_dict
            else:
                raise TypeError("A non-iterable is in the queue item for a given site, this will cause errors!")

            # This line only works if we are using a dict in the self.queue[hostname] object, which we should be with
            # the previous conversion code.
            self.queue[hostname][str(question_id)] = datetime.utcnow()
            flovis_dict = None
            if GlobalVars.flovis is not None:
                flovis_dict = {sk: list(sq.keys()) for sk, sq in self.queue.items()}

        if flovis_dict is not None:
            GlobalVars.flovis.stage('bodyfetcher/enqueued', hostname, question_id, flovis_dict)

        if should_check_site:
            # The call to add_to_queue indicated that the site should be immediately processed.
            with self.queue_lock:
                new_posts = self.queue.pop(hostname, None)
            if new_posts:
                schedule_store_bodyfetcher_queue()
                self.make_api_call_for_site(hostname, new_posts)

        site_and_posts = self.get_fist_queue_item_to_process()
        if site_and_posts:
            schedule_store_bodyfetcher_queue()
            self.make_api_call_for_site(*site_and_posts)

    def get_fist_queue_item_to_process(self):
        # We use a copy of the queue keys (sites) and lengths in order to allow
        # the queue to be changed in other threads.
        # Overall this results in a FIFO for sites which have reached their threshold, because
        # dicts are guaranteed to be iterated in insertion order in Python >= 3.6.
        # We use self.check_queue_lock here to fully dispatch one queued site at a time and allow
        # consolidation of multiple WebSocket events for the same real-world event.
        with self.check_queue_lock:
            # Getting the CPU activity also provides time for multiple potential WebSocket events to queue
            # the same post along with some time for the SE API to update and have information on the new post.
            cpu_use = psutil.cpu_percent(interval=1)
            if cpu_use > self.LAUNCH_PROCESSING_THREAD_MAXIMUM_CPU_USE_THRESHOLD:
                # We are already maxing out the CPU. Having additional threads processing posts is counterproductive.
                log_current_thread('warning', 'CPU use is {}'.format(cpu_use)
                                              + ', which is too high to launch an additional scan thread.')
                return None
            special_sites = []
            site_to_handle = None
            is_time_sensitive_time = datetime.utcnow().hour in range(4, 12)
            with self.queue_lock:
                sites_in_queue = {site: len(values) for site, values in self.queue.items()}
            # Get sites listed in special cases and as time_sensitive
            for site, length in sites_in_queue.items():
                if site in self.special_cases:
                    special_sites.append(site)
                    if length >= self.special_cases[site]:
                        site_to_handle = site
                        break
                if is_time_sensitive_time and site in self.time_sensitive:
                    special_sites.append(site)
                    if length >= 1:
                        site_to_handle = site
                        break
            else:
                # We didn't find a special site which met the applicable threshold.
                # Remove the sites which we've already considered from our copy of the queue's keys.
                for site in special_sites:
                    sites_in_queue.pop(site, None)
{'fraction_non_alphanumeric': 0.06275836151822622, 'fraction_numerical': 0.03245960165351372, 'mean_word_length': 1.3871944382148464, 'pattern_counts': {'":': 58, '<': 3, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 130, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '294039', 'n_tokens_mistral': 6078, 'n_tokens_neox': 5537, 'n_words': 1704}