<commit_before>import argparse
import logging
import os
import redis
import pickle
from datetime import datetime, timedelta
from multiprocessing import Pool, current_process
from time import sleep
from time import time
from urlparse import urlparse
from redis import WatchError

from sqlalchemy import orm, sql, text
from sqlalchemy.orm import make_transient

from app import db
from app import logger
from oa_page import oa_publisher_equivalent
from page import PageNew
from queue_main import DbQueue
from util import elapsed
from util import safe_commit

from pub import Pub # foul magic
import endpoint # magic
import pmh_record # more magic


def scrape_pages(pages):
    for page in pages:
        make_transient(page)

    # free up the connection while doing net IO
    db.session.close()
    db.engine.dispose()

    pool = get_worker_pool()
    scraped_pages = [p for p in pool.map(scrape_page, pages, chunksize=1) if p]
    logger.info(u'finished scraping all pages')
    pool.close()
    pool.join()

    logger.info(u'preparing update records')
    row_dicts = [x.__dict__ for x in scraped_pages]
    for row_dict in row_dicts:
        row_dict.pop('_sa_instance_state')

    logger.info(u'saving update records')
    db.session.bulk_update_mappings(PageNew, row_dicts)

    scraped_page_ids = [p.id for p in scraped_pages]
    return scraped_page_ids


def get_worker_pool():
    num_request_workers = int(os.getenv('GREEN_SCRAPE_PROCS_PER_WORKER', 10))
    return Pool(processes=num_request_workers, maxtasksperchild=10)


def scrape_page(page):
    worker = current_process().name
    domain = urlparse(page.url).netloc
    logger.info(u'{} started scraping page at {}: {}'.format(worker, domain, page))
    if begin_rate_limit_domain(domain):
        page.scrape()
        end_rate_limit_domain(domain)
        logger.info(u'{} finished scraping page: {}'.format(worker, page))
        return page
    else:
        logger.info(u'{} not ready to scrape {}'.format(worker, domain))
        return None


def unpickle(v):
    return pickle.loads(v) if v else None


def redis_key(domain):
    return u'green-scrape:{}'.format(domain)


def begin_rate_limit_domain(domain, interval_seconds=10):
    r = redis.from_url(os.environ.get("REDIS_URL"))

    with r.pipeline() as pipe:
        try:
            pipe.watch(redis_key(domain))

            scrape_started = unpickle(r.hget(redis_key(domain), 'started'))
            scrape_finished = unpickle(r.hget(redis_key(domain), 'finished'))

            if (scrape_started and scrape_started >= datetime.utcnow() - timedelta(hours=1)) or (
                scrape_finished and scrape_finished >= datetime.utcnow() - timedelta(seconds=interval_seconds)
            ):
                return False

            pipe.multi()
            pipe.hset(redis_key(domain), 'started', pickle.dumps(datetime.utcnow()))
            pipe.hset(redis_key(domain), 'finished', pickle.dumps(None))
            pipe.execute()
            return True
        except WatchError:
            return False


def end_rate_limit_domain(domain):
    r = redis.from_url(os.environ.get("REDIS_URL"))

    r.hset(redis_key(domain), 'started', pickle.dumps(None))
    r.hset(redis_key(domain), 'finished', pickle.dumps(datetime.utcnow()))


class DbQueueGreenOAScrape(DbQueue):
    def table_name(self, job_type):
        return 'page_green_scrape_queue'

    def process_name(self, job_type):
        return 'run_green_oa_scrape'

    def worker_run(self, **kwargs):
        run_class = PageNew

        single_id = kwargs.get("id", None)
        chunk_size = kwargs.get("chunk", 100)
        limit = kwargs.get("limit", None)
        scrape_publisher = kwargs.get("scrape_publisher", False)

        if limit is None:
            limit = float("inf")

        if single_id:
            page = run_class.query.filter(run_class.id == single_id).first()
            page.scrape()
            db.session.merge(page)
            safe_commit(db) or logger.info(u"COMMIT fail")
        else:
            index = 0
            num_updated = 0
            start_time = time()

            while num_updated < limit:<commit_msg>fix race condition<commit_after>import argparse
import logging
import os
import redis
import pickle
from datetime import datetime, timedelta
from multiprocessing import Pool, current_process
from time import sleep
from time import time
from urlparse import urlparse
from redis import WatchError

