path: root/references/generate-json.py

#!/usr/bin/env python

# Generate JSON to import Trac tickets into GitHub issues using the new import API
# described at https://gist.github.com/jonmagic/5282384165e0f86ef105

import os
import time
import json
import yaml
import sqlite3
import hashlib
import argparse
import subprocess

ticket_query = '''
SELECT
    id,
    type,
    owner,
    reporter,
    milestone,
    status,
    resolution,
    summary,
    description,
    component,
    priority,
    time       / 1000000 AS createdtime,
    changetime / 1000000 AS modifiedtime
FROM
    ticket
ORDER BY
    id
'''

comment_query = '''
SELECT
    time / 1000000 AS createdtime,
    author,
    newvalue
FROM
    ticket_change
WHERE
    ticket = ?
AND
    field = 'comment'
AND
    newvalue <> ''
ORDER BY
    time
'''

attachment_query = '''
SELECT
    id,
    filename,
    size,
    author,
    description,
    ipnr,
    time / 1000000 AS createdtime
FROM
    attachment
WHERE
    id = ?
AND
    type = 'ticket'
ORDER BY
    time, filename
'''

def isotime(t):
  return None if t == 0 else time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t))

def hashname(whatever):
  return hashlib.sha1(unicode(whatever)).hexdigest()

def ticket_text(ticket):
  d = dict(ticket, createdtime = isotime(ticket["createdtime"]), modifiedtime = isotime(ticket["modifiedtime"]))
  return u"{description}\n\n" \
         u"_Trac ticket #{id} component {component} priority {priority}, owner {owner}," \
         u" created by {reporter} on {createdtime}, last modified {modifiedtime}_\n".format(**d)

def comment_text(comment):
  d = dict(comment, createdtime = isotime(comment["createdtime"]))
  return u"{newvalue}\n\n_Trac comment by {author} on {createdtime}_\n".format(**d)

def attachment_text(attachment):
  h1  = hashname(attachment["id"])
  h2  = hashname(attachment["filename"])
  fn2 = os.path.splitext(attachment["filename"])[1]
  fn  = os.path.join(gist_url, h1[:3], h1, h2 + fn2)
  url = "{}/raw/{}/ticket.{}.{}{}".format(gist_url.rstrip("/"), gist_commit, h1, h2, fn2)
  d = dict(attachment, createdtime = isotime(comment["createdtime"]), url = url)
  return u"[{filename}]({url}) {description}\n_Trac attachment by {author} on {createdtime}_\n".format(**d)

def comment_merge(comments, attachments):
  result = []
  while comments and attachments:
    result.append(comments.pop(0) if comments[0]["created_at"] <= attachments[0]["created_at"] else attachments.pop(0))
  return result + comments + attachments

parser = argparse.ArgumentParser(formatter_class = argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("-c", "--config", type = argparse.FileType(),
                    default = "generate-json.yaml", 
                    help = "YAML config mappings")
args = parser.parse_args()

cfg = yaml.safe_load(args.config)
assignee_map   = cfg["assignees"]
type_map       = cfg["type_labels"]
resolution_map = cfg["resolution_labels"]

gist_url = cfg.get("attachment_gist_url")
if gist_url is not None:
  gist_commit = subprocess.check_output(("git", "ls-remote", gist_url, "HEAD")).split()[0]

db = sqlite3.connect(cfg["database"])
db.row_factory    = sqlite3.Row
ticket_cursor     = db.cursor()
comment_cursor    = db.cursor()
attachment_cursor = db.cursor()

if not os.path.isdir(cfg["ticket_directory"]):
  os.makedirs(cfg["ticket_directory"])

for ticket in ticket_cursor.execute(ticket_query):
  comments = comment_merge([dict(created_at = isotime(comment["createdtime"]), body = comment_text(comment))
                            for comment in comment_cursor.execute(comment_query, (ticket["id"],))],
                           [] if gist_url is None else
                           [dict(created_at = isotime(attachment["createdtime"]), body = attachment_text(attachment))
                            for attachment in attachment_cursor.execute(attachment_query, (ticket["id"],))])
  issue = dict(
    title = ticket["summary"],
    body  = ticket_text(ticket),
    created_at = isotime(ticket["createdtime"]),
    updated_at = isotime(ticket["modifiedtime"]))
  if ticket["status"] == "closed":
    issue["closed"] = True
    issue["closed_at"] = isotime(ticket["modifiedtime"])
    comments.append(dict(created_at = isotime(ticket["modifiedtime"]),
                         body = "_Closed with resolution {resolution}_\n".format(**ticket)))
  if ticket["owner"] in assignee_map:
    issue["assignee"] = assignee_map[ticket["owner"]]
  labels = [type_map.get(ticket["type"]), resolution_map.get(ticket["resolution"])]
  while None in labels:
    del labels[labels.index(None)]
  if labels:
    issue["labels"] = labels
  issue = dict(issue = issue)
  if comments:
    issue["comments"] = comments
  with open(os.path.join(cfg["ticket_directory"], "ticket_{:03d}.json".format(ticket["id"])), "wb") as f:
    json.dump(issue, f, indent = 4, sort_keys = True, separators=(",", ": "))