Test scheduler
import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from ortools.sat.python import cp_model
import requests
from dotenv import load_dotenv
load_dotenv()
app = FastAPI()
# HubSpot API
HUBSPOT_API_KEY = os.getenv(“HUBSPOT_API_KEY”)
HUBSPOT_API_URL = “https://api.hubapi.com/crm/v3/objects/emails”
# Data Models
class Preferences(BaseModel):
clients: dict # e.g., {“Client1”: [“CompanyA”, “CompanyB”], …}
# AI-driven scheduling endpoint
@app.post(“/schedule”)
def schedule_meetings(prefs: Preferences):
clients = list(prefs.clients.keys())
companies = list({comp for comps in prefs.clients.values() for comp in comps})
time_slots = [f”Slot_{i+1}” for i in range(12)]
model = cp_model.CpModel()
meetings = {}
for c in clients:
for comp in prefs.clients[c]:
for slot in time_slots:
meetings[(c, comp, slot)] = model.NewBoolVar(f”{c}_{comp}_{slot}”)
for c in clients:
for comp in prefs.clients[c]:
model.AddExactlyOne(meetings[(c, comp, slot)] for slot in time_slots)
for slot in time_slots:
for c in clients:
model.Add(sum(meetings.get((c, comp, slot), 0) for comp in companies) <= 1)
for comp in companies:
model.Add(sum(meetings.get((c, comp, slot), 0) for c in clients) <= 1)
solver = cp_model.CpSolver()
status = solver.Solve(model)
if status != cp_model.OPTIMAL:
raise HTTPException(status_code=400, detail=”No feasible solution found.”)
schedule = {}
for slot in time_slots:
for c in clients:
for comp in prefs.clients[c]:
if solver.Value(meetings[(c, comp, slot)]):
schedule.setdefault(slot, []).append({“client”: c, “company”: comp})
# Trigger emails through HubSpot
send_notifications_via_hubspot(schedule)
return schedule
# HubSpot email sender function
def send_notifications_via_hubspot(schedule):
headers = {
“Content-Type”: “application/json”,
“Authorization”: f”Bearer {HUBSPOT_API_KEY}”
}
for slot, meetings in schedule.items():
for meeting in meetings:
client_email = f”{meeting[‘client’].lower()}@example.com”
company_email = f”{meeting[‘company’].lower()}@example.com”
# Email payload example
payload_client = {
“properties”: {
“hs_email_html”: f”
You have a meeting scheduled with {meeting[‘company’]} at {slot}.
“,
“subject”: “Meeting Confirmation”,
“from_email”: “organizer@yourcompany.com”,
“to”: client_email
}
}
payload_company = {
“properties”: {
“hs_email_html”: f”
You have a meeting scheduled with {meeting[‘client’]} at {slot}.
“,
“subject”: “Meeting Confirmation”,
“from_email”: “organizer@yourcompany.com”,
“to”: company_email
}
}
# Send emails via HubSpot API
requests.post(HUBSPOT_API_URL, json=payload_client, headers=headers)
requests.post(HUBSPOT_API_URL, json=payload_company, headers=headers)