<commit_before>from iati.models import DocumentLink, Document
from iati_synchroniser.models import Dataset
from iati.activity_aggregation_calculation import ActivityAggregationCalculation
from django_rq import job
import django_rq
import datetime
from rq import Queue, Connection, Worker
from rq.job import Job
from redis import Redis
from django.conf import settings
import time
import requests
import os
import hashlib
from common.download_file import DownloadFile
from common.download_file import hash_file
import fulltext

redis_conn = Redis()


###############################
#### TASK QUEUE MANAGEMENT ####
###############################


@job
def remove_duplicates_from_parser_queue():
    raise Exception("Not implemented yet")


def delete_task_from_queue(job_id):
    Job.fetch(job_id, connection=redis_conn).delete()


def delete_all_tasks_from_queue(queue_name):

    if queue_name == "failed":
        q = django_rq.get_failed_queue()
    elif queue_name == "parser":
        q = django_rq.get_queue("parser")
    else:
        q = django_rq.get_queue("default")

    while True:
        current_job = q.dequeue()
        if not current_job:
            break
        current_job.delete()


###############################
######## PARSING TASKS ########
###############################

@job
def get_new_sources_from_iati_api():
    from django.core import management
    management.call_command('get_new_sources_from_iati_registry', verbosity=0, interactive=False)


@job
def add_new_sources_from_registry_and_parse_all():
    queue = django_rq.get_queue("default")
    queue.enqueue(get_new_sources_from_iati_api)
    queue.enqueue(parse_all_existing_sources)


