"""
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import requests

api_endpoint = "https://test-openbeta-herofi.bravestars.games/api"
api_key = "<KEY>"



default_args = {
    "owner": "anhnd",
    "depends_on_past": False,
    "start_date": datetime(2021, 11, 1),
    "email": ["<EMAIL>"],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}

def start_task():
    return 'Start process Data'


def create_snapshot_hero_task():
    print("Create snapshot hero")
    api_url = api_endpoint + "/reward/snapshot/hero?key=" + api_key + "&api-version=1.0"
    r = requests.get(api_url)
    print("api endpoint " + api_url)
    if r.status_code == 200:
        print("Process completed")
    else:
        print("Failed")

def summary_quest_reward_task():
    print("Summary quest reward")
    api_url = api_endpoint + "/reward/summary/quest?key=" + api_key + "&api-version=1.0"
    print("api endpoint " + api_url)
    r =requests.get(api_url)
    if r.status_code == 200:
        print("Summary quest reward: Success")
    else:
        print("Summary quest reward: Failed")

def summary_arena_reward_task():
    print("Summary arena reward")
    api_url = api_endpoint + "/reward/summary/arena?key=" + api_key + "&api-version=1.0"
    print("api endpoint " + api_url)
    r =requests.get(api_url)

    if r.status_code == 200:
        print("Summary arena reward: Success")
    else:
        print("Summary arena reward: Failed")



dag = DAG("process_daily_data",
            default_args=default_args,
            schedule_interval='15 0 * * *')

t1 = PythonOperator(task_id='start_task', python_callable=start_task, dag=dag)
t2 = PythonOperator(task_id='create_snapshot_hero_task', python_callable=create_snapshot_hero_task, dag=dag)
t3 = PythonOperator(task_id='summary_quest_reward_task', python_callable=summary_quest_reward_task, dag=dag)
t4 = PythonOperator(task_id='summary_arena_reward_task', python_callable=summary_arena_reward_task, dag=dag)

t2.set_upstream(t1)
t3.set_upstream(t2)
t4.set_upstream(t2)


"""
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import requests

api_endpoint = "https://test-openbeta-herofi.bravestars.games/api"
api_key = "<KEY>"



default_args = {
    "owner": "anhnd",
    "depends_on_past": False,
    "start_date": datetime(2021, 11, 1),
    "email": ["<EMAIL>"],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}

def start_task():
    return 'Start process Data'


def create_snapshot_hero_task():
    print("Create snapshot hero")
    api_url = api_endpoint + "/reward/snapshot/hero?key=" + api_key + "&api-version=1.0"
    r = requests.get(api_url)
    print("api endpoint " + api_url)
    if r.status_code == 200:
        print("Process completed")
    else:
        print("Failed")

def summary_quest_reward_task():
    print("Summary quest reward")
    api_url = api_endpoint + "/reward/summary/quest?key=" + api_key + "&api-version=1.0"
    print("api endpoint " + api_url)
    r =requests.get(api_url)
    if r.status_code == 200:
        print("Summary quest reward: Success")
    else:
        print("Summary quest reward: Failed")

def summary_arena_reward_task():
    print("Summary arena reward")
    api_url = api_endpoint + "/reward/summary/arena?key=" + api_key + "&api-version=1.0"
    print("api endpoint " + api_url)
    r =requests.get(api_url)

    if r.status_code == 200:
        print("Summary arena reward: Success")
    else:
        print("Summary arena reward: Failed")



dag = DAG("process_daily_data",
            default_args=default_args,
            schedule_interval='15 0 * * *')

t1 = PythonOperator(task_id='start_task', python_callable=start_task, dag=dag)
t2 = PythonOperator(task_id='create_snapshot_hero_task', python_callable=create_snapshot_hero_task, dag=dag)
t3 = PythonOperator(task_id='summary_quest_reward_task', python_callable=summary_quest_reward_task, dag=dag)
t4 = PythonOperator(task_id='summary_arena_reward_task', python_callable=summary_arena_reward_task, dag=dag)

t2.set_upstream(t1)
t3.set_upstream(t2)
t4.set_upstream(t2)


Code that goes along with the Airflow located at: http://airflow.readthedocs.org/en/latest/tutorial.html # 'queue': 'bash_queue', # 'pool': 'backfill', # 'priority_weight': 10, # 'end_date': datetime(2016, 1, 1),