Downloads the latest TSV files from NPS and exports them in PKGi PS3 formatted TXT files

nps2pkgi.py

import csv
import os
import re
import tempfile
import urllib.parse
import urllib.request

# Base script variables
BASE_NPS_URL = "http://nopaystation.com/tsv/"
BASE_NPS_VERSIONS = { 1: "", 2: "pending/" }
PKGI_CSV_FIELDS = ["contentid", "type", "name", "description", "rap", "url", "size", "checksum"]
REJECTED_CSV_NAME = "rejected_packages.csv"
NPS_FILE_NAMES = {
    "pkgi_games.txt": "PS3_GAMES.tsv",
    "pkgi_dlcs.txt": "PS3_DLCS.tsv",
    "pkgi_themes.txt": "PS3_THEMES.tsv",
    "pkgi_avatars.txt": "PS3_AVATARS.tsv",
    "pkgi_demos.txt": "PS3_DEMOS.tsv",
}
CONTENT_TYPE_MAPPINGS = {
    "pkgi_games.txt": 1,
    "pkgi_dlcs.txt": 2,
    "pkgi_themes.txt": 3,
    "pkgi_avatars.txt": 4,
    "pkgi_demos.txt": 5,
}
PACKAGE_NAME_HEADERS = {
    "pkgi_games.txt": "",
    "pkgi_dlcs.txt": "[DLC]",
    "pkgi_themes.txt": "[Theme]",
    "pkgi_avatars.txt": "[PSN Avatar]",
    "pkgi_demos.txt": "[Demo]",
}

# Game sorting variables
PACKAGE_NAME_BAD_CHARACTERS = [",", "\"", "™", "®", ]
PACKAGE_NAME_BAD_STRINGS = [
    "(PSX)",
    "(PS2)",
    "(PS2 Classic)",
    "(MINIS)",
    "(PS3/PSP)",
    "(PS3/PSP/PS Vita)"
]
PS1_TITLE_ID_HEADERS = [
    "NPEF",
    "NPJJ",
    "NPUF",
    "NPUI",
    "NPUJ"
]
PS2_TITLE_ID_HEADERS = [
    "NPEC",
    "NPED",
    "NPJC",
    "NPJD",
    "NPUC",
    "NPUD"
]
PS3_TITLE_ID_HEADERS = [
    "BLET",
    "NPHA",
    "NPHB",
    "NPEA",
    "NPEB",
    "NPIA",
    "NPJA",
    "NPJB",
    "NPUA",
    "NPUB",
    "NPUO",
    "NPUP",
]
MINIS_TITLE_ID_HEADERS = [
    "NPUZ"
]

# DLC sorting variables
GUITAR_HERO_IDS = [
    "BLUS30074",  # Guitar Hero 3 (US)
    "BLUS30164",  # Guitar Hero World Tour (US)
    "BLUS30292",  # Guitar Hero - Warriors of Rock (US)
    "BLES00134",  # Guitar Hero 3 (EU)
    "BLES00299",  # Guitar Hero World Tour (EU)
    "BLES00576",  # Guitar Hero 5 (EU)
    "BLES00134",  # Guitar Hero - Warriors of Rock (EU)
    "BLES00801",  # Guitar Hero - Warriors of Rock (EU) Alt
    "BLES00299",  # Guitar Hero - Warriors of Rock (EU) Alt
    "BLES00801",  # Guitar Hero - Warriors of Rock (EU) Alt
]
ROCK_BAND_IDS = [
    "BLUS30050",  # Rock Band 1 (US)
    "BLUS30147",  # Rock Band 2 (US)
    "BLUS30463",  # Rock Band 3 (US)
    "BLUS30282",  # The Beatles Rock Band (US)
    "BLES00228",  # Rock Band 1 (EU)
    "BLES00385",  # Rock Band 2 (EU)
    "BLES00986",  # Rock Band 3 (EU)
    "BLES01611",  # Rock Band 3 (EU) Alt
    "BLES00986",  # Rock Band 3 (EU) Alt
    "BLES00532",  # The Beatles Rock Band (EU)
]

