"""
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.docker_operator import DockerOperator
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta
# from sites import ALL_CRAWLERS
ALL_CRAWLERS = {
    'indeed|daily': ['general', 'SDR', 'accounting', 'blockchain','reactjs',"qa","cctv"],
    'dice|daily': ['general', 'SDR', 'accounting', 'blockchain', 'reactjs',"qa","cctv"],
#    'monster|daily': ['general', 'SDR', 'accounting', 'blockchain', 'reactjs',"qa","cctv"],
    'glassdoor|daily': ['general', 'SDR', 'accounting', 'blockchain','reactjs',"cctv"],
#    'monster_ch|daily': ['general', 'SDR', 'accounting', 'blockchain','reactjs',"cctv"],
    'dejobs|daily': ['general'],
    'seek_au|daily': ['general', 'machine_learning',"cctv"],
    'careerbuilder|daily': ['general'],
#    'datajobs|monthly': ['general', 'machine_learning',"cctv"],
    'whoishiring|daily': ['general','cctv'],
#    'cryptojobslist|monthly': ['general'],
    'crypto|daily': ['general'],
    'linkedin|hourly': ['general','cctv'],
#    'stackoverflow|mothly': ['general', 'SDR', 'accounting', 'blockchain','reactjs','cctv'],
    'simplyhired|hourly':['general'],
#    'joblift|weekly':['general'],
#    'adzuna|weekly':['general'],
#    'powertofly|weekly':['general'],
    'snagajobs|daily':[],
    'ladders|daily':[],
    'flexjobs|daily':[],
    'linkup|hourly':[],
#    'naukri|weekly':[],
    'polemployee|daily':[],
    'indeed_china|daily':[],
#    'ranstad|weekly':[],
#    'eurojobs|monthly':[],
#    'ohiomonster|monthly':[],
#    'nurse|weekly':[],
#    'veterans|weekly':[],
#    'bankcanada|weekly':[],
#    'govukjobs|weekly':[],
#    'chinacities|weekly':[],
    'reeduk|daily':[],
#    'sgjobsdb|weekly':[],
#    'usagov|weekly':[],
#    'govermentjobs|weekly':[],
    'indeed_ar|daily':[],
    'indeed_au|daily':[],
    'indeed_brasil|daily':[],
    'indeed_canada|daily':[],
    'indeed_eg|daily':[],
    'indeed_es|daily':[],
    'indeed_fi|daily':[],
    'indeed_fr|daily':[],
    'indeed_gr|daily':[],
    'indeed_ie|daily':[],
    'indeed_in|daily':[],
    'indeed_it|daily':[],
    'indeed_korea|daily':[],
    'indeed_malaysia|daily':[],
    'indeed_mx|daily':[],
    'indeed_nl|daily':[],
    'indeed_pk|daily':[],
    'indeed_pl|daily':[],
    'indeed_pt|daily':[],
    'indeed_ro|daily':[],
    'indeed_ru|daily':[],
    'indeed_sa|daily':[],
    'indeed_se|daily':[],
    'indeed_tw|daily':[],
    'indeed_ua|daily':[],
    'indeed_uk|daily':[],
    'optioncar|daily':[],
    'opcionempleo|daily':[],
#    'jobsinnetwork|weekly':[],
#    'computrabajo|weekly':[],
#    'canadajobs|monthly':[],
#    'jobillico|weekly':[],
    'bcjobs|hourly':[],
#    'jora|monthly':[],
#    'workbc|weekly':[],
#    'trabajosmx|weekly':[],
#    'joboolo|weekly':[],
#    'bayt|weekly':[],
#    'bestjobs|weekly':[],
#    'careercast|weekly':[],
#    'cvlib|weekly':[],
#    'directemploi|weekly':[],
#    'disabled|weekly':[],
#    'eurabota|weekly':[],
#    'findojobs|weekly':[],
#    'jobsonlinenl|weekly':[],
#    'livecareer|weekly':[],
#    'meteojob|weekly':[],
#    'postjobsfree|weekly':[],
#    'sercanto|weekly':[],
#    'timesjobs|weekly':[],
#    'xing|weekly':[],
#    'avito|weekly':[],
#    'juju|weekly':[],
#    'jobbird|weekly':[],

}

default_args = {
    "owner": "airflow",
    "start_date": datetime(2020, 12, 17),
    "email": ["airflow@airflow.com"],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 2,
    "depends_on_past": True,
    "retry_delay": timedelta(minutes=30),
    'pool': 'general',
}

dag = DAG("crawlers", default_args=default_args,
          schedule_interval="@daily", catchup=False)

#default_args_weekly = dict(default_args)
default_args_hourly = dict(default_args)
#default_args_monthly = dict(default_args)

default_args['start_date'] = datetime(2020, 12, 17)
default_args_hourly['start_date'] = datetime(2020, 12, 17)
#default_args_weekly['start_date'] = datetime(2020, 12, 15)
#default_args_monthly['start_date'] = datetime(2020,12,1)

dag_hourly = DAG("crawlers_hourly", default_args=default_args_hourly,
          schedule_interval="@hourly", catchup=False)

#dag_weekly = DAG("crawlers_weekly", default_args=default_args_weekly,
#          schedule_interval="@weekly", catchup=False)
# t1, t2 and t3 are examples of tasks created by instantiating operators
#dag_monthly = DAG('crawlers_monthly',default_args=default_args_monthly,schedule_interval="@monthly",catchup=False)


def load_sites(period):
    """
    period loader
    """
    def return_site_scopes():
        new_dict = {}
        for _site, scopes in ALL_CRAWLERS.items():
            if _site.split('|')[-1].strip() == period:
                new_dict[_site.split('|')[0]] = scopes
        return new_dict
    return return_site_scopes


daily_site_loader = load_sites("daily")
hourly_site_loader = load_sites("hourly")
#weekly_site_loader = load_sites("weekly")
#monthly_site_loader = load_sites("monthly")
dags_mapping = [
    (dag, daily_site_loader()),
    (dag_hourly, hourly_site_loader()),
 #   (dag_weekly, weekly_site_loader()),
 #   (dag_monthly, monthly_site_loader())
]

for _dag, _crawl in dags_mapping:
    site_crawls = []
    wait_task = DummyOperator(task_id="Wait", dag=_dag)
    start_task = DummyOperator(task_id="Start", dag=_dag)
    end_task = DummyOperator(task_id="End", dag=_dag)
    for site, scopes in _crawl.items():
        scope_tasks = []
        task_id = "-".join([site, "crawl"])
        command = 'python -m scrapy crawl {}'.format(site)
        task_to_run = DockerOperator(
            task_id=task_id,
            image='oc:airflow',
            command=command,
            docker_url='unix://var/run/docker.sock',
            network_mode='host',
            dag=_dag
        )
        site_crawls.append(task_to_run)
    start_task >> site_crawls >> wait_task >> end_task