rt2zammad/rt2zammad.py

276 lines
7.7 KiB
Python
Raw Normal View History

#!/usr/bin/env python
"""
Quick and dirty attempt to migrate issues from Request Tracker to Zammad.
"""
import base64
import json
import os
import pickle
import sys
from rt import Rt
from zammad_py import ZammadAPI
from zammad_py.api import Resource, TagList, TicketArticle
class Tag(Resource):
path_attribute = "tags"
def add(self, obj, id, item):
2020-12-30 08:18:57 +00:00
response = self._connection.session.post(
self.url + "/add",
2020-12-30 08:45:36 +00:00
data={
"object": obj,
2020-12-30 10:58:40 +00:00
"o_id": id,
2020-12-30 08:45:36 +00:00
"item": item,
},
)
return self._raise_or_return_json(response)
TEMPLATE = """{
"zammad_host": "",
"zammad_user": "",
"zammad_password": "",
"zammad_secure": true,
"rt_url": "",
"rt_user": "",
2020-08-18 11:22:10 +00:00
"rt_pass": "",
"rt_start": 1,
2020-08-18 11:22:21 +00:00
"rt_end": 1000,
"usermap": {},
"userdata": {}
}
"""
COMMENT_TEMPLATE = """
Ticket imported from Request Tracker
2020-12-30 12:14:33 +00:00
URL: https://oldsupport.weblate.org/Ticket/Display.html?id={numerical_id}
Created: {Created}
Resolved: {Resolved}
"""
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)
ZAMMADS = {}
def get_zammad(user=None):
if user not in ZAMMADS:
kwargs = {}
if user:
kwargs["on_behalf_of"] = user
ZAMMADS[user] = ZammadAPI(
host=config["zammad_host"],
username=config["zammad_user"],
password=config["zammad_password"],
is_secure=config["zammad_secure"],
2020-12-30 12:43:40 +00:00
**kwargs,
)
return ZAMMADS[user]
target = get_zammad()
target.user.me()
source = Rt(config["rt_url"], config["rt_user"], config["rt_pass"])
if not source.login():
print("Failed to login to RT!")
sys.exit(2)
2019-05-25 09:01:10 +00:00
if os.path.exists("rt2zammad.cache"):
2019-05-25 09:04:45 +00:00
# Load RT from cache
with open("rt2zammad.cache", "rb") as handle:
2019-05-25 09:01:10 +00:00
data = pickle.load(handle)
users = data["users"]
queues = data["queues"]
tickets = data["tickets"]
attachments = data["attachments"]
2019-05-25 09:01:10 +00:00
else:
2019-05-25 09:04:45 +00:00
# Load RT from remote
2019-05-25 09:01:10 +00:00
users = {}
attachments = {}
2019-05-25 09:01:10 +00:00
tickets = []
queues = set()
def ensure_user(username):
if username not in users:
users[username] = source.get_user(username)
2020-08-18 11:22:10 +00:00
for i in range(config["rt_start"], config["rt_end"]):
2020-12-30 12:44:03 +00:00
print(f"Loading ticket {i}")
2019-05-25 09:01:10 +00:00
ticket = source.get_ticket(i)
if ticket is None:
break
2020-08-18 11:08:50 +00:00
ticket["original_id"] = str(i)
queues.add(ticket["Queue"])
ensure_user(ticket["Creator"])
ensure_user(ticket["Owner"])
2020-08-18 11:08:50 +00:00
if ticket["original_id"] != ticket["numerical_id"]:
# Merged ticket
history = []
else:
history = source.get_history(i)
for item in history:
for a, title in item["Attachments"]:
attachments[a] = source.get_attachment(i, a)
ensure_user(item["Creator"])
tickets.append({"ticket": ticket, "history": history})
with open("rt2zammad.cache", "wb") as handle:
data = pickle.dump(
{
"users": users,
"queues": queues,
"tickets": tickets,
"attachments": attachments,
},
handle,
)
2019-05-25 09:04:45 +00:00
# Create tags
tag_list = TagList(target)
ticket_article = TicketArticle(target)
tag_obj = Tag(target)
tags = {tag["name"] for tag in tag_list.all()}
2019-05-25 09:04:45 +00:00
for queue in queues:
queue = queue.lower().split()[0]
if queue not in tags:
tag_list.create({"name": queue})
STATUSMAP = {"new": 1, "open": 2, "resolved": 4, "rejected": 4, "deleted": 4}
USERMAP = {}
for user in target.user.all():
USERMAP[user["email"].lower()] = user
def get_user(userdata, attr="login", default=None):
email = userdata["EmailAddress"]
lemail = email.lower()
2020-08-18 11:22:21 +00:00
if lemail in config["usermap"]:
email = lemail = config["usermap"][lemail]
# Search existing users
if lemail not in USERMAP:
for user in target.user.search({"query": email}):
USERMAP[user["email"].lower()] = user
# Create new one
if lemail 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
user = target.user.create(kwargs)
USERMAP[user["email"].lower()] = user
if default is None:
return USERMAP[lemail][attr]
2020-12-30 07:59:58 +00:00
return USERMAP[lemail].get(attr, default)
# Create tickets
for ticket in tickets:
2020-08-18 11:08:50 +00:00
label = "RT-{}".format(ticket["ticket"]["original_id"])
2020-12-30 12:14:41 +00:00
creator = get_user(users[ticket["ticket"]["Creator"]])
2020-12-30 12:42:48 +00:00
2020-12-30 12:14:41 +00:00
print(f"Importing {label} ({creator})")
2020-12-30 12:30:22 +00:00
create_args = {
2020-12-30 12:42:48 +00:00
"title": ticket["ticket"]["Subject"],
2020-12-30 12:30:22 +00:00
"group": "Users",
"customer": creator,
"note": "RT-import:{}".format(ticket["ticket"]["original_id"]),
"article": {
"subject": ticket["ticket"]["Subject"],
},
}
2020-12-30 12:43:06 +00:00
merged = False
2020-08-18 11:08:50 +00:00
if ticket["ticket"]["original_id"] != ticket["ticket"]["numerical_id"]:
# Merged ticket
2020-12-30 12:43:06 +00:00
merged = True
2020-12-30 12:30:22 +00:00
create_args["state_id"] = 4
create_args["article"]["body"] = "RT ticket merged into {}".format(
2020-12-30 12:43:40 +00:00
ticket["ticket"]["numerical_id"]
)
2020-12-30 12:43:06 +00:00
new = get_zammad(creator).ticket.create(create_args)
else:
create_args["state_id"] = STATUSMAP[ticket["ticket"]["Status"]]
create_args["article"]["body"] = ticket["history"][0]["Content"]
new = get_zammad(creator).ticket.create(create_args)
2020-12-30 12:30:22 +00:00
print(f"Created ticket {new['id']}")
2020-12-30 12:43:06 +00:00
if ticket["ticket"]["Owner"] and ticket["ticket"]["Owner"] != "Nobody":
2020-12-30 12:43:40 +00:00
get_zammad().ticket.update(
new["id"], {"owner_id": get_user(users[ticket["ticket"]["Owner"]], "id")}
)
2020-12-30 12:43:06 +00:00
if merged:
# Do not add comments to merged ticket
continue
tag_obj.add("Ticket", new["id"], ticket["ticket"]["Queue"].lower().split()[0])
ticket_article.create(
{
"ticket_id": new["id"],
"body": COMMENT_TEMPLATE.format(**ticket["ticket"]),
"internal": True,
}
)
for item in ticket["history"]:
if item["Type"] not in ("Correspond", "Comment"):
continue
files = []
for a, title in item["Attachments"]:
data = attachments[a]
if data["Filename"] in ("", "signature.asc"):
continue
files.append(
{
"filename": data["Filename"],
"data": base64.b64encode(data["Content"]).decode("utf-8"),
"mime-type": data["ContentType"],
}
)
creator_id = get_user(users[item["Creator"]], "id")
chown = creator_id != new["customer_id"] and "Agent" not in get_user(
users[item["Creator"]], "roles", []
)
if chown:
target.ticket.update(new["id"], {"customer_id": creator_id})
TicketArticle(get_zammad(get_user(users[item["Creator"]]))).create(
{
"ticket_id": new["id"],
"body": item["Content"],
"internal": item["Type"] == "Comment",
"attachments": files,
}
)
if chown:
target.ticket.update(new["id"], {"customer_id": new["customer_id"]})