chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,308 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
TypeScript Executor - Execute TypeScript methods from Python
|
||||
Supports calling TypeScript functions with complex parameters and async/await
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
class TypeScriptExecutor:
|
||||
"""TypeScript method executor"""
|
||||
|
||||
def __init__(self, working_dir: Optional[str] = None):
|
||||
"""Initialize TypeScript executor
|
||||
|
||||
Args:
|
||||
working_dir: Working directory, defaults to current directory
|
||||
"""
|
||||
self.working_dir = working_dir or os.getcwd()
|
||||
|
||||
def call_method(self, ts_file_path: str, method_name: str, *args, **kwargs) -> Any:
|
||||
"""Call TypeScript method
|
||||
|
||||
Args:
|
||||
ts_file_path: TypeScript file path
|
||||
method_name: Method name
|
||||
*args: Positional arguments
|
||||
**kwargs: Keyword arguments
|
||||
|
||||
Returns:
|
||||
Execution result
|
||||
"""
|
||||
|
||||
# Convert relative path to absolute path
|
||||
if not os.path.isabs(ts_file_path):
|
||||
ts_file_path = os.path.join(self.working_dir, ts_file_path)
|
||||
|
||||
# Ensure the target file exists
|
||||
if not os.path.exists(ts_file_path):
|
||||
raise FileNotFoundError(f"TypeScript file not found: {ts_file_path}")
|
||||
|
||||
# Get the directory of the target file
|
||||
target_dir = os.path.dirname(ts_file_path)
|
||||
|
||||
# Create wrapper script
|
||||
wrapper_code = self._create_wrapper_script(
|
||||
ts_file_path, method_name, list(args), kwargs
|
||||
)
|
||||
|
||||
# Create temporary file in the same directory as the target file
|
||||
temp_fd, temp_file = tempfile.mkstemp(suffix=".ts", dir=target_dir)
|
||||
|
||||
try:
|
||||
# Write wrapper script
|
||||
with os.fdopen(temp_fd, "w", encoding="utf-8") as f:
|
||||
f.write(wrapper_code)
|
||||
|
||||
# Execute TypeScript code using ts-node
|
||||
# Use subprocess.Popen to capture output in real-time
|
||||
process = subprocess.Popen(
|
||||
["npx", "ts-node", temp_file],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
cwd=self.working_dir,
|
||||
bufsize=1, # Line buffering
|
||||
universal_newlines=True,
|
||||
)
|
||||
|
||||
stdout_lines = []
|
||||
stderr_lines = []
|
||||
|
||||
# Read output line by line and display console.log in real-time
|
||||
while True:
|
||||
output = process.stdout.readline()
|
||||
if output == "" and process.poll() is not None:
|
||||
break
|
||||
if output:
|
||||
line = output.strip()
|
||||
stdout_lines.append(line)
|
||||
|
||||
# Try to parse as JSON to see if it's the final result
|
||||
try:
|
||||
json.loads(line)
|
||||
# If it parses as JSON, it might be the final result, don't print it yet
|
||||
except json.JSONDecodeError:
|
||||
# If it's not JSON, it's likely a console.log, so print it
|
||||
print(line)
|
||||
|
||||
# Get any remaining stderr
|
||||
stderr_output = process.stderr.read()
|
||||
if stderr_output:
|
||||
stderr_lines.append(stderr_output.strip())
|
||||
|
||||
return_code = process.poll()
|
||||
|
||||
if return_code != 0:
|
||||
error_msg = "\n".join(stderr_lines) if stderr_lines else "Unknown error"
|
||||
raise RuntimeError(f"TypeScript execution failed: {error_msg}")
|
||||
|
||||
# Parse results from the last line that looks like JSON
|
||||
for line in reversed(stdout_lines):
|
||||
if line.strip():
|
||||
try:
|
||||
# Try to parse as JSON
|
||||
return json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
# If no JSON found, return the last non-empty line
|
||||
for line in reversed(stdout_lines):
|
||||
if line.strip():
|
||||
return line
|
||||
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Execution error: {str(e)}")
|
||||
finally:
|
||||
# Clean up temporary file
|
||||
os.unlink(temp_file)
|
||||
|
||||
def _create_wrapper_script(
|
||||
self,
|
||||
ts_file_path: str,
|
||||
method_name: str,
|
||||
args: List[Any],
|
||||
kwargs: Dict[str, Any],
|
||||
) -> str:
|
||||
"""Create wrapper script
|
||||
|
||||
Args:
|
||||
ts_file_path: TypeScript file path
|
||||
method_name: Method name
|
||||
args: Positional arguments
|
||||
kwargs: Keyword arguments
|
||||
|
||||
Returns:
|
||||
Wrapper script code
|
||||
"""
|
||||
|
||||
# Use relative path for import, since temp file is in the same directory
|
||||
ts_filename = os.path.basename(ts_file_path)
|
||||
|
||||
# Remove .ts extension, since import doesn't need it
|
||||
if ts_filename.endswith(".ts"):
|
||||
import_path = "./" + ts_filename[:-3]
|
||||
else:
|
||||
import_path = "./" + ts_filename
|
||||
|
||||
args_json = json.dumps(args)
|
||||
kwargs_json = json.dumps(kwargs)
|
||||
|
||||
wrapper_code = f"""
|
||||
import * as targetModule from '{import_path}';
|
||||
|
||||
async function executeMethod() {{
|
||||
try {{
|
||||
// Prepare arguments
|
||||
const args: any[] = {args_json};
|
||||
const kwargs: any = {kwargs_json};
|
||||
|
||||
// Get method
|
||||
const method = (targetModule as any).{method_name};
|
||||
if (typeof method !== 'function') {{
|
||||
throw new Error(`Method '{method_name}' does not exist or is not a function`);
|
||||
}}
|
||||
|
||||
// Call method
|
||||
let result: any;
|
||||
if (Object.keys(kwargs).length > 0) {{
|
||||
// If there are keyword arguments, pass them as the last parameter
|
||||
result = await method(...args, kwargs);
|
||||
}} else {{
|
||||
// Only positional arguments
|
||||
result = await method(...args);
|
||||
}}
|
||||
|
||||
// Output result
|
||||
console.log(JSON.stringify(result));
|
||||
}} catch (error: any) {{
|
||||
console.error(JSON.stringify({{
|
||||
error: (error as Error).message,
|
||||
stack: (error as Error).stack
|
||||
}}));
|
||||
process.exit(1);
|
||||
}}
|
||||
}}
|
||||
|
||||
executeMethod();
|
||||
"""
|
||||
return wrapper_code
|
||||
|
||||
|
||||
# Convenience function
|
||||
def call_ts_method(
|
||||
ts_file: str, method_name: str, *args, working_dir: Optional[str] = None, **kwargs
|
||||
) -> Any:
|
||||
"""Convenience function: Call TypeScript method
|
||||
|
||||
Args:
|
||||
ts_file: TypeScript file path
|
||||
method_name: Method name
|
||||
*args: Positional arguments
|
||||
working_dir: Working directory
|
||||
**kwargs: Keyword arguments
|
||||
|
||||
Returns:
|
||||
Execution result
|
||||
"""
|
||||
executor = TypeScriptExecutor(working_dir)
|
||||
return executor.call_method(ts_file, method_name, *args, **kwargs)
|
||||
|
||||
|
||||
# Usage example
|
||||
if __name__ == "__main__":
|
||||
|
||||
# Create test TypeScript file
|
||||
test_ts_content = """
|
||||
export function add(a: number, b: number): number {
|
||||
return a + b;
|
||||
}
|
||||
|
||||
export function greet(name: string, options?: { formal?: boolean }): string {
|
||||
const greeting = options?.formal ? "Hello" : "Hi";
|
||||
return `${greeting}, ${name}!`;
|
||||
}
|
||||
|
||||
export async function processData(data: any[]): Promise<{ count: number; items: any[] }> {
|
||||
// Simulate async processing
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
|
||||
return {
|
||||
count: data.length,
|
||||
items: data.map(item => ({ processed: true, original: item }))
|
||||
};
|
||||
}
|
||||
|
||||
export function complexFunction(
|
||||
numbers: number[],
|
||||
config: { multiplier: number; offset: number }
|
||||
): { result: number[]; sum: number } {
|
||||
const result = numbers.map(n => n * config.multiplier + config.offset);
|
||||
const sum = result.reduce((a, b) => a + b, 0);
|
||||
|
||||
return { result, sum };
|
||||
}
|
||||
"""
|
||||
|
||||
# Write test file
|
||||
with open("test_methods.ts", "w") as f:
|
||||
f.write(test_ts_content)
|
||||
|
||||
try:
|
||||
# Create executor
|
||||
executor = TypeScriptExecutor()
|
||||
|
||||
print("=== TypeScript Method Execution Test ===")
|
||||
|
||||
# Test 1: Simple function
|
||||
print("\n1. Testing simple addition function:")
|
||||
result = executor.call_method("test_methods.ts", "add", 10, 20)
|
||||
print(f" add(10, 20) = {result}")
|
||||
|
||||
# Test 2: Function with optional parameters
|
||||
print("\n2. Testing greeting function:")
|
||||
result1 = executor.call_method("test_methods.ts", "greet", "Alice")
|
||||
print(f" greet('Alice') = {result1}")
|
||||
|
||||
result2 = executor.call_method(
|
||||
"test_methods.ts", "greet", "Bob", {"formal": True}
|
||||
)
|
||||
print(f" greet('Bob', {{formal: true}}) = {result2}")
|
||||
|
||||
# Test 3: Async function
|
||||
print("\n3. Testing async function:")
|
||||
result = executor.call_method(
|
||||
"test_methods.ts", "processData", [1, 2, 3, "hello"]
|
||||
)
|
||||
print(f" processData([1, 2, 3, 'hello']) = {result}")
|
||||
|
||||
# Test 4: Complex function
|
||||
print("\n4. Testing complex function:")
|
||||
result = executor.call_method(
|
||||
"test_methods.ts",
|
||||
"complexFunction",
|
||||
[1, 2, 3, 4, 5],
|
||||
{"multiplier": 2, "offset": 1},
|
||||
)
|
||||
print(f" complexFunction([1,2,3,4,5], {{multiplier:2, offset:1}}) = {result}")
|
||||
|
||||
# Test 5: Using convenience function
|
||||
print("\n5. Testing convenience function:")
|
||||
result = call_ts_method("test_methods.ts", "add", 100, 200)
|
||||
print(f" call_ts_method('test_methods.ts', 'add', 100, 200) = {result}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
|
||||
finally:
|
||||
# Clean up test file
|
||||
if os.path.exists("test_methods.ts"):
|
||||
os.remove("test_methods.ts")
|
||||
Reference in New Issue
Block a user