<commit_before>import os
import argparse
from time import time
from time import sleep
from sqlalchemy import sql
from sqlalchemy import exc
from subprocess import call
from sqlalchemy import text
from sqlalchemy import orm
import heroku3
from pprint import pprint
import datetime


from app import db
from app import logger

from queue_main import DbQueue
from util import elapsed
from util import safe_commit

from repository import Repository

class DbQueueRepo(DbQueue):
    def table_name(self, job_type):
        table_name = "repository"
        return table_name

    def process_name(self, job_type):
        process_name = "run_repo" # formation name is from Procfile
        return process_name

    def maint(self, **kwargs):
        repos = Repository.query.filter(Repository.name==None, Repository.error==None).all()
        num_to_commit = 0
        for my_repo in repos:
            if not my_repo.name:
                my_repo.set_repo_info()
                num_to_commit += 1
                db.session.merge(my_repo)
                logger.info(u"my_repo: {}".format(my_repo))
            if num_to_commit >= 1:
                safe_commit(db)
                num_to_commit = 0
        safe_commit(db)

    def add_pmh_record(self, **kwargs):
        repo_id = kwargs.get("id", None)
        record_id = kwargs.get("recordid")
        my_repo = Repository.query.get(repo_id)
        print "my_repo", my_repo
        my_pmh_record = my_repo.get_pmh_record(record_id)
        print "my_pmh_record", my_pmh_record

        my_pmh_record.mint_pages()

        # for my_page in my_pmh_record.pages:
        #     print "my_page", my_page
        #     my_page.scrape()

        db.session.merge(my_pmh_record)
        print my_pmh_record.pages

        safe_commit(db)



    def worker_run(self, **kwargs):
        single_obj_id = kwargs.get("id", None)
        chunk = kwargs.get("chunk")
        queue_table = "repository"
        run_method = kwargs.get("method")
        run_class = Repository

        limit = 1 # just do one repo at a time

        text_query_pattern = """WITH picked_from_queue AS (
                   SELECT *
                   FROM   {queue_table}
                   WHERE
                   (most_recent_year_harvested is null or (most_recent_year_harvested < now() - interval '1 day'))
                   and (last_harvest_started is null or last_harvest_finished is not null or last_harvest_started < now() - interval '1 day')
                    and error is null and ready_to_run=true
                   ORDER BY random() -- not rand, because want it to be different every time
               LIMIT  {chunk}
               FOR UPDATE SKIP LOCKED
               )
            UPDATE {queue_table} queue_rows_to_update
            SET    last_harvest_started=now() at time zone 'utc', last_harvest_finished=null
            FROM   picked_from_queue
            WHERE picked_from_queue.id = queue_rows_to_update.id
            RETURNING picked_from_queue.*;"""
        text_query = text_query_pattern.format(
            chunk=chunk,
            queue_table=queue_table
        )

        index = 0
        start_time = time()
        while True:
            new_loop_start_time = time()
            if single_obj_id:
                objects = [run_class.query.filter(run_class.id == single_obj_id).first()]
            else:
                logger.info(u"the queue query is:\n{}".format(text_query))
                # logger.info(u"looking for new jobs")
                objects = run_class.query.from_statement(text(text_query)).execution_options(autocommit=True).all()
                # logger.info(u"finished get-new-objects query in {} seconds".format(elapsed(new_loop_start_time)))

            if not objects:
                # logger.info(u"sleeping for 5 seconds, then going again")
                sleep(5)<commit_msg>give it an hour to finish before we'd restart<commit_after>import os
import argparse
from time import time
from time import sleep
from sqlalchemy import sql
from sqlalchemy import exc
from subprocess import call
from sqlalchemy import text
from sqlalchemy import orm
import heroku3
from pprint import pprint
import datetime


from app import db
from app import logger

from queue_main import DbQueue
from util import elapsed
from util import safe_commit

from repository import Repository

class DbQueueRepo(DbQueue):
    def table_name(self, job_type):
        table_name = "repository"
        return table_name

    def process_name(self, job_type):
        process_name = "run_repo" # formation name is from Procfile
        return process_name

    def maint(self, **kwargs):
        repos = Repository.query.filter(Repository.name==None, Repository.error==None).all()
        num_to_commit = 0
        for my_repo in repos:
            if not my_repo.name:
                my_repo.set_repo_info()
                num_to_commit += 1
                db.session.merge(my_repo)
                logger.info(u"my_repo: {}".format(my_repo))
            if num_to_commit >= 1:
                safe_commit(db)
                num_to_commit = 0
        safe_commit(db)

    def add_pmh_record(self, **kwargs):
        repo_id = kwargs.get("id", None)
        record_id = kwargs.get("recordid")
        my_repo = Repository.query.get(repo_id)
        print "my_repo", my_repo
        my_pmh_record = my_repo.get_pmh_record(record_id)
        print "my_pmh_record", my_pmh_record

        my_pmh_record.mint_pages()

        # for my_page in my_pmh_record.pages:
        #     print "my_page", my_page
        #     my_page.scrape()

        db.session.merge(my_pmh_record)
        print my_pmh_record.pages

        safe_commit(db)



    def worker_run(self, **kwargs):
        single_obj_id = kwargs.get("id", None)
        chunk = kwargs.get("chunk")
        queue_table = "repository"
        run_method = kwargs.get("method")
        run_class = Repository

        limit = 1 # just do one repo at a time

        text_query_pattern = """WITH picked_from_queue AS (
                   SELECT *
                   FROM   {queue_table}
                   WHERE
                   (most_recent_year_harvested is null or (most_recent_year_harvested < now() - interval '1 day'))
                   and (last_harvest_started is null or
                        last_harvest_started < now() - interval '1 hour' or
                        last_harvest_finished is not null or
                        last_harvest_started < now() - interval '1 day')
                    and error is null and ready_to_run=true
                   ORDER BY random() -- not rand, because want it to be different every time
               LIMIT  {chunk}
               FOR UPDATE SKIP LOCKED
               )
            UPDATE {queue_table} queue_rows_to_update
            SET    last_harvest_started=now() at time zone 'utc', last_harvest_finished=null
            FROM   picked_from_queue
            WHERE picked_from_queue.id = queue_rows_to_update.id
            RETURNING picked_from_queue.*;"""
        text_query = text_query_pattern.format(
            chunk=chunk,
            queue_table=queue_table
        )

        index = 0
        start_time = time()
        while True:
            new_loop_start_time = time()
            if single_obj_id:
                objects = [run_class.query.filter(run_class.id == single_obj_id).first()]
            else:
                logger.info(u"the queue query is:\n{}".format(text_query))
                # logger.info(u"looking for new jobs")
                objects = run_class.query.from_statement(text(text_query)).execution_options(autocommit=True).all()
                # logger.info(u"finished get-new-objects query in {} seconds".format(elapsed(new_loop_start_time)))

            if not objects:
                # logger.info(u"sleeping for 5 seconds, then going again")
                sleep(5)
{'fraction_non_alphanumeric': 0.07683544303797468, 'fraction_numerical': 0.002911392405063291, 'mean_word_length': 2.07431906614786, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 5, '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': '3570155', 'n_tokens_mistral': 2358, 'n_tokens_neox': 2236, 'n_words': 583}