py_lua_helper: extract exported data from binary files created by loader.lua

Author: DarkCaster

python_lua_helper/py_lua_helper.py

@@ -67,7 +67,6 @@ def __init__(
         # Initialize internal state
         self._variables: Dict[str, str] = {}
         self._metadata: Dict[str, str] = {}
-        self._export_list: List[str] = []
 
         # Initialize temporary directory
         self._setup_temp_dir()
@@ -151,11 +150,10 @@ def _setup_temp_dir(self):
                 self.temp_dir = tempfile.mkdtemp(
                     prefix="lua-helper-", dir=self.temp_dir
                 )
-        # Create data storage directories in selected temp dir
-        self.meta_dir = os.path.join(self.temp_dir, "meta")
-        self.data_dir = os.path.join(self.temp_dir, "data")
-        os.makedirs(self.meta_dir)
-        os.makedirs(self.data_dir)
+        # Define temp files for data exchange
+        self.meta_file = os.path.join(self.temp_dir, "meta.tmp")
+        self.data_file = os.path.join(self.temp_dir, "data.tmp")
+        self.index_file = os.path.join(self.temp_dir, "index.tmp")
 
     def _detect_lua_binary(self):
         """Detect appropriate Lua binary based on version requirements."""
@@ -283,28 +281,56 @@ def _run_lua_loader(self):
         except subprocess.TimeoutExpired:
             raise RuntimeError("Lua loader timed out")
 
-    def _parse_results(self):
-        """Parse exported variables from temporary files."""
-        # Read the list of exported variables
-        try:
-            self._export_list = os.listdir(self.data_dir)
-        except Exception:
-            self._export_list = []
+    def _parse_text_fields(self, file_path: str):
+        """
+        Parses a binary file containing UTF-8 text fields generated by loader.lua
 
-        # Load each variable
-        for filename in self._export_list:
-            try:
-                # Read data
-                with open(os.path.join(self.data_dir, filename), "rb") as f:
-                    self._variables[filename] = f.read().decode(
-                        "utf-8", errors="ignore"
+        Args:
+            file_path (str): The path to the binary file.
+
+        Returns:
+            list: A list of strings containing the parsed UTF-8 fields.
+        """
+        fields = []
+        with open(file_path, "rb") as f:
+            file_content = f.read()
+        # Locate the start marker, set initial position
+        header_marker = b"start\xff\x00"
+        start_pos = file_content.find(header_marker)
+        if start_pos == -1:
+            raise ValueError(f"Failed to locate start marker at {file_path}")
+        current_pos = start_pos + len(header_marker)
+        # Read UTF8 text fields, separated with special sequence
+        separator = b"\xff\x00"
+        while True:
+            # Find the next separator
+            sep_pos = file_content.find(separator, current_pos)
+            if sep_pos != -1:
+                # Extract the bytes for the current field
+                field_bytes = file_content[current_pos:sep_pos]
+                field_str = field_bytes.decode("utf-8")
+                fields.append(field_str)
+                # Move pointer past the separator
+                current_pos = sep_pos + len(separator)
+            else:
+                # No separator found, this is the last field (no terminator at the end)
+                if current_pos < len(file_content):
+                    raise ValueError(
+                        f"Surplus data at the end of file, at pos: {current_pos}"
                     )
-                # Read metadata
-                with open(os.path.join(self.meta_dir, filename), "rb") as f:
-                    self._metadata[filename] = f.read().decode("utf-8", errors="ignore")
-            except Exception:
-                # Skip problematic files
-                continue
+                break
+        return fields
+
+    def _parse_results(self):
+        """Parse exported variables from temporary files."""
+        # Read exported data
+        exports = self._parse_text_fields(self.index_file)
+        meta = self._parse_text_fields(self.meta_file)
+        data = self._parse_text_fields(self.data_file)
+
+        for index, value in enumerate(exports):
+            self._variables[value] = data[index]
+            self._metadata[value] = meta[index]
 
     def _cleanup(self):
         """Clean up temporary directory."""