Files
langflow/scripts/gp/gp_client.py
RamGopalSrikar a4540bbcdf feat(i18n): refactor language constants and improve GP pipeline
- Extract SUPPORTED_LANGUAGES into src/frontend/src/constants/languages.ts
- Update LanguageForm to import from shared constants, add aria-label
- Add settings.languageSelectAriaLabel key to en.json
- Update LanguageSelector component to use shared SUPPORTED_LANGUAGES
- Fix gp_client and download_translations scripts
- Update GP upload/download/nightly GitHub Actions workflows
- Add GP script tests
2026-03-26 17:12:28 -04:00

101 lines
3.1 KiB
Python

"""GP (Globalization Pipeline) REST API client.
Handles HMAC authentication and common API operations.
"""
import base64
import hashlib
import hmac
import json
import os
from datetime import datetime, timezone
import requests
from dotenv import load_dotenv
load_dotenv()
BASE_URL = "https://g11n-pipeline-api.straker.global/translate/rest"
GP_USER_ID = os.getenv("GP_ADMIN_USER_ID")
GP_PASSWORD = os.getenv("GP_ADMIN_PASSWORD")
GP_INSTANCE = os.getenv("GP_INSTANCE", "langflow-test")
GP_BUNDLE = os.getenv("GP_BUNDLE", "langflow-ui")
TARGET_LANGS = ["fr", "ja", "es", "de", "pt", "zh-Hans"]
REQUEST_TIMEOUT = 30
VERIFY_SSL = os.getenv("GP_VERIFY_SSL", "true").lower() != "false"
def get_headers(url, method, body=None):
"""Generate GP-HMAC auth headers. url must be the full URL."""
date = datetime.now(timezone.utc)
date_string = date.strftime("%a, %d %b %Y %H:%M:%S %Z").replace("UTC", "GMT")
if method.upper() == "GET":
msg = "GET" + "\n" + url + "\n" + date_string + "\n"
else:
msg = method.upper() + "\n" + url + "\n" + date_string + "\n" + json.dumps(body)
message = bytes(msg, "ISO-8859-1")
password = bytes(GP_PASSWORD, "ISO-8859-1")
signature = hmac.new(password, msg=message, digestmod=hashlib.sha1).digest()
hmac_header = "GP-HMAC " + GP_USER_ID + ":" + base64.b64encode(signature).decode()
headers = {
"Authorization": hmac_header,
"GP-Date": date_string,
"accept": "application/json",
}
if method.upper() == "PATCH":
headers["Content-Type"] = "application/merge-patch+json"
elif method.upper() in ("PUT", "POST"):
headers["Content-Type"] = "application/json"
return headers
def list_bundles():
"""List all bundles in the GP instance."""
url = f"{BASE_URL}/{GP_INSTANCE}/v2/bundles"
response = requests.get(url, headers=get_headers(url, "GET"), verify=VERIFY_SSL, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
return response.json()
def create_bundle(source_lang="en"):
"""Create a new bundle in GP."""
url = f"{BASE_URL}/{GP_INSTANCE}/v2/bundles/{GP_BUNDLE}"
body = {"sourceLanguage": source_lang, "targetLanguages": TARGET_LANGS}
response = requests.put(
url,
headers=get_headers(url, "PUT", body),
json=body,
verify=VERIFY_SSL,
timeout=REQUEST_TIMEOUT,
)
response.raise_for_status()
return response.json()
def upload_strings(strings, lang="en"):
"""Upload resource strings for a language to GP."""
url = f"{BASE_URL}/{GP_INSTANCE}/v2/bundles/{GP_BUNDLE}/{lang}"
response = requests.put(
url,
headers=get_headers(url, "PUT", strings),
json=strings,
verify=VERIFY_SSL,
timeout=REQUEST_TIMEOUT,
)
response.raise_for_status()
return response.json()
def get_strings(lang):
"""Download translated strings for a language from GP."""
url = f"{BASE_URL}/{GP_INSTANCE}/v2/bundles/{GP_BUNDLE}/{lang}"
response = requests.get(url, headers=get_headers(url, "GET"), verify=VERIFY_SSL, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
return response.json()