recipes/recipe_proto/turboci/update.py

#!/usr/bin/env vpython3
# Copyright 2025 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Automatically updates the .proto files in this directory."""

import json
import os
import re
import tarfile

from pathlib import Path

import requests

BASE_URL = 'https://chromium.googlesource.com/infra/turboci/proto'
LOG_URL = BASE_URL + '/+log/main/{}?format=JSON&n=1'
TAR_URL = BASE_URL + '/+archive/{}/{}.tar.gz'

REPO_PATH = 'turboci/data/chrome/depot_tools'


def print_commit_message(prev_commit: str, target: str):
  print('{0:=^80}'.format(' START[Commit Message] '))
  print('[TurboCI] Import latest protos.')
  print()
  changelog = f'{BASE_URL}/+log/{prev_commit}..{target}'
  print(f'Changelog: {changelog}')
  print()
  clog = json.loads(requests.get(f'{changelog}?format=JSON').text[4:])
  for commit in clog['log']:
    first_line = commit['message'].split('\n')[0]
    print(f'{commit["commit"][:8]} {first_line}')
  print()
  print('{0:=^80}'.format(' END[Commit Message] '))


def main():
  """Automatically updates the .proto files in this directory."""
  this_dir = Path(__file__).parent

  readme_path = this_dir / 'README.md'

  with open(readme_path, 'r') as rmd:
    prev_commit = re.match(r'.*/([a-f0-9]{40})/.*', rmd.read(),
                           re.MULTILINE | re.DOTALL).group(1)

  to_remove = set()
  for root, _, files in os.walk(this_dir):
    for file in files:
      if file.endswith('.proto'):
        to_remove.add(Path(root) / file)

  resp = requests.get(LOG_URL.format(REPO_PATH))
  target = str(json.loads(resp.text[4:])['log'][0]['commit'])

  if prev_commit == target:
    print('Nothing to update.')
    return

  print(f'Updating to {target!r}')

  base_dir = this_dir.parent / REPO_PATH
  resp = requests.get(TAR_URL.format(target, REPO_PATH), stream=True).raw
  with tarfile.open(mode='r', fileobj=resp) as tar:
    for item in tar:
      if not item.name.endswith('.proto'):
        continue
      if item.name.endswith('_test.proto'):
        print(f'Skipping {item.name!r}')
        continue
      if 'internal' in item.name or 'googleapis' in item.name:
        print(f'Skipping {item.name!r}')
        continue
      print(f'Extracting {item.name!r}')
      tar.extract(item, path=base_dir)
      to_remove.discard(base_dir / item.name)

  if to_remove:
    print('Removing stale proto')
    for file in to_remove:
      os.remove(file)

  with open(readme_path, 'w') as rmd:
    print('Generated by update.py. DO NOT EDIT.', file=rmd)
    print('These protos were copied from:', file=rmd)
    print(f'{BASE_URL}/+/{target}/{REPO_PATH}', file=rmd)

  print_commit_message(prev_commit, target)

  print('Done.')


if __name__ == '__main__':
  main()