---

from airflow.utils.dates import days_ago
from service.notify_failure import notify_failure
from processing_class import processing_postgre_to_mssql as f
from processing_class import processing_file_to_db as d
from airflow.models.dag import DAG
from airflow.models import Variable


default_args = {
    'owner': 'gfc',
    'depends_on_past': False,
    # 'start_date': days_ago(2),
    'retries': 0,
    'on_failure_callback': notify_failure,
}

tags = ['gfc']

root_directory_files = Variable.get('directory_upp')  # \\bi.oos.ru\bi-data\13 onec\02 upp-uu
check_gfc = root_directory_files + r'\Проверка GFC'
smb_connection_id = 'folder_bi'

connector_id_stage = 'mssql_dwh'
connector_id_source = 'GoodsForeCast'
db_source_stage_qa = 'GFC.SSNS.Stage.QA'

db_stage = 'Stage'
db_core = 'dwh'
schema_source = 'public'
schema_stage = 'gfc'
schema_core = 'gfc'

with DAG(dag_id='gfc_check_gfc_1c', default_args=default_args, tags=tags, schedule_interval=None,
         catchup=False, ):
    table_name_source = 'vw_check_gfc_1c'
    table_name_stage = 'check_gfc_1c'
    table_name_core = 'check_gfc_1c'

    fields_stage = ['product_id', 'product_code', 'bom_id', 'srcbomcode', 'isfinite', 'workcenter_id',
                    'workcenter_calendar_id', 'minincrement', 'locationtoid', 'forecastplancskubody_warehouseid',
                    'warehouseproductmatrix_warehouseid', 'productunitofmeasurevalue', 'quantity', 'target_product_id',
                    'begindate', 'enddate']

    main = f.LoadPostgre(connector_id_stage, db_stage, schema_stage, table_name_stage, fields_stage,
                         connector_id_source, db_source_stage_qa, schema_source, table_name_source=table_name_source,
                         db_core=db_core, schema_core=schema_core)

    main.clear_table() >> main.insert_to_stage() >> main.delete_full_insert()

with DAG(dag_id='gfc_check_workcenters', default_args=default_args, tags=tags, schedule_interval=None,
         catchup=False, ):
    table_name_source = 'vw_check_workcenters'
    table_name_stage = 'check_workcenters'
    table_name_core = 'check_workcenters'

    fields_stage = ['code', 'workcenter', 'workcentercalendarmapping', 'workcenterconnectivity',
                    'workcenterinterimstorage', 'productid', 'name']

    main = f.LoadPostgre(connector_id_stage, db_stage, schema_stage, table_name_stage, fields_stage,
                         connector_id_source, db_source_stage_qa, schema_source, table_name_source=table_name_source,
                         db_core=db_core, schema_core=schema_core)

    main.clear_table() >> main.insert_to_stage() >> main.delete_full_insert()

with DAG(dag_id='gfc_check_products', default_args=default_args, tags=tags, schedule_interval=None,
         catchup=False, ):
    table_name_source = 'vw_check_products'
    table_name_stage = 'check_products'
    table_name_core = 'check_products'

    fields_stage = ['product_code', 'product_name', 'product', 'bom', 'bomcontent', 'workcenter_code', 'workcenter',
                    'specificationredefenition', 'workcentercalendarmapping', 'workcenterbom', 'minincrement',
                    'processingcost', 'cogs', 'outprice', 'warehouseproductmatrix', 'productunitofmeasurevalue']

    main = f.LoadPostgre(connector_id_stage, db_stage, schema_stage, table_name_stage, fields_stage,
                         connector_id_source, db_source_stage_qa, schema_source, table_name_source=table_name_source,
                         db_core=db_core, schema_core=schema_core)

    main.clear_table() >> main.insert_to_stage() >> main.delete_full_insert()

with DAG(dag_id='gfc_check_logistical_parameters', default_args=default_args, tags=tags, schedule_interval=None,
         catchup=False, ):
    table_name_source = 'vw_logistical_parameters'
    table_name_stage = 'check_logistical_parameters'
    table_name_core = 'check_logistical_parameters'

    fields_stage = ['supplier_code', 'supplier_id', 'supplier_warehouse_id', 'transportation_type_name',
                    'transportation_type_id', 'logistic_route_id', 'inprice', 'logisticscost', 'minincrement',
                    'productid', 'leadtime']

    main = f.LoadPostgre(connector_id_stage, db_stage, schema_stage, table_name_stage, fields_stage,
                         connector_id_source, db_source_stage_qa, schema_source, table_name_source=table_name_source,
                         db_core=db_core, schema_core=schema_core)

    main.clear_table() >> main.insert_to_stage() >> main.delete_full_insert()

with DAG(dag_id='gfc_error_manual', default_args=default_args, tags=tags, catchup=False, schedule_interval=None):
    file = 'Справочник ошибок.xlsx'
    name_sheet = 'Ошибки'
    table_name_stage = 'error_manual'

    field_source = ['Код ошибки', 'Группа', 'Тип ошибки', 'Вид', 'Категория', 'Владелец', 'Описание', 'Ответственный',
                    'email', 'Действия по устранению', 'Правила распознавания']
    field_stage = ['Код ошибки', 'Группа', 'Тип ошибки', 'Вид', 'Категория', 'Владелец', 'Описание', 'Ответственный',
                   'email', 'Действия по устранению', 'Правила распознавания']
    main = d.LoadFile(connector_id_stage, db_stage, 'gfc', table_name_stage, field_stage, root_directory_files +
                      r'\Качество данных', smb_connection_id, name_sheet=name_sheet, file_name=file, db_core='dc',
                      schema_core='dbo', fields_source=field_source)

    main.clear_table() >> main.insert_to_stage() >> main.merge_table(unic_field=['Код ошибки'],
                                                                     field_is_deleted=False)