|
#!/usr/bin/env python3 |
|
# SPDX-License-Identifier: MIT |
|
# |
|
# merge_mr.py -- Merge a Tencent Working merge request via REST API. |
|
# |
|
# Tencent Working (腾讯工蜂, https://git.tencent.com) does not currently expose |
|
# a "merge this MR" button to its CLI tooling, so day-to-day MR completion has |
|
# to go through the v3 REST API. This script wraps the three calls you actually |
|
# need (project lookup -> MR lookup -> PUT merge) and prints a concise summary. |
|
# |
|
# Quick start: |
|
# |
|
# # 1. Generate a Personal Access Token at: |
|
# # https://git.tencent.com/profile/personal_access_tokens |
|
# # Required scope: api (read+write). |
|
# # 2. Export it (or put it in ~/.netrc, see --help): |
|
# export TENCENT_GIT_TOKEN=xxxxxxxxxxxxxxxxxxxx |
|
# # 3. Merge: |
|
# ./merge_mr.py CGameAIPartner/AISquad/Protocol 150 |
|
# |
|
# Common options: |
|
# |
|
# --squash Squash-merge instead of a merge commit. |
|
# --remove-source-branch Delete the source branch after merge. |
|
# --message "..." Custom merge commit message. |
|
# --dry-run Look up the MR and print what *would* be sent; |
|
# do not call PUT. Default-on guard for safety; |
|
# pass --no-dry-run (or -y) to actually merge. |
|
# --host git.example.com Use a different Tencent Working deployment. |
|
# |
|
# Authentication priority: |
|
# |
|
# 1. --token <value> |
|
# 2. $TENCENT_GIT_TOKEN environment variable |
|
# 3. ~/.netrc entry whose machine matches the host (or its parent domain) |
|
# |
|
# Exit codes: |
|
# |
|
# 0 merge succeeded (or dry-run preview printed) |
|
# 1 generic / network / unexpected error |
|
# 2 bad arguments |
|
# 3 authentication missing or rejected |
|
# 4 MR not in a mergeable state (already merged, closed, WIP, conflicts) |
|
# |
|
# Notes on the underlying API (verified empirically against git.tencent.com, |
|
# 2026-05): |
|
# |
|
# * The API is rooted at /api/v3/, not /api/v4/. |
|
# * Auth header is `PRIVATE-TOKEN: <pat>`, the standard GitLab PAT scheme. |
|
# `Authorization: Bearer ...` returns 401 here. |
|
# * Project lookup accepts a URL-encoded full path, including 3+ level |
|
# namespaces (e.g. `a/b/c/repo`). |
|
# * The merge endpoint is *singular*: |
|
# PUT /api/v3/projects/<pid>/merge_request/<INTERNAL_ID>/merge |
|
# -- not the plural `merge_requests` you'd expect from upstream GitLab, |
|
# and it takes the MR's *internal* id (the global numeric `id`), not the |
|
# per-project iid you see in the URL bar. To resolve internal id from iid: |
|
# GET /api/v3/projects/<pid>/merge_requests?iid=<iid> -> [0].id |
|
# |
|
# Author: chen3feng (https://github.com/chen3feng) |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import json |
|
import netrc |
|
import os |
|
import sys |
|
import urllib.error |
|
import urllib.parse |
|
import urllib.request |
|
from typing import Any |
|
|
|
|
|
DEFAULT_HOST = "git.tencent.com" |
|
USER_AGENT = "merge-mr.py/1.0 (+https://gist.github.com/chen3feng)" |
|
|
|
|
|
# --------------------------------------------------------------------------- |
|
# Auth |
|
# --------------------------------------------------------------------------- |
|
|
|
def resolve_token(cli_token: str | None, host: str) -> str | None: |
|
"""Return a PAT, trying --token, env var, then ~/.netrc. |
|
|
|
For ~/.netrc we accept either an exact host match (`git.tencent.com`) or |
|
its registrable parent (`tencent.com`), since some helper tools store |
|
credentials under the parent domain. |
|
""" |
|
if cli_token: |
|
return cli_token |
|
|
|
env = os.environ.get("TENCENT_GIT_TOKEN") |
|
if env: |
|
return env |
|
|
|
try: |
|
rc = netrc.netrc() |
|
except (FileNotFoundError, netrc.NetrcParseError): |
|
return None |
|
|
|
candidates = [host] |
|
parts = host.split(".") |
|
if len(parts) > 2: |
|
candidates.append(".".join(parts[-2:])) |
|
for cand in candidates: |
|
entry = rc.authenticators(cand) |
|
if entry: |
|
_login, _account, password = entry |
|
if password: |
|
return password |
|
return None |
|
|
|
|
|
# --------------------------------------------------------------------------- |
|
# Tiny REST helper |
|
# --------------------------------------------------------------------------- |
|
|
|
class ApiError(RuntimeError): |
|
def __init__(self, status: int, body: str, url: str): |
|
super().__init__(f"HTTP {status} for {url}: {body[:300]}") |
|
self.status = status |
|
self.body = body |
|
self.url = url |
|
|
|
|
|
def api_request( |
|
method: str, |
|
url: str, |
|
token: str, |
|
form: dict[str, str] | None = None, |
|
) -> Any: |
|
"""Make a REST call and return parsed JSON (or raw text if not JSON).""" |
|
data = None |
|
headers = { |
|
"PRIVATE-TOKEN": token, |
|
"Accept": "application/json", |
|
"User-Agent": USER_AGENT, |
|
} |
|
if form is not None: |
|
data = urllib.parse.urlencode(form).encode("utf-8") |
|
headers["Content-Type"] = "application/x-www-form-urlencoded" |
|
|
|
req = urllib.request.Request(url, data=data, method=method, headers=headers) |
|
try: |
|
with urllib.request.urlopen(req, timeout=30) as resp: |
|
raw = resp.read().decode("utf-8", errors="replace") |
|
status = resp.status |
|
except urllib.error.HTTPError as e: |
|
raw = e.read().decode("utf-8", errors="replace") if e.fp else "" |
|
raise ApiError(e.code, raw, url) from None |
|
except urllib.error.URLError as e: |
|
raise RuntimeError(f"Network error contacting {url}: {e.reason}") from None |
|
|
|
if status >= 400: |
|
raise ApiError(status, raw, url) |
|
if not raw: |
|
return None |
|
try: |
|
return json.loads(raw) |
|
except json.JSONDecodeError: |
|
return raw |
|
|
|
|
|
# --------------------------------------------------------------------------- |
|
# High-level operations |
|
# --------------------------------------------------------------------------- |
|
|
|
def lookup_project(host: str, token: str, full_path: str) -> dict: |
|
encoded = urllib.parse.quote(full_path, safe="") |
|
return api_request("GET", f"https://{host}/api/v3/projects/{encoded}", token) |
|
|
|
|
|
def lookup_mr(host: str, token: str, project_id: int, iid: int) -> dict: |
|
url = ( |
|
f"https://{host}/api/v3/projects/{project_id}/merge_requests" |
|
f"?iid={iid}" |
|
) |
|
result = api_request("GET", url, token) |
|
if not isinstance(result, list) or not result: |
|
raise SystemExit( |
|
f"error: MR with iid={iid} not found in project {project_id}" |
|
) |
|
return result[0] |
|
|
|
|
|
def merge_mr( |
|
host: str, |
|
token: str, |
|
project_id: int, |
|
internal_id: int, |
|
*, |
|
squash: bool, |
|
remove_source_branch: bool, |
|
message: str | None, |
|
) -> dict: |
|
form: dict[str, str] = {} |
|
if squash: |
|
form["squash"] = "true" |
|
if remove_source_branch: |
|
form["should_remove_source_branch"] = "true" |
|
if message: |
|
form["merge_commit_message"] = message |
|
|
|
# NB: singular `merge_request`, internal id, not iid. |
|
url = ( |
|
f"https://{host}/api/v3/projects/{project_id}" |
|
f"/merge_request/{internal_id}/merge" |
|
) |
|
return api_request("PUT", url, token, form=form or None) |
|
|
|
|
|
# --------------------------------------------------------------------------- |
|
# CLI |
|
# --------------------------------------------------------------------------- |
|
|
|
def parse_args(argv: list[str]) -> argparse.Namespace: |
|
p = argparse.ArgumentParser( |
|
prog="merge_mr.py", |
|
description="Merge a Tencent Working merge request via REST API.", |
|
formatter_class=argparse.RawDescriptionHelpFormatter, |
|
epilog=( |
|
"Examples:\n" |
|
" ./merge_mr.py CGameAIPartner/AISquad/Protocol 150\n" |
|
" ./merge_mr.py group/proj 42 --squash --remove-source-branch -y\n" |
|
" ./merge_mr.py group/proj 42 --message 'Merge !42 into master' -y\n" |
|
), |
|
) |
|
p.add_argument( |
|
"project", |
|
help="Full project path, e.g. group/subgroup/repo (any nesting depth).", |
|
) |
|
p.add_argument( |
|
"iid", |
|
type=int, |
|
help="MR iid as shown in the web URL (.../merge_requests/<iid>).", |
|
) |
|
p.add_argument( |
|
"--host", |
|
default=os.environ.get("TENCENT_GIT_HOST", DEFAULT_HOST), |
|
help=f"API host (default: {DEFAULT_HOST}, or $TENCENT_GIT_HOST).", |
|
) |
|
p.add_argument( |
|
"--token", |
|
help="Personal access token (defaults to $TENCENT_GIT_TOKEN or ~/.netrc).", |
|
) |
|
p.add_argument("--squash", action="store_true", help="Squash-merge.") |
|
p.add_argument( |
|
"--remove-source-branch", |
|
action="store_true", |
|
help="Delete source branch after merge.", |
|
) |
|
p.add_argument( |
|
"--message", |
|
help="Custom merge commit message (only used for non-squash merges).", |
|
) |
|
|
|
g = p.add_mutually_exclusive_group() |
|
g.add_argument( |
|
"-y", |
|
"--yes", |
|
dest="dry_run", |
|
action="store_false", |
|
help="Actually perform the merge (default is dry-run).", |
|
) |
|
g.add_argument( |
|
"--dry-run", |
|
dest="dry_run", |
|
action="store_true", |
|
help="Preview the request without sending the PUT (default).", |
|
) |
|
p.set_defaults(dry_run=True) |
|
|
|
return p.parse_args(argv) |
|
|
|
|
|
def main(argv: list[str]) -> int: |
|
args = parse_args(argv) |
|
|
|
if "/" not in args.project: |
|
print( |
|
f"error: project must look like 'group/repo' or 'group/sub/repo' " |
|
f"(got {args.project!r})", |
|
file=sys.stderr, |
|
) |
|
return 2 |
|
|
|
token = resolve_token(args.token, args.host) |
|
if not token: |
|
print( |
|
f"error: no token found. Pass --token, set $TENCENT_GIT_TOKEN, or " |
|
f"add a 'machine {args.host}' entry to ~/.netrc.", |
|
file=sys.stderr, |
|
) |
|
return 3 |
|
|
|
try: |
|
project = lookup_project(args.host, token, args.project) |
|
except ApiError as e: |
|
if e.status == 401: |
|
print("error: 401 Unauthorized -- token rejected.", file=sys.stderr) |
|
return 3 |
|
if e.status == 404: |
|
print( |
|
f"error: project {args.project!r} not found on {args.host} " |
|
f"(or token lacks access).", |
|
file=sys.stderr, |
|
) |
|
return 1 |
|
print(f"error: project lookup failed: {e}", file=sys.stderr) |
|
return 1 |
|
|
|
pid = project["id"] |
|
print(f"project: {project['path_with_namespace']} (id={pid})") |
|
|
|
mr = lookup_mr(args.host, token, pid, args.iid) |
|
internal_id = mr["id"] |
|
state = mr.get("state") |
|
merge_status = mr.get("merge_status") |
|
print( |
|
f"MR !{args.iid}: {mr.get('title','')!r}\n" |
|
f" internal_id = {internal_id}\n" |
|
f" state = {state}\n" |
|
f" merge_status= {merge_status}\n" |
|
f" source = {mr.get('source_branch')} -> {mr.get('target_branch')}" |
|
) |
|
|
|
if state != "opened": |
|
print( |
|
f"error: MR is in state {state!r}, refusing to merge " |
|
f"(expected 'opened').", |
|
file=sys.stderr, |
|
) |
|
return 4 |
|
if merge_status not in ("can_be_merged", "unchecked"): |
|
print( |
|
f"error: merge_status is {merge_status!r}, refusing to merge.", |
|
file=sys.stderr, |
|
) |
|
return 4 |
|
|
|
plan = [] |
|
if args.squash: |
|
plan.append("squash=true") |
|
if args.remove_source_branch: |
|
plan.append("should_remove_source_branch=true") |
|
if args.message: |
|
plan.append(f"merge_commit_message={args.message!r}") |
|
plan_str = ", ".join(plan) if plan else "(no extra options)" |
|
|
|
if args.dry_run: |
|
print(f"\n[dry-run] would PUT merge_request/{internal_id}/merge {plan_str}") |
|
print("[dry-run] re-run with -y / --yes to actually merge.") |
|
return 0 |
|
|
|
print(f"\nmerging... ({plan_str})") |
|
try: |
|
result = merge_mr( |
|
args.host, |
|
token, |
|
pid, |
|
internal_id, |
|
squash=args.squash, |
|
remove_source_branch=args.remove_source_branch, |
|
message=args.message, |
|
) |
|
except ApiError as e: |
|
print(f"error: merge call failed: {e}", file=sys.stderr) |
|
return 4 if e.status in (403, 405, 406, 409) else 1 |
|
|
|
final_state = result.get("state") if isinstance(result, dict) else None |
|
print(f"done. final state = {final_state}") |
|
if isinstance(result, dict): |
|
url = result.get("web_url") |
|
if url: |
|
print(f" {url}") |
|
return 0 |
|
|
|
|
|
if __name__ == "__main__": |
|
sys.exit(main(sys.argv[1:])) |