erynwells.me/scripts/import-nethack-logfile.py

#!/usr/bin/env python3
# Eryn Wells <eryn@erynwells.me>

'''
Converts a Nethack logfile (or record file) to JSON for easier parsing by my
website's templating engine.

See https://nethackwiki.com/wiki/Logfile for information about the format of these
logfiles.
'''

import argparse
import datetime
import json
from locale import normalize
import os.path
import subprocess
import sys

DUNGEONS = {
    0: 'The Dungeons of Doom',
    1: 'Gehennom',
    2: 'The Gnomish Mines',
    3: 'The Quest',
    4: 'Sokoban',
    5: 'Fort Ludios',
    6: "Vlad's Tower",
    7: 'The Elemental Planes',
}

# The "dungeon level" field is normally positive, but negatives indicate one of these
# levels.
SPECIAL_DUNGEON_LEVELS = {
    -5: 'Astral Plane',
    -4: 'Plane of Water',
    -3: 'Plane of Fire',
    -2: 'Plane of Air',
    -1: 'Plane of Earth',
}

RACES = {
    'Elf': 'Elf',
    'Gno': 'Gnome',
    'Hum': 'Human',
}

ROLES = {
    'Arc': 'Archaeologist',
    'Kni': 'Knight',
    'Mon': 'Monk',
    'Pri': 'Priest',
    'Ran': 'Ranger',
    'Rog': 'Rogue',
    'Sam': 'Samurai',
    'Val': 'Valkyrie',
}

GENDERS = {
    'Fem': 'Female',
}

ALIGNMENTS = {
    'Law': 'Lawful',
    'Neu': 'Neutral',
    'Cha': 'Chaotic',
}

def parse_args(argv, *a, **kw):
    parser = argparse.ArgumentParser(*a, **kw)
    parser.add_argument('-o', '--output', help='Path to the output file')
    parser.add_argument('logfile', help='Path to the Nethack log file to convert')
    args = parser.parse_args(argv)
    return args

def main(argv):
    args = parse_args(argv[1:], prog=argv[0])

    if not os.path.isfile(args.logfile):
        print('Given path is not a real file!', file=sys.stderr)
        return -1

    hostname = subprocess.check_output(['hostname', '-s']).decode('ascii').strip()

    records = []
    with open(args.logfile) as logfile:
        for record in logfile:
            fields = record.split(maxsplit=15)

            dungeon_number = int(fields[2])
            dungeon_name = DUNGEONS[dungeon_number]

            dungeon_level = int(fields[3])
            if dungeon_level > 0:
                dungeon_level_descriptive = f"Level {dungeon_level}"
            else:
                dungeon_level_descriptive = SPECIAL_DUNGEON_LEVELS[dungeon_level]

            start_date = datetime.datetime.strptime(fields[9], '%Y%m%d').strftime('%Y-%m-%d')
            end_date = datetime.datetime.strptime(fields[8], '%Y%m%d').strftime('%Y-%m-%d')

            name, cause_of_death = (s.strip() for s in fields[15].split(',', maxsplit=1))
            role = fields[11]
            race = fields[12]
            gender = fields[13]
            alignment = fields[14]

            records.append({
                'score': int(fields[1]),
                'dungeon': {
                    'n': int(fields[2]),
                    'name': dungeon_name,
                    'level': {'n': dungeon_level, 'descriptive': dungeon_level_descriptive},
                },
                'end_date': end_date,
                'start_date': start_date,
                'character': {
                    'name': name,
                    'descriptor': f'{name}-{role}-{race}-{gender}-{alignment}',
                    'max_level': int(fields[4]),
                    'hp': {'n': int(fields[5]), 'max': int(fields[6])},
                    'role': {'short': role, 'descriptive': ROLES[role]},
                    'race': {'short': race, 'descriptive': RACES[race]},
                    'gender': {'short': gender, 'descriptive': GENDERS[gender]},
                    'alignment': {'short': alignment, 'descriptive': ALIGNMENTS[alignment]},
                },
                'death': {'n': int(fields[7]), 'cause': cause_of_death},
                'system': {
                    'hostname': hostname,
                    'user_id': int(fields[10]),
                    'nethack_version': fields[0],
                }
            })

    output_object = {
        'generated': datetime.datetime.now().isoformat(),
        'logfile': records,
    }

    logfile_has_changed = False
    output_path = args.output
    output_file = None

    if output_path and output_path != '-':
        with open(output_path, 'r') as existing_logfile:
            existing_logfile_object = json.load(existing_logfile)
            logfile_has_changed = existing_logfile_object.get('logfile', {}) != records 

        if logfile_has_changed:
            output_file = open(output_path, 'w')
    else:
        output_file = sys.stdout

    if output_file:
        json.dump(output_object, output_file, indent=2)
        output_file.write('\n')
        output_file.close()
    else:
        print('No changes to logfile')

    return 0

if __name__ == '__main__':
    import sys
    result = main(sys.argv)
    sys.exit(0 if not result else result)