#!/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")} # ) def create_zammad_ticket(id, zammad, h2t): 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: return # 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}) ### 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) os.makedirs("failed", exist_ok=True) ticket_ids = os.listdir("tickets/") print(f"Found {len(ticket_ids)} tickets on filesystem.") for id in ticket_ids: try: create_zammad_ticket(id, zammad, h2t) except KeyboardInterrupt: print("Received keyboard interrupt. Exiting.") sys.exit() except: print(f"Failed to import RT#{id}") dumpfile = f"failed/{id}" with open(dumpfile, "wb") as handle: pickle.dump(id, handle)