Compile Valve/Steam protobufs with betterproto

protos.py

#!/usr/bin/env python

# MIT License
#
# Copyright (c) 2023 tuokri
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

# Works with https://github.com/tuokri/python-betterproto/tree/cec89d915d72cc86a86815845ef44c0022a48df1

import asyncio.subprocess
from pathlib import Path
from typing import List

# Path to SteamDatabase/Protobufs repo.
PROTO_REPO_PATH = Path("Protobufs/").resolve()


async def run_protoc(proto_dir: Path, common_dirs: List[Path]):
    print(f"running protoc on: '{proto_dir}'")
    proto_files = [p for p in proto_dir.glob("*.proto")]
    for cd in common_dirs:
        proto_files.extend(cd.rglob("*.proto"))

    proto_files_args = [str(p.resolve()) for p in proto_files]

    common_dirs_args = [
        f"--proto_path={cd}" for cd in common_dirs
    ]

    out_dir = Path(__file__).parent / "output/proto" / proto_dir.name
    out_dir = out_dir.resolve()
    out_dir.mkdir(parents=True, exist_ok=True)
    workdir = out_dir.parent / f"workdir_{proto_dir.name}"
    workdir = workdir.resolve()
    workdir.mkdir(parents=True, exist_ok=True)

    protoc_coros = []
    for pd_arg in proto_files_args:
        protoc_coros.append(
            (await asyncio.subprocess.create_subprocess_exec(
                "protoc",
                *[
                    f"--python_betterproto_out={out_dir}",
                    f"--proto_path={proto_dir}",
                    *common_dirs_args,
                    pd_arg,
                ],
                cwd=workdir,
            )).wait()
        )
    await asyncio.gather(*protoc_coros)


async def main():
    proto_dirs = [
        p.resolve() for p in PROTO_REPO_PATH.iterdir()
        if p.is_dir()
    ]
    common_dirs = [
        PROTO_REPO_PATH / "google/protobuf/",
    ]
    for pd in proto_dirs:
        await run_protoc(pd, common_dirs)


if __name__ == "__main__":
    asyncio.run(main())