# TODO: Remove attachment
REMOVE_USELESS_ATTACHMENT = [
    " - (DEMO)",
	" - DEMO",
	" - TRIAL",
	" TRIAL VERSION"
]

# Script statistic variables
INPUT_COUNT = 0
VALID_COUNT = 0
REJECTED_COUNT = 0


def convert_tsv_to_pkgi(tsv_file_path, pkgi_file):
    """Converts the given NPS TSV file to PKGi PS3 format"""
    with open(tsv_file_path, newline="", encoding="utf8") as tsv_file:
        tsv_data_dict = csv.DictReader(tsv_file, delimiter="\t", quotechar="\"")
        for line in tsv_data_dict:

            # Increment counter for how many entries have been processed
            global INPUT_COUNT
            INPUT_COUNT += 1

            if verify_nps_entry(line):
                # Compile CSV data into dictionary
                line_data = {
                    "contentid": line.get("Content ID"),
                    "type": CONTENT_TYPE_MAPPINGS[pkgi_file],
                    "name": generate_package_name(line.get("Name"), PACKAGE_NAME_HEADERS[pkgi_file], line.get("Title ID")),
                    "description": "",
                    "rap": line.get("RAP"),
                    "url": line.get("PKG direct link"),
                    "size": line.get("File Size"),
                    "checksum": line.get("SHA256")
                }

                # Print status message
                print("[{}] Adding '{}'...".format(pkgi_file, line_data.get("name")))

                # Write CSV data to file
                with open(pkgi_file, "a", newline="", encoding="utf8") as csv_output:
                    write_output = csv.DictWriter(csv_output, fieldnames=PKGI_CSV_FIELDS)
                    write_output.writerow(line_data)


def download_nps_tsv(temp_folder_path, file_name):
    """Downloads the TSV file to the temp folder and returns the resulting file path"""
    tsv_file_path = os.path.join(temp_folder_path, file_name)

    with open(tsv_file_path, "wb") as file:
        for nps_type in BASE_NPS_VERSIONS:
            url = BASE_NPS_URL + BASE_NPS_VERSIONS[nps_type]

            download_link = (urllib.parse.urljoin(url, file_name))

            req = urllib.request.Request(download_link, data=None, headers={"User-Agent": "<INSERT USER AGENT HERE>"})

            file.write(urllib.request.urlopen(req).read())

    return tsv_file_path


def generate_package_name(package_name, header, title_id):
    """Generates a nicely formatted package name and returns it"""

    # "pkgi_games.txt" does not pass in a predetermined header as we need to apply one based on format
    if not header:

        # Remove bad strings from package name
        for package_string in PACKAGE_NAME_BAD_STRINGS:
            if " {}".format(package_string) in package_name:
                package_name = "{}".format(package_name.replace(" {}".format(package_string), ""))
            if package_string in package_name:
                package_name = "{}".format(package_name.replace(package_string, ""))

        # Determine package format by title ID and add header
        package_title_id_header = title_id[0:4]
        if package_title_id_header in PS1_TITLE_ID_HEADERS:
            package_name = "[PS1] {}".format(package_name)
        elif package_title_id_header in PS2_TITLE_ID_HEADERS:
            package_name = "[PS2] {}".format(package_name)
        elif package_title_id_header in PS3_TITLE_ID_HEADERS:
            package_name = "[PS3] {}".format(package_name)
        elif package_title_id_header in MINIS_TITLE_ID_HEADERS:
            package_name = "[MINIS] {}".format(package_name)
        else:
            package_name = "[UNKNOWN] {}".format(package_name)

    # Add custom headers to DLC entries for games with massive amount of DLC
    if header == PACKAGE_NAME_HEADERS["pkgi_dlcs.txt"]:
        if title_id in GUITAR_HERO_IDS:
            package_name = "[Guitar Hero] {}".format(package_name)
        if title_id in ROCK_BAND_IDS:
            package_name = "[Rock Band] {}".format(package_name)

    # Append header to start of package name
    if header:
        package_name = "{} {}".format(header, package_name)

    # Remove bad characters from package name
    for char in PACKAGE_NAME_BAD_CHARACTERS:
        if char in package_name:
            package_name = package_name.replace(char, "")

    return package_name


