Forbid to use curly brackets in msg translations, ref BGforgeNet/Fallout2_Restoration_Project#336

Author: burner1024

msg2po/core.py

@@ -31,6 +31,7 @@
             "context": "{{{index}}}{{{context}}}{{{value}}}\n",
             "female": "separate",
         },
+        "forbidden_characters": ["{", "}"],
     },
     "sve": {
         "pattern": r"(\d+):(.*)",
@@ -41,6 +42,7 @@
             "default": "{index}:{value}\n",
             "female": "separate",
         },
+        "forbidden_characters": [],
     },
     "txt": {
         "pattern": r"(\d+):(.*)",
@@ -52,6 +54,7 @@
             "default": "{index}:{value}\n",
             "female": "separate",
         },
+        "forbidden_characters": [],
     },
     "tra": {
         "pattern": r"@(\d+)\s*?=\s*?~([^~]*?)~(?:\s)?(?:\[([^]]*)\])?(?:~([^~]*)~)?",
@@ -65,6 +68,7 @@
             "context": "@{index} = ~{value}~ [{context}]\n",
             "female": "@{index} = ~{value}~ ~{female}~\n",
         },
+        "forbidden_characters": [],
     },
 }
 
@@ -542,6 +546,14 @@ def get_line_format(e, ext: str):
     """
     ff = FILE_FORMAT[ext]
     line_format = ff["line_format"]
+
+    forbidden_characters = ff["forbidden_characters"]
+    for fc in forbidden_characters:
+        if fc in e["value"]:
+            print(f"ERROR: {ext} strings may not contain '{fc}' character")
+            print(f"\t\tentry: {e}")
+            raise ValueError("Invalid translation character")
+
     if e["context"] is not None:  # entry with context
         lfrm = line_format["context"]
     elif (
@@ -789,6 +801,7 @@ def __init__(self, filepath: str, is_source: False, encoding=CONFIG.encoding):
         self.pattern = self.fformat["pattern"]
         self.dotall = self.fformat["dotall"]
         self.filepath = filepath
+        self.forbidden_characters = self.fformat["forbidden_characters"]
 
         try:  # comment for all entries in file
             self.comment = self.fformat["comment"]
@@ -827,6 +840,12 @@ def __init__(self, filepath: str, is_source: False, encoding=CONFIG.encoding):
             entry.occurence = (filepath, str(index))
             entry.value = str(line[self.fformat["value"]])
 
+            for fc in self.forbidden_characters:
+                if fc in entry.value:
+                    print(f"ERROR: {fext} strings may not contain '{fc}' character")
+                    print(f"\t\tentry: {entry}")
+                    raise ValueError("Invalid translation character")
+
             # skip invalid '000' entries in MSG files
             if fext == "msg" and index == "000":
                 print(

msg2po/unpoify.py

@@ -1,12 +1,10 @@
 #!/usr/bin/env python3
-# coding: utf-8
 
 import os
 import argparse
 import sys
-from multiprocessing import Pool
+import concurrent.futures
 from polib import pofile
-from functools import partial
 from msg2po.core import (
     CONFIG,
     LanguageMap,
@@ -37,8 +35,9 @@ def extract_po(pf: str, language_map: LanguageMap):
     pf is po file basename
     """
     po_path = os.path.join(CONFIG.po_dirname, pf)
-    print("processing {}".format(po_path))
-    po = pofile(po_path)  # open once
+    print(f"processing {po_path}")
+    # Open PO once, it's a heavy op
+    po = pofile(po_path)
     trans_map = translation_entries(po)
     female_map = female_entries(po)
 
@@ -47,38 +46,44 @@ def extract_po(pf: str, language_map: LanguageMap):
     for ef in sorted(trans_map):
         enc = get_enc(po_path, ef)
         ef_extract_path = os.path.join(dst_dir, ef)
-        print("Extracting {} from {} into {} with encoding {}".format(ef, po_path, ef_extract_path, enc))
+        print(f"Extracting {ef} from {po_path} into {ef_extract_path} with encoding {enc}")
         po2file(po, ef_extract_path, enc, ef, dst_dir=dst_dir, trans_map=trans_map, female_map=female_map)
 
     enc = get_enc(po_path)
-    print("Extracted {} into {} with encoding {}".format(po_path, dst_dir, enc))
+    print(f"Extracted {po_path} into {dst_dir} with encoding {enc}")
 
 
 def main():
     po_dir = CONFIG.po_dir
     language_map = LanguageMap()
 
-    # find PO files
-    po_files = []
-    for dir_name, subdir_list, file_list in os.walk(po_dir):
-        for f in file_list:
-            if get_ext(f) == "po":
-                po_files.append(f)
-    if po_files == []:
-        print("no PO files found in directory {}".format(po_dir))
+    # Find PO files
+    po_files = [f for _, _, files in os.walk(po_dir) for f in files if get_ext(f) == "po"]
+
+    if not po_files:
+        print(f"no PO files found in directory {po_dir}")
         sys.exit(1)
 
     with cd(CONFIG.tra_dir):
-        # extract PO files
-        pool = Pool()
-        try:
-            r = pool.map_async(partial(extract_po, language_map=language_map), po_files)
-            pool.close()
-            codes = r.get()  # noqa: F841 - need to get results to see if there's an exception
-        except KeyboardInterrupt:
-            pool.terminate()
-        finally:
-            pool.join()
+        with concurrent.futures.ProcessPoolExecutor() as executor:
+            futures = {executor.submit(extract_po, pf, language_map): pf for pf in po_files}
+
+            for future in concurrent.futures.as_completed(futures):
+                pf = futures[future]
+                try:
+                    future.result()
+                except ValueError as e:
+                    print(f"ValueError in file {pf}: {e}")
+                    executor.shutdown(wait=False, cancel_futures=True)
+                    sys.exit(1)
+                except KeyboardInterrupt:
+                    print("Interrupted by user, terminating execution...")
+                    executor.shutdown(wait=False, cancel_futures=True)
+                    sys.exit(1)
+                except Exception as e:
+                    print(f"Unhandled exception in file {pf}: {e}")
+                    executor.shutdown(wait=False, cancel_futures=True)
+                    sys.exit(1)
 
 
 if __name__ == "__main__":