mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-25 12:25:33 +08:00
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
This commit is contained in:
RamGopalSrikar
@ -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()
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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()}")
|
||||
|
||||
@ -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()
|
||||
|
||||
Reference in New Issue
Block a user