112 lines
4.6 KiB
Python
112 lines
4.6 KiB
Python
import os
|
|
import sys
|
|
import csv
|
|
import json
|
|
import urllib.request
|
|
import urllib.parse
|
|
|
|
# Config
|
|
BOT_NAME = "pragmatismogb"
|
|
API_URL = "http://localhost:8080"
|
|
LOGIN_USER = os.getenv("CRM_USER", "admin")
|
|
LOGIN_PASS = os.getenv("CRM_PASS", "admin")
|
|
|
|
def login():
|
|
print(f"Logging in to {API_URL} as {LOGIN_USER}...")
|
|
req = urllib.request.Request(f"{API_URL}/api/auth/login",
|
|
data=json.dumps({"email": LOGIN_USER, "password": LOGIN_PASS}).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req) as response:
|
|
if response.status == 200:
|
|
body = json.loads(response.read().decode('utf-8'))
|
|
token = body.get("token") or body.get("access_token")
|
|
# Alternatively check for Set-Cookie header
|
|
cookie = response.getheader("Set-Cookie")
|
|
if token:
|
|
print("Login successful (Got Token)")
|
|
return {"type": "bearer", "value": token}
|
|
elif cookie:
|
|
print("Login successful (Got Cookie)")
|
|
return {"type": "cookie", "value": cookie.split(';')[0]}
|
|
else:
|
|
return {"type": "bearer", "value": ""} # Unknown state
|
|
else:
|
|
print(f"Login failed: {response.status}")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Login failed: {e}")
|
|
return None
|
|
|
|
def import_csv(auth, leads_file, opps_file):
|
|
headers = {"bot-name": BOT_NAME, "Content-Type": "application/json"}
|
|
if auth:
|
|
if auth["type"] == "bearer":
|
|
headers["Authorization"] = f"Bearer {auth['value']}"
|
|
elif auth["type"] == "cookie":
|
|
headers["Cookie"] = auth["value"]
|
|
|
|
# Import leads
|
|
try:
|
|
with open(leads_file, "r") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
val_str = row.get("value", "").strip()
|
|
val = float(val_str) if val_str else 0.0
|
|
payload = {
|
|
"title": row.get("title", ""),
|
|
"description": row.get("description", ""),
|
|
"value": val,
|
|
"currency": row.get("currency", "USD"),
|
|
"source": row.get("source", "")
|
|
}
|
|
|
|
req = urllib.request.Request(f"{API_URL}/api/crm/leads",
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers=headers,
|
|
method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req) as response:
|
|
print(f"[+] Created lead: {payload['title']} -> {response.status}")
|
|
except Exception as e:
|
|
print(f"[-] Failed lead: {payload['title']} -> {e}")
|
|
except FileNotFoundError:
|
|
print(f"Could not find {leads_file}")
|
|
|
|
# Import opps
|
|
try:
|
|
with open(opps_file, "r") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
val_str = row.get("value", "").strip()
|
|
val = float(val_str) if val_str else 0.0
|
|
pr_str = row.get("probability", "").strip()
|
|
pr = int(pr_str) if pr_str else 25
|
|
payload = {
|
|
"name": row.get("name", ""),
|
|
"description": row.get("description", ""),
|
|
"value": val,
|
|
"currency": row.get("currency", "USD"),
|
|
"stage": row.get("stage", "qualification"),
|
|
"probability": pr
|
|
}
|
|
|
|
req = urllib.request.Request(f"{API_URL}/api/crm/opportunities",
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers=headers,
|
|
method="POST")
|
|
try:
|
|
with urllib.request.urlopen(req) as response:
|
|
print(f"[+] Created opportunity: {payload['name']} -> {response.status}")
|
|
except Exception as e:
|
|
print(f"[-] Failed opportunity: {payload['name']} -> {e}")
|
|
except FileNotFoundError:
|
|
print(f"Could not find {opps_file}")
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 3:
|
|
print("Usage: python3 import_csv.py <leads_csv> <opps_csv>")
|
|
sys.exit(1)
|
|
|
|
auth_data = login()
|
|
import_csv(auth_data, sys.argv[1], sys.argv[2])
|