add workflow 泰克CRM,dev

dev/workflow/TK_Cust/tk_crm/泰克CRM/wf_dag_tk_crm.py

@@ -23,7 +23,7 @@ default_args = {
 }
 
 dag = DAG('wf_dag_tk_crm', default_args=default_args,
-schedule_interval="0 0 * * *",
+schedule_interval="0 18 * * 5",
 catchup=False,
 dagrun_timeout=timedelta(minutes=160),
 max_active_runs=3)
@@ -209,6 +209,27 @@ params={'my_param':"d_crm_contact_agi"},
 depends_on_past=False, 
 retries=3, 
 dag=dag)
+cust_contact_mapping = SSHOperator(
+ssh_hook=sshHook,
+task_id='cust_contact_mapping',
+command='/data/airflow/etl/COM/run_psql.sh {{ ds_nodash }} {{params.my_param}} >>/data/airflow/logs/run_tpt_{{ds_nodash}}.log 2>&1 ',
+params={'my_param':"cust_contact_mapping_agi"},
+depends_on_past=False, 
+retries=3, 
+dag=dag)
+cust_contact_info = SSHOperator(
+ssh_hook=sshHook,
+task_id='cust_contact_info',
+command='/data/airflow/etl/COM/run_psql.sh {{ ds_nodash }} {{params.my_param}} >>/data/airflow/logs/run_tpt_{{ds_nodash}}.log 2>&1 ',
+params={'my_param':"cust_contact_info_agi"},
+depends_on_past=False, 
+retries=3, 
+dag=dag)
+trigger_trg_cust = TriggerDagRunOperator(
+ssh_hook=sshHook,
+task_id='trigger_trg_cust',
+trigger_dag_id="wf_dag_tk_cust",
+dag=dag)
 file_CRM_account >> crm_account_4545
 file_CRM_contact_ccp >> crm_contact_ccp_5681
 file_CCP_mapping_table >> ccp_mapping_table_8972
@@ -221,5 +242,8 @@ crm_account_4545 >> t01_crm_account
 file_crm_opp >> crm_opp
 crm_opp >> t01_crm_opportunity
 t01_crm_contact >> d_crm_contact
-d_crm_contact >> task_failed 
+d_crm_contact >> cust_contact_mapping
+cust_contact_mapping >> cust_contact_info
+cust_contact_info >> trigger_trg_cust
+trigger_trg_cust >> task_failed 
 file_china_city >> task_failed