codeowners-pipeline.yml

stages:
  - notify

notify_codeowners:
  stage: notify
  script:
    - |
      cat > notify_codeowners.py <<'EOF'
      import os, sys, json, fnmatch, urllib.request, urllib.parse

      CODEOWNERS = os.environ.get('CODEOWNERS', '')
      CI_API_V4_URL = os.environ.get('CI_API_V4_URL')
      CI_PROJECT_ID = os.environ.get('CI_PROJECT_ID')
      MR_IID = os.environ.get('CI_MERGE_REQUEST_IID')
      GITLAB_TOKEN = os.environ.get('GITLAB_API_TOKEN')

      if not MR_IID:
          print("not a merge request - exiting")
          sys.exit(0)
      if not GITLAB_TOKEN:
          print("missing GITLAB_API_TOKEN")
          sys.exit(1)

      def api_get(url):
          req = urllib.request.Request(url)
          req.add_header("PRIVATE-TOKEN", GITLAB_TOKEN)
          with urllib.request.urlopen(req) as response:
              return response.read().decode('utf-8')

      def api_post(url, data):
          data_encoded = urllib.parse.urlencode(data).encode('utf-8')
          req = urllib.request.Request(url, data=data_encoded, method="POST")
          req.add_header("PRIVATE-TOKEN", GITLAB_TOKEN)
          with urllib.request.urlopen(req) as response:
              return response.read().decode('utf-8')

      def api_put(url, data):
          data_encoded = urllib.parse.urlencode(data).encode('utf-8')
          req = urllib.request.Request(url, data=data_encoded, method="PUT")
          req.add_header("PRIVATE-TOKEN", GITLAB_TOKEN)
          with urllib.request.urlopen(req) as response:
              return response.read().decode('utf-8')

      try:
          mr_changes_url = f"{CI_API_V4_URL}/projects/{CI_PROJECT_ID}/merge_requests/{MR_IID}/changes"
          changes_resp = api_get(mr_changes_url)
          changes = json.loads(changes_resp).get("changes", [])
      except Exception as e:
          print("failed to fetch mr changes:", e)
          sys.exit(1)

      changed_files = [change["new_path"] for change in changes]

      rules = []
      for line in CODEOWNERS.splitlines():
          line = line.strip()
          if not line or line.startswith("#"):
              continue
          parts = line.split()
          pattern = parts[0]
          owners = parts[1:]
          rules.append((pattern, owners))

      owner_files = {}
      for file in changed_files:
          matched_owners = None
          for pattern, owners in rules:
              p = pattern
              if p.startswith("/"):
                  p = p[1:]
              if p.endswith("/"):
                  p += "*"
              if fnmatch.fnmatch(file, p):
                  matched_owners = owners
          if matched_owners:
              for owner in matched_owners:
                  owner_files.setdefault(owner, set()).add(file)

      if not owner_files:
          print("no matching codeowners found")
          sys.exit(0)

      marker = "<!-- codeowners-notify -->"
      lines = [marker, "## codeowners notification", ""]
      for owner, files in owner_files.items():
          lines.append(f"{owner}:")
          for f in sorted(files):
              lines.append(f"- {f}")
          lines.append("")
      comment_body = "\n".join(lines)

      try:
          notes_url = f"{CI_API_V4_URL}/projects/{CI_PROJECT_ID}/merge_requests/{MR_IID}/notes"
          notes_resp = api_get(notes_url)
          notes = json.loads(notes_resp)
      except Exception as e:
          print("failed to fetch mr notes:", e)
          sys.exit(1)

      note_id = None
      for note in notes:
          if marker in note.get("body", ""):
              note_id = note.get("id")
              break

      if note_id:
          update_url = f"{notes_url}/{note_id}"
          try:
              api_put(update_url, {"body": comment_body})
              print("updated codeowners notification")
          except Exception as e:
              print("failed to update note:", e)
              sys.exit(1)
      else:
          try:
              api_post(notes_url, {"body": comment_body})
              print("created codeowners notification")
          except Exception as e:
              print("failed to create note:", e)
              sys.exit(1)
      EOF
    - python notify_codeowners.py
  only:
    - merge_requests
  variables:
    CODEOWNERS: |
      # sample codeowners file in github syntax
      *.js @frontend-team
      /docs/ @doc-team