261 lines
8.7 KiB
Python
Executable File
261 lines
8.7 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import pickle
|
|
import sys
|
|
import html2text
|
|
|
|
from zammad_py import ZammadAPI
|
|
from zammad_py.api import Resource, TagList, TicketArticle
|
|
|
|
TEMPLATE = """{
|
|
"zammad_url": "",
|
|
"zammad_user": "",
|
|
"zammad_password": "",
|
|
"rt_url": "",
|
|
"rt_user": "",
|
|
"rt_pass": "",
|
|
"rt_start": 1,
|
|
"rt_end": 1000,
|
|
}
|
|
"""
|
|
|
|
COMMENT_TEMPLATE = """
|
|
Ticket imported from Request Tracker
|
|
|
|
URL: https://support.ungleich.ch/Ticket/Display.html?id={numerical_id}
|
|
Created: {Created}
|
|
Resolved: {Resolved}
|
|
"""
|
|
|
|
STATUSMAP = {"new": 1, "open": 2, "resolved": 4, "rejected": 4, "deleted": 4}
|
|
USERMAP = {}
|
|
|
|
ZAMMAD_SESSIONS = {}
|
|
|
|
### helpers ###
|
|
|
|
def get_zammad_session(impersonated_user=None):
|
|
if impersonated_user in ZAMMAD_SESSIONS:
|
|
return ZAMMAD_SESSIONS[impersonated_user]
|
|
else:
|
|
kwargs = {}
|
|
if impersonated_user:
|
|
kwargs["on_behalf_of"] = impersonated_user
|
|
session = ZammadAPI(
|
|
url=config["zammad_host"],
|
|
username=config["zammad_user"],
|
|
password=config["zammad_password"],
|
|
**kwargs,
|
|
)
|
|
ZAMMAD_SESSIONS[impersonated_user or config["zammad_user"]] = session
|
|
|
|
return session
|
|
|
|
def maybe_create_zammad_user(userdata, zammad_session, attr="login", default=None):
|
|
# Map disabled users (in RT) to mock user in Zammad.
|
|
if type(userdata) is str and userdata.lower() in USERMAP:
|
|
email = userdata
|
|
elif type(userdata) is str or "EmailAddress" not in userdata:
|
|
userdata = {
|
|
'EmailAddress': 'technik@ungleich.ch',
|
|
'RealName': 'Disabled RT'
|
|
}
|
|
email = userdata["EmailAddress"]
|
|
elif "EmailAddress" in userdata:
|
|
email = userdata["EmailAddress"]
|
|
else:
|
|
raise ValueError("Invalid userdata")
|
|
|
|
lowercase_email = email.lower()
|
|
if lowercase_email not in USERMAP:
|
|
kwargs = {"email": email}
|
|
kwargs.update(config["userdata"])
|
|
if "RealName" in userdata:
|
|
realname = userdata["RealName"]
|
|
if ", " in realname:
|
|
last, first = realname.split(", ", 1)
|
|
elif " " in realname:
|
|
first, last = realname.split(None, 1)
|
|
else:
|
|
last = realname
|
|
first = ""
|
|
kwargs["lastname"] = last
|
|
kwargs["firstname"] = first
|
|
try:
|
|
user = zammad_session.user.create(kwargs)
|
|
USERMAP[lowercase_email] = user
|
|
except:
|
|
# The use probably exist already...
|
|
result = list(zammad.user.search(lowercase_email))
|
|
if len(result) == 1:
|
|
user = list.pop(result)
|
|
USERMAP[lowercase_email] = user
|
|
else:
|
|
print(f"Could not create/fetch user {lowercase_email}")
|
|
|
|
if default is None:
|
|
return USERMAP[lowercase_email][attr]
|
|
|
|
return USERMAP[lowercase_email].get(attr, default)
|
|
|
|
# def ensure_user_is_zammad_agent(user, zammad_session):
|
|
# print(f"Promoting user {user} to Agent")
|
|
# id = maybe_create_zammad_user(user, zammad_session, "id")
|
|
# roles = maybe_create_zammad_user(user, zammad_session, "roles")
|
|
# if "Agent" not in roles:
|
|
# zammad_session.user.update(
|
|
# id, {"roles": roles.append("Agent")}
|
|
# )
|
|
|
|
### main logic ###
|
|
|
|
if not os.path.exists("rt2zammad.json"):
|
|
print("Missing rt2zammad.json!")
|
|
print("Create one based on following template:")
|
|
print(TEMPLATE)
|
|
sys.exit(1)
|
|
|
|
with open("rt2zammad.json") as handle:
|
|
config = json.load(handle)
|
|
|
|
h2t = html2text.HTML2Text()
|
|
zammad = get_zammad_session()
|
|
|
|
os.makedirs("users", exist_ok=True)
|
|
os.makedirs("tickets", exist_ok=True)
|
|
os.makedirs("attachments", exist_ok=True)
|
|
|
|
ticket_ids = os.listdir("tickets/")
|
|
print(f"Found {len(ticket_ids)} tickets on filesystem.")
|
|
|
|
for id in ticket_ids:
|
|
with open(f"tickets/{id}", "rb") as handle:
|
|
rt_ticket = pickle.load(handle)
|
|
|
|
label = "RT-{}".format(rt_ticket["ticket"]["original_id"])
|
|
creator = rt_ticket["ticket"]["Creator"]
|
|
with open(f"users/{creator}", "rb") as handle:
|
|
rt_creator = pickle.load(handle)
|
|
|
|
# The RT user object doesn't always contain everything (e.g. disabled users).
|
|
if "EmailAddress" in rt_creator:
|
|
creator = rt_creator["EmailAddress"]
|
|
print(f"Importing {label} ({creator})")
|
|
|
|
zammad_creator = maybe_create_zammad_user(rt_creator, zammad)
|
|
|
|
# Make sure that we use what we created in Zammad, independently of what we
|
|
# had in RT.
|
|
creator = zammad_creator
|
|
|
|
zammad_ticket_template = {
|
|
"title": rt_ticket["ticket"]["Subject"],
|
|
"group": "Users",
|
|
"customer": creator,
|
|
"note": "RT-import:{}".format(rt_ticket["ticket"]["original_id"]),
|
|
"article": {
|
|
"subject": rt_ticket["ticket"]["Subject"],
|
|
},
|
|
}
|
|
|
|
# Ticket creation.
|
|
merged = False
|
|
if rt_ticket["ticket"]["original_id"] != rt_ticket["ticket"]["numerical_id"]:
|
|
merged = True
|
|
zammad_ticket_template["state_id"] = STATUSMAP["resolved"]
|
|
zammad_ticket_template["article"]["body"] = "RT ticket merged into {}".format(
|
|
rt_ticket["ticket"]["numerical_id"]
|
|
)
|
|
zammad_ticket = get_zammad_session(creator).ticket.create(zammad_ticket_template)
|
|
else:
|
|
zammad_ticket_template["state_id"] = STATUSMAP[rt_ticket["ticket"]["Status"]]
|
|
body = rt_ticket["history"][0]["Content"] or 'RT Import: empty comment.'
|
|
zammad_ticket_template["article"]["body"] = body
|
|
zammad_ticket = get_zammad_session(creator).ticket.create(zammad_ticket_template)
|
|
|
|
print(f"Created Zammad ticket {zammad_ticket['id']} for {label}")
|
|
|
|
if rt_ticket["ticket"]["Owner"] and rt_ticket["ticket"]["Owner"] != "Nobody":
|
|
zammad_owner_id = maybe_create_zammad_user(rt_ticket["ticket"]["Owner"], zammad, "id")
|
|
zammad.ticket.update(
|
|
zammad_ticket["id"], {"owner_id": zammad_owner_id}
|
|
)
|
|
|
|
# Ignore comments for merged tickets.
|
|
if merged:
|
|
continue
|
|
|
|
# Internal note regarding the RT-Zammad import.
|
|
TicketArticle(zammad).create(
|
|
{
|
|
"ticket_id": zammad_ticket["id"],
|
|
"body": COMMENT_TEMPLATE.format(**rt_ticket["ticket"]),
|
|
"internal": True,
|
|
}
|
|
)
|
|
|
|
# Comments/notes within the ticket.
|
|
for entry in rt_ticket["history"]:
|
|
if entry["Type"] not in ("Correspond", "Comment"):
|
|
continue
|
|
|
|
# Attachments.
|
|
files = []
|
|
may_contain_entry_content = None
|
|
for a, title in entry["Attachments"]:
|
|
with open(f"attachments/{id}/{a}", "rb") as handle:
|
|
data = pickle.load(handle)
|
|
|
|
if data["Filename"] == "signature.asc":
|
|
continue
|
|
|
|
file_template = {
|
|
"filename": data["Filename"],
|
|
"data": base64.b64encode(data["Content"]).decode("utf-8"),
|
|
"mime-type": data["ContentType"],
|
|
}
|
|
|
|
if data["Filename"] == '':
|
|
if may_contain_entry_content is None:
|
|
may_contain_entry_content = file_template
|
|
else:
|
|
files.append(file_template)
|
|
|
|
# Comment/note.
|
|
entry_creator = entry["Creator"]
|
|
with open(f"users/{entry_creator}", "rb") as handle:
|
|
rt_entry_creator = pickle.load(handle)
|
|
|
|
zammad_entry_creator = maybe_create_zammad_user(rt_entry_creator, zammad)
|
|
entry_content = entry["Content"]
|
|
if entry_content == '' and may_contain_entry_content:
|
|
file = may_contain_entry_content
|
|
entry_content = base64.b64decode(file["data"]).decode("utf-8")
|
|
if file["mime-type"]:
|
|
entry_content = h2t.handle(entry_content)
|
|
|
|
zammad_entry_template = {
|
|
"ticket_id": zammad_ticket["id"],
|
|
"body": entry_content,
|
|
"internal": entry["Type"] == "Comment",
|
|
"attachments": files,
|
|
}
|
|
entry_creator_id = maybe_create_zammad_user(zammad_entry_creator, zammad, "id")
|
|
|
|
# We temporarly change the ticket's creator to create a new entry
|
|
# without giving its author 'Agent' privileges.
|
|
restore_creator_to = None
|
|
if entry_creator_id != zammad_ticket["customer_id"]:
|
|
zammad_ticket = zammad.ticket.update(zammad_ticket["id"], {"customer_id": entry_creator_id})
|
|
restore_creator_to = zammad_ticket["customer_id"]
|
|
|
|
zammad_ticket_id = zammad_ticket["id"]
|
|
print(f"-> Adding a new entry/comment to ticket {zammad_ticket_id} ({zammad_entry_creator})...")
|
|
TicketArticle(get_zammad_session(zammad_entry_creator)).create(zammad_entry_template)
|
|
|
|
if restore_creator_to != None:
|
|
zammad.ticket.update(zammad_ticket["id"], {"customer_id": restore_creator_to})
|