jarupdater-metadata-template/release.py

#!/usr/bin/python3

import json, os, time, tempfile, shutil
from hashlib import sha256

META_FILENAME = 'meta-v1.json'
BASE_URL = open('url.txt').read().strip()
BASE_DIR = 'data'
WAITING_DIR = 'pending'
TEMP_DIR = tempfile.TemporaryDirectory()

# file scanner functions

def match_name(filename: str, extension: str=None, exact_name: str=None):
  if exact_name is not None:
    return filename == exact_name
  elif extension is not None:
    return filename.lower().endswith(extension.lower())
  return True

def scan_for_file(ask: bool=False, extension: str=None, exact_name: str=None) -> tuple[str]:
  for file in os.scandir():
    if file.is_dir(): continue
    if not match_name(file.name, extension, exact_name): continue

    if ask:
      if not confirm(f"Found {file.name} in the current directory, do you want to proceed with it?"):
        return (None, None)

    return (file.path, file.name)
  return (None, None)

def wait_for_file(waiting_dir: str, extension: str=None) -> tuple[str]:
  print(f"Please put a {extension} file in {waiting_dir}")
  while True:
    files = [i for i in os.scandir(waiting_dir)]

    if len(files) == 0:
      time.sleep(0.5)
      continue

    file = files[0]
    filepath = file.path
    filename = file.name

    if match_name(filename, extension):
      break
    else:
      os.remove(filepath)
      print(f"Not a {extension} file: {filename}")

  return (filepath, filename)

def just_find_file(name: str) -> tuple[str]:
  spl = name.split('.')
  extension = spl[-1]
  exact_name = name[:-len(extension)-1] if len(spl) > 1 else None

  filepath, filename = scan_for_file(True, extension, exact_name)
  if filepath is None:
    try:
      os.makedirs(WAITING_DIR, exist_ok=True)
      filepath, filename = wait_for_file(WAITING_DIR, extension)
    except KeyboardInterrupt:
      os.rmdir(WAITING_DIR)
      return
    
  if filepath is not None:
    tpath = os.path.join(TEMP_DIR.name, filename)
    shutil.move(filepath, tpath)
    filepath = tpath
    
  return (filepath, filename)

# directory util fnctions

def make_path(channel: str, version: str=None, filename: str=None) -> str:
  args = [channel, version, filename]
  args = [i for i in args if i is not None]
  return os.path.join(BASE_DIR, *args)

def list_channels() -> list[str]:
  return next(os.walk(BASE_DIR))[1]

# verification functions

def channel_exists(channel: str) -> bool:
  """Checks if channel exists"""
  return os.path.isdir(make_path(channel))

def check_version_exists(version: str) -> str | None:
  """
  Check if version exists in any channel.
  If yes, returns that channel, otherwise None
  """
  for channel in list_channels():
    if os.path.isdir(make_path(channel, version)):
      return channel
    
# metadata functions

def load_latest_data(channel: str) -> dict:
  path = make_path(channel, 'latest', META_FILENAME)
  if os.path.isfile(path):
    return json.loads(open(path).read())
  return {'id': 0}

def write_metadata(channel: str, metadata: dict):
  version = metadata['label']
  metadata = json.dumps(metadata)
  for filepath in [make_path(channel, version, META_FILENAME), make_path(channel, 'latest', META_FILENAME)]:
    with open(filepath, 'w') as file:
      file.write(metadata)

def commit_and_push(channel: str, version: str):
  os.system('git add .')
  os.system(f'git commit -m "[releaser] Release {version} on {channel}"')
  os.system('git push')

# other

def confirm(prompt: str) -> bool:
  confirmed = input(prompt + ' (Y/N) ')
  return confirmed.lower() == 'y'

def hash_file(filepath: str) -> str:
  with open(filepath, 'rb') as file:
    return sha256(file.read()).hexdigest()

# main

def main():
  channel = input('Channel? ({}) '.format(', '.join(list_channels())))
  if channel == '' or not channel_exists(channel):
    print(f"Channel {channel} doesn't exist")
    return

  version = input('New version? ')
  exists_in = check_version_exists(version)
  if exists_in is not None:
    print(f"This version already exists in channel {exists_in}. Choose a different version.")
    return

  jar_filepath, jar_filename = just_find_file('jar')

  changelog = confirm('Do you want to include a changelog?')
  if changelog:
    chlog_filepath, chlog_filename = just_find_file('changelog.txt')

  #

  latest_data = load_latest_data(channel)
  hash = hash_file(jar_filepath)

  version_dir = make_path(channel, version)

  os.makedirs(make_path(channel, 'latest'), exist_ok=True)
  os.mkdir(version_dir)
  shutil.move(jar_filepath, os.path.join(version_dir, jar_filename))
  if changelog: shutil.move(chlog_filepath, os.path.join(version_dir, 'changelog.txt'))

  metadata = {
    'label': version,
    'id': latest_data['id'] + 1,
    'timestamp': int(time.time()),
    'file': f'{BASE_URL}/{BASE_DIR}/{channel}/{version}/{jar_filename}',
    'sha256': hash
  }

  write_metadata(channel, metadata)

  try:
    os.rmdir(WAITING_DIR)
  except FileNotFoundError:
    pass

  if confirm("Commit and push?"):
    commit_and_push(channel, version)

  print("Done")

if __name__ == "__main__":
  main()