import os, requests, time, base64
from collections import defaultdict
LV_KEY = os.environ["LEVER_API_KEY"]
MV = os.environ["MAVERA_API_KEY"]
LV_BASE = "https://api.lever.co/v1"
MV_BASE = "https://app.mavera.io/api/v1"
MV_H = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
lv_auth = base64.b64encode(f"{LV_KEY}:".encode()).decode()
LV_H = {"Authorization": f"Basic {lv_auth}"}
def lv_get(path, params=None):
r = requests.get(f"{LV_BASE}{path}", headers=LV_H, params=params or {})
if r.status_code == 429:
time.sleep(1)
return lv_get(path, params)
r.raise_for_status()
return r.json()
# 1. Pull active postings
postings = lv_get("/postings", {"state": "published"}).get("data", [])
# 2. Pull opportunities with source data
opps = []
offset = None
while len(opps) < 800:
params = {"limit": 100}
if offset:
params["offset"] = offset
resp = lv_get("/opportunities", params)
opps.extend(resp.get("data", []))
offset = resp.get("next")
if not offset:
break
time.sleep(0.15)
# 3. Aggregate source performance
source_stats = defaultdict(lambda: {"applied": 0, "screened": 0, "onsited": 0, "offered": 0, "hired": 0})
stage_resp = lv_get("/stages")
stage_names = {s["id"]: s["text"].lower() for s in stage_resp.get("data", [])}
for opp in opps:
sources = opp.get("sources", [])
source = sources[0] if sources else "Direct"
stage_name = stage_names.get(opp.get("stage", ""), "unknown")
source_stats[source]["applied"] += 1
if any(kw in stage_name for kw in ["screen", "phone"]):
source_stats[source]["screened"] += 1
if "onsite" in stage_name or "final" in stage_name:
source_stats[source]["onsited"] += 1
if "offer" in stage_name:
source_stats[source]["offered"] += 1
if opp.get("isArchived") and "hired" in stage_name:
source_stats[source]["hired"] += 1
# 4. Mave analysis
source_block = "\n".join(
f"- {src}: Applied={s['applied']} Screen={s['screened']} Onsite={s['onsited']} Offer={s['offered']} Hired={s['hired']}"
for src, s in sorted(source_stats.items(), key=lambda x: -x[1]["applied"])[:10]
)
analysis = requests.post(f"{MV_BASE}/mave/chat", headers=MV_H, json={
"message": f"""Analyze recruitment source effectiveness from this Lever data:
{source_block}
For each source:
1. Conversion rate through funnel stages
2. Quality indicator (hired/applied ratio)
3. Where candidates drop off
4. Recommended recruitment marketing strategy for this channel
5. What messaging resonates on this channel vs others"""
}).json()
print("=== Source Effectiveness Analysis ===")
print(analysis.get("content", "")[:1500])
# 5. Generate source-optimized content
for source in list(source_stats.keys())[:3]:
stats = source_stats[source]
gen = requests.post(f"{MV_BASE}/generations", headers=MV_H, json={
"prompt": (
f"Write a recruitment marketing post optimized for {source}. "
f"This source has {stats['applied']} applicants with {stats['hired']} hires "
f"({stats['hired']/max(stats['applied'],1)*100:.0f}% conversion). "
f"Match the tone and format conventions of {source}. 150-200 words."
),
}).json()
print(f"\n--- {source} Content ---")
print(gen.get("output", gen.get("content", gen.get("text", "")))[:500]
time.sleep(0.5)