def verify_nps_entry(nps_entry):
    """Verifies the NPS entry and returns true or false, if false also adds entry info to the rejection CSV"""

    def verification_failed(reason_rejected):
        """Sub function that adds the data for the entry that failed verification to a separate CSV"""

        # Increment counter for how many entries have been rejected
        global REJECTED_COUNT
        REJECTED_COUNT += 1

        # Compile CSV data into dictionary
        rejected_csv_fields = ["Reason Rejected", "Title ID", "Region", "Name", "PKG direct link", "RAP", "Content ID",
                               "Last Modification Date", "Download .RAP file", "File Size", "SHA256"]
        line_data = {"Reason Rejected": reason_rejected}
        for value in rejected_csv_fields:
            if value != "Reason Rejected":
                line_data.update({value: nps_entry.get(value)})

        # Write CSV data to file
        is_new_file = not os.path.isfile(REJECTED_CSV_NAME)
        with open(REJECTED_CSV_NAME, "a", newline="", encoding="utf8") as csv_output:
            write_output = csv.DictWriter(csv_output, fieldnames=rejected_csv_fields)
            if is_new_file:
                write_output.writeheader()
            write_output.writerow(line_data)

    # Skip entries that do not have a valid name
    if nps_entry.get("Name") == "UNKNOWN TITLE" or nps_entry.get("Name") == "":
        verification_failed("Invalid Name")
        return False

    # Skip entries that do not have a content ID
    content_id_regex = re.compile(r"[A-Z]{2}[0-9]{4}-[A-Z]{4}[0-9]{5}_[0-9]{2}-\S{16}")
    if not content_id_regex.fullmatch(nps_entry.get("Content ID")):
        verification_failed("Invalid Content ID")
        return False

    # Skip entries that do not have a valid title ID
    title_id_regex = re.compile(r"[A-Z]{4}[0-9]{5}")
    if not title_id_regex.fullmatch(nps_entry.get("Title ID")):
        verification_failed("Invalid Title ID")
        return False

    # Skip entries that do not have a download link
    if nps_entry.get("PKG direct link") == "MISSING" or nps_entry.get("PKG direct link") == "":
        verification_failed("Invalid PKG Link")
        return False

    # Skip entries that do not have a valid RAP
	# TODO: Add "* [MISSING RAP]
    #if nps_entry.get("RAP") == "MISSING" or nps_entry.get("RAP") == "":
    #    verification_failed("Invalid RAP")
    #    return False

    # If the entry is not rejected prior to this, it must be valid so return true
    global VALID_COUNT
    VALID_COUNT += 1
    return True


def main():
    """Downloads the latest TSV files from NPS and exports them in PKGi PS3 formatted TXT files"""

    # Delete any existing PKGi files
    for filename in os.listdir():
        if filename in NPS_FILE_NAMES or filename == REJECTED_CSV_NAME:
            os.remove(filename)

    # Process NPS files
    with tempfile.TemporaryDirectory() as temp_folder_path:
        for file in NPS_FILE_NAMES:
            tsv_file_path = download_nps_tsv(temp_folder_path, NPS_FILE_NAMES[file])
            convert_tsv_to_pkgi(tsv_file_path, file)

    # Print results message
    print()
    print("Results:")
    print(" NPS entries processed: {}".format(INPUT_COUNT))
    print(" Valid NPS entries added to PKGi files: {}".format(VALID_COUNT))
    print(" Rejected NPS entries: {}".format(REJECTED_COUNT))


if __name__ == "__main__":
    main()