From f0fd0040e840809e4487efea8bc9d3207727f965 Mon Sep 17 00:00:00 2001 From: RamGopalSrikar Date: Wed, 18 Mar 2026 10:23:42 -0400 Subject: [PATCH] fix: resolve ruff linting issues in GP scripts - Fix quote style, import ordering, docstring format - Use Path instead of os.makedirs/open/os.path.join - Add missing timeouts to requests calls - Move noqa comments to correct lines for S501 --- scripts/gp/download_translations.py | 32 ++++++----- scripts/gp/gp_client.py | 85 ++++++++++++++++++----------- scripts/gp/test_auth.py | 60 ++++++++++---------- scripts/gp/upload_strings.py | 19 ++++--- 4 files changed, 110 insertions(+), 86 deletions(-) diff --git a/scripts/gp/download_translations.py b/scripts/gp/download_translations.py index 8ef371c6c6..cc8ea7c07b 100644 --- a/scripts/gp/download_translations.py +++ b/scripts/gp/download_translations.py @@ -1,21 +1,23 @@ -""" -Download translated strings from GP and save as locale JSON files. +"""Download translated strings from GP and save as locale JSON files. Usage: python download_translations.py --output path/to/locales/ """ -import json -import os + import argparse -from gp_client import get_strings, TARGET_LANGS, GP_BUNDLE +import json +from pathlib import Path + +from gp_client import TARGET_LANGS, get_strings def main(): parser = argparse.ArgumentParser(description="Download translations from GP") - parser.add_argument('--output', required=True, help='Directory to save translated JSON files') + parser.add_argument("--output", required=True, help="Directory to save translated JSON files") args = parser.parse_args() - os.makedirs(args.output, exist_ok=True) + output_dir = Path(args.output) + output_dir.mkdir(parents=True, exist_ok=True) for lang in TARGET_LANGS: print(f"Downloading '{lang}' translations...") @@ -24,25 +26,27 @@ def main(): # Extract just the key:value strings from the response strings = { - key: entry.get('value', '') if isinstance(entry, dict) else entry - for key, entry in result.get('resourceStrings', {}).items() + key: entry.get("value", "") if isinstance(entry, dict) else entry + for key, entry in result.get("resourceStrings", {}).items() } if not strings: print(f" No strings yet for '{lang}' (translation may still be in progress)") continue - output_file = os.path.join(args.output, f"{lang}.json") - with open(output_file, 'w', encoding='utf-8') as f: - json.dump(strings, f, ensure_ascii=False, indent=2) + output_file = output_dir / f"{lang}.json" + output_file.write_text( + json.dumps(strings, ensure_ascii=False, indent=2), + encoding="utf-8", + ) print(f" Saved {len(strings)} strings to {output_file}") - except Exception as e: + except Exception as e: # noqa: BLE001 print(f" Error downloading '{lang}': {e}") print("\nDone.") -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/scripts/gp/gp_client.py b/scripts/gp/gp_client.py index 1fb0c923e1..aef7ae01d8 100644 --- a/scripts/gp/gp_client.py +++ b/scripts/gp/gp_client.py @@ -1,80 +1,99 @@ -""" -GP (Globalization Pipeline) REST API client. +"""GP (Globalization Pipeline) REST API client. + Handles HMAC authentication and common API operations. """ -import hmac -import hashlib + import base64 +import hashlib +import hmac import json import os -import requests 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'] +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 def get_headers(url, method, body=None): """Generate GP-HMAC auth headers. url must be the full URL.""" date = datetime.now(timezone.utc) - dateString = date.strftime('%a, %d %b %Y %H:%M:%S %Z').replace('UTC', 'GMT') + date_string = date.strftime("%a, %d %b %Y %H:%M:%S %Z").replace("UTC", "GMT") - if method.upper() == 'GET': - msg = 'GET' + '\n' + url + '\n' + dateString + '\n' + if method.upper() == "GET": + msg = "GET" + "\n" + url + "\n" + date_string + "\n" else: - msg = method.upper() + '\n' + url + '\n' + dateString + '\n' + json.dumps(body) + 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') + message = bytes(msg, "ISO-8859-1") + password = bytes(GP_PASSWORD, "ISO-8859-1") - signature = hmac.new(password, msg=message, digestmod=hashlib.sha1).digest() - hmacHeader = 'GP-HMAC ' + GP_USER_ID + ':' + base64.b64encode(signature).decode() + signature = hmac.new(password, msg=message, digestmod=hashlib.sha1).digest() + hmac_header = "GP-HMAC " + GP_USER_ID + ":" + base64.b64encode(signature).decode() headers = { - 'Authorization': hmacHeader, - 'GP-Date': dateString, - 'accept': 'application/json', + "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' + 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=False) + response = requests.get(url, headers=get_headers(url, "GET"), verify=False, timeout=REQUEST_TIMEOUT) # noqa: S501 response.raise_for_status() return response.json() -def create_bundle(source_lang='en'): - url = f"{BASE_URL}/{GP_INSTANCE}/v2/bundles/{GP_BUNDLE}" +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=False) + response = requests.put( + url, + headers=get_headers(url, "PUT", body), + json=body, + verify=False, # noqa: S501 + timeout=REQUEST_TIMEOUT, + ) response.raise_for_status() return response.json() -def upload_strings(strings, lang='en'): +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=False) + response = requests.put( + url, + headers=get_headers(url, "PUT", strings), + json=strings, + verify=False, # noqa: S501 + 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=False) + response = requests.get(url, headers=get_headers(url, "GET"), verify=False, timeout=REQUEST_TIMEOUT) # noqa: S501 response.raise_for_status() return response.json() diff --git a/scripts/gp/test_auth.py b/scripts/gp/test_auth.py index cb560f4882..57b3bee3e8 100644 --- a/scripts/gp/test_auth.py +++ b/scripts/gp/test_auth.py @@ -1,9 +1,12 @@ -import hmac -import hashlib -from datetime import datetime, timezone +"""Test GP authentication and basic API connectivity.""" + import base64 +import hashlib +import hmac import json import os +from datetime import datetime, timezone + import requests from dotenv import load_dotenv @@ -11,63 +14,60 @@ load_dotenv() def calulate_hmacheader(endpoint, request_type, body): - """ - This function calculate hmacheader dynamically - """ + """Calculate HMAC header dynamically.""" date = datetime.now(timezone.utc) - dateString = date.strftime('%a, %d %b %Y %H:%M:%S %Z').replace('UTC', 'GMT') + date_string = date.strftime("%a, %d %b %Y %H:%M:%S %Z").replace("UTC", "GMT") - userId = os.getenv('GP_ADMIN_USER_ID') - password = os.getenv('GP_ADMIN_PASSWORD') + user_id = os.getenv("GP_ADMIN_USER_ID") + password = os.getenv("GP_ADMIN_PASSWORD") - if request_type.upper() == 'GET': - body = '' - msg = 'GET'+ '\n' + endpoint + '\n' + dateString + '\n' + body + if request_type.upper() == "GET": + body = "" + msg = "GET" + "\n" + endpoint + "\n" + date_string + "\n" + body else: - msg = request_type.upper() + '\n' + endpoint + '\n' + dateString + '\n' + json.dumps(body) + msg = request_type.upper() + "\n" + endpoint + "\n" + date_string + "\n" + json.dumps(body) - print(f"[DEBUG] userId : {userId}") - print(f"[DEBUG] dateString: {dateString}") + print(f"[DEBUG] userId : {user_id}") + print(f"[DEBUG] dateString: {date_string}") print(f"[DEBUG] endpoint : {endpoint}") - print(f"[DEBUG] msg : {repr(msg)}") + print(f"[DEBUG] msg : {msg!r}") message = bytes(msg, "ISO-8859-1") password = bytes(password, "ISO-8859-1") signature = hmac.new(password, msg=message, digestmod=hashlib.sha1).digest() - hmacHeader = 'GP-HMAC ' + userId + ':' + base64.b64encode(signature).decode() + hmac_header = "GP-HMAC " + user_id + ":" + base64.b64encode(signature).decode() print(f"[DEBUG] signature : {base64.b64encode(signature).decode()}") - print(f"[DEBUG] auth : {hmacHeader}") + print(f"[DEBUG] auth : {hmac_header}") - if request_type.upper() == 'PATCH': + if request_type.upper() == "PATCH": header = { - 'Authorization': hmacHeader, - 'GP-Date': dateString, - 'accept': 'application/json', - 'Content-Type': "application/merge-patch+json" + "Authorization": hmac_header, + "GP-Date": date_string, + "accept": "application/json", + "Content-Type": "application/merge-patch+json", } else: header = { - 'Authorization': hmacHeader, - 'GP-Date': dateString, - 'accept': 'application/json' + "Authorization": hmac_header, + "GP-Date": date_string, + "accept": "application/json", } return header # Test: List bundles -GP_INSTANCE = os.getenv('GP_INSTANCE', 'langflow-test') +GP_INSTANCE = os.getenv("GP_INSTANCE", "langflow-test") BASE_URL = "https://g11n-pipeline-api.straker.global/translate/rest" endpoint = f"{BASE_URL}/{GP_INSTANCE}/v2/bundles" - print(f"[DEBUG] full URL : {endpoint}") print() -headers = calulate_hmacheader(endpoint, 'GET', '') +headers = calulate_hmacheader(endpoint, "GET", "") print() -response = requests.get(endpoint, headers=headers, verify=False) +response = requests.get(endpoint, headers=headers, verify=False, timeout=30) # noqa: S501 print(f"[DEBUG] status : {response.status_code}") print(f"[DEBUG] response : {response.json()}") diff --git a/scripts/gp/upload_strings.py b/scripts/gp/upload_strings.py index 520e2dcda5..28e01d2d3e 100644 --- a/scripts/gp/upload_strings.py +++ b/scripts/gp/upload_strings.py @@ -1,27 +1,28 @@ -""" -Upload English source strings to GP. +"""Upload English source strings to GP. Usage: python upload_strings.py --source path/to/en.json """ -import json + import argparse -from gp_client import list_bundles, create_bundle, upload_strings, GP_BUNDLE +import json +from pathlib import Path + +from gp_client import GP_BUNDLE, create_bundle, list_bundles, upload_strings def main(): parser = argparse.ArgumentParser(description="Upload strings to GP") - parser.add_argument('--source', required=True, help='Path to English source JSON file') + parser.add_argument("--source", required=True, help="Path to English source JSON file") args = parser.parse_args() # Load source strings - with open(args.source, 'r', encoding='utf-8') as f: - strings = json.load(f) + strings = json.loads(Path(args.source).read_text(encoding="utf-8")) print(f"Loaded {len(strings)} strings from {args.source}") # Create bundle if it doesn't exist existing = list_bundles() - if GP_BUNDLE not in existing.get('bundleIds', []): + if GP_BUNDLE not in existing.get("bundleIds", []): print(f"Creating bundle '{GP_BUNDLE}'...") create_bundle() print("Bundle created.") @@ -34,5 +35,5 @@ def main(): print(f"Done: {result}") -if __name__ == '__main__': +if __name__ == "__main__": main()