app.py

import os
import requests
import threading
import time
from datetime import datetime, timedelta
from pyairtable import Table
from flask import Flask, request, jsonify
from dateutil.parser import parse

# --- Configuration ---
print("⚙️ CONFIG Loading environment variables...", flush=True)
AIRTABLE_API_KEY = os.getenv("AIRTABLE_API_KEY")
AIRTABLE_BASE_ID = os.getenv("AIRTABLE_BASE_ID")
GEELARK_API_KEY = os.getenv("GEELARK_API_KEY")
GEELARK_APP_ID = os.getenv("GEELARK_APP_ID")
GEELARK_API_URL = "https://openapi.geelark.com/open/v1"
TABLE_MAIN = "Main Table"
TABLE_MESSAGES = "Telegram Message"
DAY_SETUP = 1
AUTOMATION_NAME = "IG Setup - Day 1"
SEMAPHORE = threading.Semaphore(5)

if not AIRTABLE_API_KEY or not AIRTABLE_BASE_ID:
    raise EnvironmentError("❌ Missing AIRTABLE_API_KEY or AIRTABLE_BASE_ID")
if not GEELARK_API_KEY or not GEELARK_APP_ID:
    raise EnvironmentError("❌ Missing GEELARK_API_KEY or GEELARK_APP_ID")

print("[CONFIG] Initializing Airtable tables...", flush=True)
main_table = Table(AIRTABLE_API_KEY, AIRTABLE_BASE_ID, TABLE_MAIN)
message_table = Table(AIRTABLE_API_KEY, AIRTABLE_BASE_ID, TABLE_MESSAGES)

# --- Global Lock ---
main_lock = threading.Lock()
last_trigger_ts = datetime.utcnow()

# --- Helpers ---
def geelark_api(endpoint, payload=None, method="POST"):
    headers = {
        "Content-Type": "application/json",
        "Authorization": GEELARK_API_KEY,
        "appId": GEELARK_APP_ID
    }
    url = f"{GEELARK_API_URL}/{endpoint.lstrip('/')}"
    print(f"🌐 API Calling {endpoint}...", flush=True)
    response = requests.post(url, headers=headers, json=payload) if method == "POST" else requests.get(url, headers=headers, params=payload)
    if response.ok:
        return response.json()
    print(f"🚨 ERROR {endpoint} failed: {response.text}", flush=True)
    return None

def get_flow_id_by_name(flow_name):
    print("📊 DEBUG Retrieving all available flows...", flush=True)
    print(f"ℹ️ INFO Looking up flowId for '{flow_name}'...", flush=True)
    response = geelark_api("/task/flow/list", payload={"page": 1, "pageSize": 100})
    print(f"🧾 Full flow list response: {response}", flush=True)
    if response and "data" in response and "items" in response["data"]:
        for flow in response["data"]["items"]:
            if flow.get("title") == flow_name:
                print(f"✅ FOUND Flow ID: {flow.get('id')}", flush=True)
                return flow.get("id")
    print(f"[ERROR] Flow name '{flow_name}' not found.", flush=True)
    return None

def get_today_latest_schedule():
    print("🕒 SCHEDULE Calculating latest scheduleAt...", flush=True)
    today_str = datetime.utcnow().strftime("%Y-%m-%d")
    records = main_table.all(formula=f"AND(IS_SAME({{Date Task ID}}, '{today_str}', 'day'), {{Last Task ID}} != '')")
    records.sort(key=lambda r: r['fields'].get("Date Task ID", ''), reverse=True)
    if not records:
        print("[SCHEDULE] No previous tasks today. Using now().", flush=True)
        return datetime.utcnow()
    last_start = parse(records[0]['fields'].get("Date Task ID"))
    max_duration = max([int(r['fields'].get("Task Duration", 0)) for r in records[:5]])
    print(f"[SCHEDULE] Last task ends at: {last_start + timedelta(seconds=max_duration)}", flush=True)
    return last_start + timedelta(seconds=max_duration)

def build_param_map(fields):
    return {
        "Old Username": fields.get("Old Username", ""),
        "Password": fields.get("Password", ""),
        "Email": fields.get("Email", ""),
        "Email PW": fields.get("Email PW", "")
    }

def create_rpa_task(profile, flow_id, schedule_at):
    param_map = build_param_map(profile['fields'])
    payload = {
        "flowId": flow_id,
        "scheduleAt": int(schedule_at.timestamp() * 1000),
        "paramMap": param_map
    }
    print(f"📝 TASK Creating task for profile {profile['fields'].get('Old Username')}...", flush=True)
    response = geelark_api("/task/rpa/add", payload)
    if response and response.get("data"):
        return response["data"].get("taskId")
    return None

def process_profile(profile, flow_id, schedule_at):
    with SEMAPHORE:
        record_id = profile["id"]
        fields = profile["fields"]
        print(f"🔄 PROCESS Processing profile {fields.get('Old Username')}...", flush=True)

        if not fields.get("Start Date Setup"):
            main_table.update(record_id, {"Start Date Setup": datetime.utcnow().strftime("%Y-%m-%d")})

        duration = fields.get("Task Duration")
        if not duration:
            duration = 270
            main_table.update(record_id, {"Task Duration": duration})

        task_id = create_rpa_task(profile, flow_id, schedule_at)
        if task_id:
            print(f"✅ SUCCESS Task ID: {task_id}", flush=True)
            main_table.update(record_id, {
                "Scheduled Today": True,
                "Last Task ID": task_id,
                "Panel Status": f"Scheduled Day {DAY_SETUP}",
                f"Day {DAY_SETUP} Completed?": True
            })
            message_table.create({
                "Summary Report": f"✅ Day 1 task scheduled for {fields.get('Old Username')} at {schedule_at}",
                "IG Profiles": [record_id],
                "Project Status": "Account Setup"
            })
            phone_id = fields.get("Phone ID")
            if phone_id:
                print(f"🧹 CLEANUP Stopping phone {phone_id}...", flush=True)
                geelark_api("/phone/stop", {"id": phone_id})
        else:
            print(f"❌ FAIL Task creation failed for {fields.get('Old Username')}", flush=True)
            main_table.update(record_id, {"Panel Status": "Error"})

# --- Main Logic ---
def run_batches():
    while True:
        print("📦 LOAD Loading unscheduled profiles...", flush=True)
        profiles = main_table.all(formula=f"AND({{Day Setup}} = {DAY_SETUP}, NOT({{Scheduled Today}}), {{Panel Status}} != 'Error')")
        if not profiles:
            print("🏁 DONE No more profiles to process.", flush=True)
            break

        batch = profiles[:5]
        print(f"📤 BATCH Scheduling batch of {len(batch)} profiles...", flush=True)
        schedule_at = get_today_latest_schedule()

        threads = []
        for profile in batch:
            t = threading.Thread(target=process_profile, args=(profile, flow_id, schedule_at))
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

        time.sleep(3)

# --- Entry Point ---
app = Flask(__name__)

@app.route("/run", methods=["POST"])
def run():
    global flow_id
    print("🚀 /run endpoint triggered", flush=True)
    flow_id = get_flow_id_by_name(AUTOMATION_NAME)
    if flow_id:
        run_batches()
        return jsonify({"status": "success"}), 200
    else:
        return jsonify({"status": "error", "message": "Flow ID not found"}), 400

if __name__ == "__main__":
    print("🚀 Script Day 1 starting...", flush=True)
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 5000)))