from flask import Blueprint, request import requests import logging from datetime import datetime from firebase.firebase import Firebase import pytz import utility.function as func import os import uuid from bigquery.bq import BigQuery from feature.audience.craft import BQCraft, Query from feature.automationV2.src.models.execution_context import ExecutionContext from feature.automationV2.src.models import Workflow from feature.automationV2.src import nodes from feature.automationV2.src.engine import NodeFactory from feature.automationV2.src.integrations.database_connection import DatabaseConnection from feature.automationV2.src.integrations.database_logger import DatabaseLogger logging.basicConfig(level=logging.INFO) feature_automation_v2 = Blueprint('feature_automation_v2', __name__, url_prefix='/feature/automation/v2') fb = Firebase(host=os.environ.get("FIREBASE_HOST")) timezone = pytz.timezone('Asia/Bangkok') @feature_automation_v2.route('/', methods=['GET', 'POST', 'PUT', 'DELETE']) @feature_automation_v2.route('', methods=['GET', 'POST', 'PUT', 'DELETE']) def feature_automation_v2_func(): now_request = datetime.now().utcnow().isoformat() try: if request.method == 'POST': automationJSON = request.get_json() if type(automationJSON) != dict: return {'status': 'error', 'message': f"Automation flow must provide as dict"}, 500 uuidStr = str(uuid.uuid4()) automationJSON['automation_id'] = uuidStr if 'automation_id' not in automationJSON: return {'status': 'error', 'message': 'automation_id is require'}, 500 db_conn = DatabaseConnection() database_logger = DatabaseLogger(db_connection=db_conn) wf = Workflow(automationJSON['automation_id'], automationJSON, database_logger=database_logger ) validate_json = wf.validate_structure() if validate_json: return {'status': 'error', 'erors': validate_json}, 500 wf.install_automation_context(fb, now_request) start_node = automationJSON['nodes'][wf.start_node_id] start_node['node_id'] = wf.start_node_id node_factory = NodeFactory() node_factory.register_node_type('trigger', nodes.TriggerNode) trigger_node = node_factory.create_node(start_node, wf.property_id, database_logger=database_logger) schedule, schedule_datetime = trigger_node.generate_schedule(start_node['datebegin'], start_node['dateend'], wf.settings['timezone']) database_logger.generate_trigger(wf.property_id, wf.workflow_id, wf.version, schedule) return {'status': 'ok', 'automation_id': uuidStr, 'message': f"Automation {uuidStr} setup successfully"}, 200 elif request.method == 'GET': data = request.args for req in ['property_id', 'automation_id']: if req not in data: return {"status":"error",'message': f"{req} is require"}, 400 property_id = data.get('property_id') automation_id = data.get('automation_id') display_type = data.get('display_type', 'list') if display_type == 'list': if automation_id != 'all': automation_context = fb.db.reference().child(f"account/{property_id}/automation_v2/{automation_id}").get() ordered_automation_desc = [automation_context] else: automation_context = fb.db.reference().child(f"account/{property_id}/automation_v2").get() if automation_context: returnAutomation = [] for a in automation_context: au = automation_context[a] if au['display']: au.update({'automation_id': a}) returnAutomation.append(au) ordered_automation_desc = sorted(returnAutomation, key=lambda x: x['createdate'], reverse=True) else: return {'status': 'not fond', 'message': f'Propery {property_id} has no automation flow'}, 400 #Time zone display for auto in ordered_automation_desc: for key in ["createdate", "lastupdate"]: if key in auto and auto[key]: try: # Parse ISO datetime string (assuming UTC) dt = datetime.fromisoformat(auto[key]) if dt.tzinfo is None: dt = dt.replace(tzinfo=pytz.UTC) # Convert to GMT+7 dt_gmt7 = dt.astimezone(timezone) # Store back as ISO string for readability auto[key] = dt_gmt7.isoformat() except Exception as e: logging.error(f"Error converting {key}: {e}") return {'status': 'ok', 'data': ordered_automation_desc}, 200 elif display_type == 'calendar': date_start = data.get('date_start', None) date_end = data.get('date_end', None) status = data.get('status', 'all') if not date_start and not date_end: return {'status': 'not fond', 'message': f'Disply type calendar must provide date_start and date_end'}, 400 archiveList = fb.db.reference().child(f"account/{property_id}/automation_archive").get() archiveList = [] if not archiveList else archiveList from utility.cloudsql import CloudSQL from dashboard.genQuery import GenerateQuery import pandas as pd genq = GenerateQuery(property_id=property_id) csql = CloudSQL("asia-southeast1", os.environ.get("INSTANCE_NAME")) csql.create_engine(os.environ.get("POSTGRES_USER"), os.environ.get("POSTGRES_PASSWORD"), os.environ.get("POSTGRES_DB")) query = genq.getAutomationCalendarV2(date_start, date_end, status) result = csql.query(query) df_calendar = pd.DataFrame(result, columns=['datetime_trigger', 'automation_id', 'automation_name', 'status', 'active_status']) if df_calendar.empty: return {'status': 'ok', 'data': []}, 200 df_calendar['datetime_trigger'] = ( pd.to_datetime(df_calendar['datetime_trigger'], utc=True).dt.tz_convert('Asia/Bangkok').dt.strftime('%Y-%m-%d %H:%M:%S') ) if archiveList: df_calendar = df_calendar[~df_calendar['automation_id'].isin(archiveList)] df_calendar.sort_values(by='datetime_trigger', inplace=True) result_json = df_calendar.to_dict(orient='records') return {'status': 'ok', 'data': result_json}, 200 elif request.method == 'PUT': automationJSON = request.get_json() if type(automationJSON) != dict: return {'status': 'error', 'message': f"Automation flow must provide as dict"}, 500 if 'automation_id' not in automationJSON: return {'status': 'error', 'message': 'automation_id is require'}, 500 db_conn = DatabaseConnection() database_logger = DatabaseLogger(db_connection=db_conn) wf = Workflow(automationJSON['automation_id'], automationJSON, database_logger=database_logger ) validate_json = wf.validate_structure() if validate_json: return {'status': 'error', 'erors': validate_json}, 500 #update wf.update_automation_context(fb, now_request) return {'status': 'ok', 'message': f"Automation {automationJSON['automation_id']} update successfully"}, 200 elif request.method == 'DELETE': data = request.get_json() for req in ['property_id', 'automation_id']: if req not in data: return {'message': f"{req} is require"}, 400 property_id = data.get('property_id') automation_id = data.get('automation_id') #Get archive archiveList = fb.db.reference().child(f"account/{property_id}/automation_archive").get() archiveList = [] if not archiveList else archiveList archiveList.append(automation_id) setArchive = list(set(archiveList)) fb.db.reference().child(f"account/{property_id}/automation_archive").set(setArchive) fb.db.reference().child(f"account/{property_id}/automation_v2/{automation_id}/display").set(False) #update trigger manager db_conn = DatabaseConnection() database_logger = DatabaseLogger(db_connection=db_conn) result = database_logger.update_trigger_mannager(property_id, automation_id, 0) result = database_logger.update_work_flow_status(property_id, automation_id, 3) return {'status': 'ok', 'message': f"Autoantion ID {automation_id} was archive"}, 200 return {'status': 'error', 'message': 'Method not allowed'}, 405 except Exception as e: logging.error(e) return {'status': 'error', 'message': f"{e}"},500 @feature_automation_v2.route('/status', methods=['POST']) @feature_automation_v2.route('/status/', methods=['POST']) def feature_automatio_v2_status(): try: if request.method == 'POST': data = request.get_json() for req in ['property_id', 'automation_id', 'status']: if req not in data: return {'message': f"{req} is require"}, 400 property_id = data.get("property_id") automation_id = data.get("automation_id") status = data.get("status") try: status = bool(status) except ValueError: return {'message': f'''status value must be boolean type'''}, 400 #Check audience exist automationExist = fb.db.reference().child(f"account/{property_id}/automation_v2/{automation_id}").get(shallow=True) if not automationExist: return {'status': 'not fond', 'message': f'Automation Id {automation_id} not fond'},400 fb.db.reference().child(f"account/{property_id}/automation_v2/{automation_id}/active").set(status) #Update workflow and trigger manager db_conn = DatabaseConnection() database_logger = DatabaseLogger(db_connection=db_conn) result = database_logger.update_work_flow_status(property_id, automation_id, status) result = database_logger.update_trigger_mannager(property_id, automation_id, status) return {'status': 'ok', 'message': f"Autoantion ID {automation_id} was updated status to {'active' if status else 'inactive'}"}, 200 return {'status': 'error', 'message': 'Method not allowed'}, 405 except Exception as e: logging.error(e) return {'status': 'error', 'message': f"{e}"},500 @feature_automation_v2.route('/trigger-manager', methods=['POST']) @feature_automation_v2.route('/trigger-manager/', methods=['POST']) def feature_automation_v2_trigger_manager(): try: from feature.automationV2.trigger_manager import execute_workflow execute_workflow() return {'status': 'ok', 'message': 'All workflow in this schedual run successfully'}, 200 except Exception as e: logging.error(e) return {'status': 'error', 'message': f"{e}"},500 @feature_automation_v2.route('/trigger-waiting', methods=['POST']) @feature_automation_v2.route('/trigger-waiting/', methods=['POST']) def feature_automation_v2_trigger_waiting(): try: from feature.automationV2.trigger_waiting import execute_workflow_waiting status, message = execute_workflow_waiting() return {'status': 'ok', 'message': message}, 200 except Exception as e: logging.error(e) return {'status': 'error', 'message': f"{e}"},500 @feature_automation_v2.route('/manual', methods=['POST']) @feature_automation_v2.route('/manual/', methods=['POST']) def feature_automation_v2_manual(): try: data = request.get_json() for req in ['property_id', 'automation_id']: if req not in data: return {"status":"error",'message': f"{req} is require"}, 400 property_id = data.get("property_id") automation_id = data.get("automation_id") from feature.automationV2.trigger_manual import execute_workflow_manual data = execute_workflow_manual(property_id, automation_id) return {'status': 'ok', 'data': data}, 200 except Exception as e: logging.error(e) return {'status': 'error', 'message': f"{e}"},500 @feature_automation_v2.route('/execution', methods=['GET']) @feature_automation_v2.route('/execution/', methods=['GET']) def feature_automatio_v2_execution(): try: if request.method == 'GET': db_conn = DatabaseConnection() database_logger = DatabaseLogger(db_connection=db_conn) data = request.args for req in ['property_id', 'automation_id', 'execution_id']: if req not in data: return {"status":"error",'message': f"{req} is require"}, 400 property_id = data.get('property_id') automation_id = data.get('automation_id') execution_id = data.get('execution_id', 'all') result = database_logger.get_automation_execution(property_id, automation_id, execution_id) if result: return {'status': 'ok', 'data': result}, 200 else: return {'status': 'ok', 'data': []}, 200 return {'status': 'error', 'message': 'Method not allowed'}, 405 except Exception as e: logging.error(e) return {'status': 'error', 'message': f"{e}"},500 @feature_automation_v2.route('/execution/data', methods=['GET']) @feature_automation_v2.route('/execution/data/', methods=['GET']) def feature_automatio_v2_execution_data(): try: if request.method == 'GET': db_conn = DatabaseConnection() database_logger = DatabaseLogger(db_connection=db_conn) data = request.args for req in ['property_id', 'automation_id', 'execution_id']: if req not in data: return {"status":"error",'message': f"{req} is require"}, 400 property_id = data.get('property_id') automation_id = data.get('automation_id') execution_id = data.get('execution_id') result = database_logger.get_execution_data(property_id, automation_id, execution_id) if result: return {'status': 'ok', 'data': result}, 200 else: return {'status': 'ok', 'data': []}, 200 return {'status': 'error', 'message': 'Method not allowed'}, 405 except Exception as e: logging.error(e) return {'status': 'error', 'message': f"{e}"},500