rt2zammad/fs2zammad

#!/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, retries=3):
    try:
        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 = zammad.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 = zammad.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.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})
    except KeyboardInterrupt:
        print("Received keyboard interrupt. Exiting.")
        sys.exit()
    except:
        print(f"Failed to import RT-#{id} .. ({retries} retries left)")
        if retries > 0:
            create_zammad_ticket(id, zammad, h2t, retries - 1)

### 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, 3)
    except SystemExit:
        sys.exit()
    except:
        print(f"Failed to import RT#{id}")
        dumpfile = f"failed/{id}"
        with open(dumpfile, "wb") as handle:
            pickle.dump(id, handle)