Analyze Go package dependencies.

analyze.py

#!/usr/bin/env python3

__doc__ = """Analyze Go package dependencies."""

import argparse
import json
import os
import shlex
import subprocess
import sys
import tempfile

from typing import List, Dict


def download_packages(packages: List[str], tmp_dir: str):
  for pkg in packages:
    cmd = ["git", "clone", f"https://{pkg}.git"]
    print(f"+ {shlex.join(cmd)}")
    subprocess.run(cmd, cwd=tmp_dir, check=True)


def parse_go_mod(target_dir: str) -> Dict[str, List[str]]:
  cmd = ["go", "list", "-m", "-json", "all"]
  print(f"+ {shlex.join(cmd)}")
  go_list_raw = subprocess.run(
      cmd, cwd=target_dir, check=True,
      stdout=subprocess.PIPE).stdout.decode("utf-8")

  # 개별 JSON 객체를 목록으로 분할
  go_list_raw_items = go_list_raw.strip().split("}\n{")
  go_list = []
  for item_raw in go_list_raw_items:
    # 분할된 각 항목에 중괄호를 복원하고 JSON 로드
    if not item_raw.startswith("{"):
      item_raw = "{" + item_raw
    if not item_raw.endswith("}"):
      item_raw = item_raw + "}"
    go_list.append(json.loads(item_raw))

  main_package = None
  dependencies = {}
  for pkg in go_list:
    if pkg.get("Main"):
      main_package = pkg["Path"]
      break

  if not main_package:
    raise Exception("Main package not found.")

  for pkg in go_list:
    # Main, Indirect 패키지는 제외
    if pkg.get("Main") or pkg.get("Indirect"):
      continue
    dependencies[main_package] = dependencies.get(main_package,
                                                  []) + [pkg["Path"]]

  return dependencies


def remove_not_interesting_dependencies(
    dependencies: Dict[str,
                       List[str]], packages: List[str]) -> Dict[str, List[str]]:
  removed_dependencies = {}
  for key, values in dependencies.items():
    if key not in packages:
      continue
    new_values = []
    for value in values:
      if value not in packages:
        continue
      new_values.append(value)
    removed_dependencies[key] = new_values
  return removed_dependencies


def generate_dependency_graph_dot_file(dependencies: Dict[str, List[str]],
                                       output_file: str):
  with open(output_file, 'w') as f:
    f.write("digraph G {\n")
    for key, values in dependencies.items():
      for value in values:
        f.write(f'    "{key}" -> "{value}";\n')
    f.write("}\n")


def print_dependencies(dependencies: Dict[str, List[str]]):
  for key, values in dependencies.items():
    for value in values:
      print(f"{key} -> {value}")


def main():
  parser = argparse.ArgumentParser(description=__doc__)
  parser.add_argument(
      "packages", metavar="PKG", nargs="+", help="Go package list")
  args = parser.parse_args()

  with tempfile.TemporaryDirectory() as tmp_dir:
    os.mkdir(os.path.join(tmp_dir, "src"))

    print("Downloading packages...")
    download_packages(args.packages, tmp_dir)

    print("Analyzing dependencies...")
    dependencies = {}
    for pkg in args.packages:
      print(f"  {pkg}")
      dependencies.update(
          parse_go_mod(os.path.join(tmp_dir, os.path.basename(pkg))))

    print("Removing not interesting dependencies...")
    dependencies = remove_not_interesting_dependencies(dependencies,
                                                       args.packages)

    print_dependencies(dependencies)

    print("Generating dependency graph...")
    generate_dependency_graph_dot_file(dependencies, "graph.dot")

    print("Generating dependency graph image...")
    cmd = ["dot", "-Tpng", "graph.dot", "-o", "graph.png"]
    print(f"+ {shlex.join(cmd)}")
    subprocess.run(cmd, check=True)

    print("Done.")


if __name__ == "__main__":
  sys.exit(main())