@job
def force_parse_all_existing_sources():
    """
    First parse all organisation sources, then all activity sources
    """
    queue = django_rq.get_queue("parser")

    for e in Dataset.objects.all().filter(filetype=2):
        queue.enqueue(force_parse_source_by_url, args=(e.source_url,), timeout=14400)

    for e in Dataset.objects.all().filter(filetype=1):
        queue.enqueue(force_parse_source_by_url, args=(e.source_url,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def parse_all_existing_sources():
    """
    First parse all organisation sources, then all activity sources
    """
    queue = django_rq.get_queue("parser")

    for e in Dataset.objects.all().filter(filetype=2):
        queue.enqueue(parse_source_by_url, args=(e.source_url,), timeout=14400)

    for e in Dataset.objects.all().filter(filetype=1):
        queue.enqueue(parse_source_by_url, args=(e.source_url,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def parse_all_sources_by_publisher_ref(org_ref):
    queue = django_rq.get_queue("parser")
    for e in Dataset.objects.filter(publisher__publisher_iati_id=org_ref):
        queue.enqueue(parse_source_by_url, args=(e.source_url,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def force_parse_by_publisher_ref(org_ref):
    queue = django_rq.get_queue("parser")
    for e in Dataset.objects.filter(publisher__publisher_iati_id=org_ref):
        queue.enqueue(force_parse_source_by_url, args=(e.source_url,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def force_parse_source_by_url(url, update_searchable=False):
    if Dataset.objects.filter(source_url=url).exists():
        xml_source = Dataset.objects.get(source_url=url)
        xml_source.process(force_reparse=True)

    queue = django_rq.get_queue("parser")
    if update_searchable and settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def parse_source_by_url(url):
    if Dataset.objects.filter(source_url=url).exists():
        xml_source = Dataset.objects.get(source_url=url)
        xml_source.process()


@job
def calculate_activity_aggregations_per_source(source_ref):
    aac = ActivityAggregationCalculation()
    aac.parse_activity_aggregations_by_source(source_ref)


@job
def delete_source_by_id(source_id):
    try:
        Dataset.objects.get(id=source_id).delete()
    except Dataset.DoesNotExist:
        return False


@job
def delete_sources_not_found_in_registry_in_x_days(days):

    if int(days) < 6:
        raise Exception("The task queue only allows deletion of sources when not found for +5 days")

    for source in Dataset.objects.all():
        current_date = float(datetime.datetime.now().strftime('%s'))<commit_msg>make task queue parse by id instead of dataset url<commit_after>from iati.models import DocumentLink, Document
from iati_synchroniser.models import Dataset
from iati.activity_aggregation_calculation import ActivityAggregationCalculation
from django_rq import job
import django_rq
import datetime
from rq import Queue, Connection, Worker
from rq.job import Job
from redis import Redis
from django.conf import settings
import time
import requests
import os
import hashlib
from common.download_file import DownloadFile
from common.download_file import hash_file
import fulltext

redis_conn = Redis()


###############################
#### TASK QUEUE MANAGEMENT ####
###############################


@job
def remove_duplicates_from_parser_queue():
    raise Exception("Not implemented yet")


def delete_task_from_queue(job_id):
    Job.fetch(job_id, connection=redis_conn).delete()


def delete_all_tasks_from_queue(queue_name):

    if queue_name == "failed":
        q = django_rq.get_failed_queue()
    elif queue_name == "parser":
        q = django_rq.get_queue("parser")
    else:
        q = django_rq.get_queue("default")

    while True:
        current_job = q.dequeue()
        if not current_job:
            break
        current_job.delete()


###############################
######## PARSING TASKS ########
###############################

@job
def get_new_sources_from_iati_api():
    from django.core import management
    management.call_command('get_new_sources_from_iati_registry', verbosity=0, interactive=False)


@job
def add_new_sources_from_registry_and_parse_all():
    queue = django_rq.get_queue("default")
    queue.enqueue(get_new_sources_from_iati_api)
    queue.enqueue(parse_all_existing_sources)


@job
def force_parse_all_existing_sources():
    """
    First parse all organisation sources, then all activity sources
    """
    queue = django_rq.get_queue("parser")

    for e in Dataset.objects.all().filter(filetype=2):
        queue.enqueue(force_parse_source_by_id, args=(e.id,), timeout=14400)

    for e in Dataset.objects.all().filter(filetype=1):
        queue.enqueue(force_parse_source_by_id, args=(e.id,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def parse_all_existing_sources():
    """
    First parse all organisation sources, then all activity sources
    """
    queue = django_rq.get_queue("parser")

    for e in Dataset.objects.all().filter(filetype=2):
        queue.enqueue(parse_source_by_id, args=(e.id,), timeout=14400)

    for e in Dataset.objects.all().filter(filetype=1):
        queue.enqueue(parse_source_by_id, args=(e.id,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def parse_all_sources_by_publisher_ref(org_ref):
    queue = django_rq.get_queue("parser")
    for e in Dataset.objects.filter(publisher__publisher_iati_id=org_ref):
        queue.enqueue(parse_source_by_id, args=(e.id,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def force_parse_by_publisher_ref(org_ref):
    queue = django_rq.get_queue("parser")
    for e in Dataset.objects.filter(publisher__publisher_iati_id=org_ref):
        queue.enqueue(force_parse_source_by_id, args=(e.id,), timeout=14400)

    if settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def force_parse_source_by_url(url, update_searchable=False):
    if Dataset.objects.filter(source_url=url).exists():
        xml_source = Dataset.objects.get(source_url=url)
        xml_source.process(force_reparse=True)

    queue = django_rq.get_queue("parser")
    if update_searchable and settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def force_parse_source_by_id(source_id, update_searchable=False):
    if Dataset.objects.filter(pk=source_id).exists():
        xml_source = Dataset.objects.get(source_url=url)
        xml_source.process(force_reparse=True)

    queue = django_rq.get_queue("parser")
    if update_searchable and settings.ROOT_ORGANISATIONS:
        queue.enqueue(start_searchable_activities_task, args=(0,), timeout=300)


@job
def parse_source_by_url(url):
    if Dataset.objects.filter(source_url=url).exists():
        xml_source = Dataset.objects.get(source_url=url)
        xml_source.process()


@job
def parse_source_by_id(source_id):
    if Dataset.objects.filter(pk=source_id).exists():
        xml_source = Dataset.objects.get(source_url=url)
        xml_source.process()


@job
def calculate_activity_aggregations_per_source(source_ref):
    aac = ActivityAggregationCalculation()
    aac.parse_activity_aggregations_by_source(source_ref)


@job
def delete_source_by_id(source_id):
    try:
        Dataset.objects.get(id=source_id).delete()
    except Dataset.DoesNotExist:
        return False


@job
def delete_sources_not_found_in_registry_in_x_days(days):

    if int(days) < 6:
        raise Exception("The task queue only allows deletion of sources when not found for +5 days")

    for source in Dataset.objects.all():
        current_date = float(datetime.datetime.now().strftime('%s'))
{'fraction_non_alphanumeric': 0.1227896868517238, 'fraction_numerical': 0.01165662353057394, 'mean_word_length': 4.7818389491719016, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3273051', 'n_tokens_mistral': 3382, 'n_tokens_neox': 3226, 'n_words': 602}