from sqlalchemy import orm, sql, text
from sqlalchemy.orm import make_transient

from app import db
from app import logger
from oa_page import oa_publisher_equivalent
from page import PageNew
from queue_main import DbQueue
from util import elapsed
from util import safe_commit

from pub import Pub # foul magic
import endpoint # magic
import pmh_record # more magic


def scrape_pages(pages):
    for page in pages:
        make_transient(page)

    # free up the connection while doing net IO
    db.session.close()
    db.engine.dispose()

    pool = get_worker_pool()
    scraped_pages = [p for p in pool.map(scrape_page, pages, chunksize=1) if p]
    logger.info(u'finished scraping all pages')
    pool.close()
    pool.join()

    logger.info(u'preparing update records')
    row_dicts = [x.__dict__ for x in scraped_pages]
    for row_dict in row_dicts:
        row_dict.pop('_sa_instance_state')

    logger.info(u'saving update records')
    db.session.bulk_update_mappings(PageNew, row_dicts)

    scraped_page_ids = [p.id for p in scraped_pages]
    return scraped_page_ids


def get_worker_pool():
    num_request_workers = int(os.getenv('GREEN_SCRAPE_PROCS_PER_WORKER', 10))
    return Pool(processes=num_request_workers, maxtasksperchild=10)


def scrape_page(page):
    worker = current_process().name
    domain = urlparse(page.url).netloc
    logger.info(u'{} started scraping page at {}: {}'.format(worker, domain, page))
    if begin_rate_limit_domain(domain):
        page.scrape()
        end_rate_limit_domain(domain)
        logger.info(u'{} finished scraping page: {}'.format(worker, page))
        return page
    else:
        logger.info(u'{} not ready to scrape {}'.format(worker, domain))
        return None


def unpickle(v):
    return pickle.loads(v) if v else None


def redis_key(domain, property):
    return u'green-scrape:{}:{}'.format(domain, property)


def begin_rate_limit_domain(domain, interval_seconds=10):
    r = redis.from_url(os.environ.get("REDIS_URL"))

    with r.pipeline() as pipe:
        try:
            pipe.watch(redis_key(domain, 'started'))
            pipe.watch(redis_key(domain, 'finished'))

            scrape_started = unpickle(r.get(redis_key(domain, 'started')))
            scrape_finished = unpickle(r.get(redis_key(domain, 'finished')))

            if (scrape_started and scrape_started >= datetime.utcnow() - timedelta(hours=1)) or (
                scrape_finished and scrape_finished >= datetime.utcnow() - timedelta(seconds=interval_seconds)
            ):
                return False

            pipe.multi()
            pipe.set(redis_key(domain, 'started'), pickle.dumps(datetime.utcnow()))
            pipe.set(redis_key(domain, 'finished'), pickle.dumps(None))
            pipe.execute()
            return True
        except WatchError:
            return False


def end_rate_limit_domain(domain):
    r = redis.from_url(os.environ.get("REDIS_URL"))

    r.set(redis_key(domain, 'started'), pickle.dumps(None))
    r.set(redis_key(domain, 'finished'), pickle.dumps(datetime.utcnow()))


class DbQueueGreenOAScrape(DbQueue):
    def table_name(self, job_type):
        return 'page_green_scrape_queue'

    def process_name(self, job_type):
        return 'run_green_oa_scrape'

    def worker_run(self, **kwargs):
        run_class = PageNew

        single_id = kwargs.get("id", None)
        chunk_size = kwargs.get("chunk", 100)
        limit = kwargs.get("limit", None)
        scrape_publisher = kwargs.get("scrape_publisher", False)

        if limit is None:
            limit = float("inf")

        if single_id:
            page = run_class.query.filter(run_class.id == single_id).first()
            page.scrape()
            db.session.merge(page)
            safe_commit(db) or logger.info(u"COMMIT fail")
        else:
            index = 0
            num_updated = 0
            start_time = time()

            while num_updated < limit:
{'fraction_non_alphanumeric': 0.10015500178848218, 'fraction_numerical': 0.0031000357696434958, 'mean_word_length': 3.085728202630297, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6240662', 'n_tokens_mistral': 2723, 'n_tokens_neox': 2599, 'n_words': 638}