feat(csv): add opt-in skip for already-translated batches

- Skip candidate batches when every target is translated
- Keep mixed batches and use global BATCHSIZE for checks
- Add CSV setting checkbox and unit coverage
This commit is contained in:
DazedAnon 2026-07-21 16:12:20 -05:00
parent 825bd3a354
commit 65a7ee51c2
3 changed files with 288 additions and 33 deletions

View file

@ -55,6 +55,7 @@ class CSVTab(QWidget):
# Row settings
"SKIP_HEADER_ROW": True,
"USE_TARGET_IF_NOT_EMPTY": False,
"SKIP_IF_TARGET_TRANSLATED": False,
# Output settings
"WRITE_TO_NEXT_COLUMN": False,
@ -188,6 +189,14 @@ class CSVTab(QWidget):
self.use_target_if_not_empty_cb = QCheckBox("Use Target if Not Empty")
self.use_target_if_not_empty_cb.setToolTip("If target column already has text, use that instead of source (T++ style)")
left_column.addWidget(self.use_target_if_not_empty_cb)
self.skip_if_target_translated_cb = QCheckBox("Skip if Target Translated")
self.skip_if_target_translated_cb.setToolTip(
"Check candidate rows in batches matching the global Batch Size setting. "
"Skip a batch only if every target is already translated (non-empty, no Japanese). "
"If any row still needs work, the whole batch is translated."
)
left_column.addWidget(self.skip_if_target_translated_cb)
left_column.addStretch()
@ -269,6 +278,7 @@ class CSVTab(QWidget):
self.speaker_column_spin.setValue(0) # 0 = None
self.skip_header_cb.setChecked(True)
self.use_target_if_not_empty_cb.setChecked(True)
self.skip_if_target_translated_cb.setChecked(False)
self.write_next_column_cb.setChecked(False)
self.parse_name_tags_cb.setChecked(False)
self.parse_m_markers_cb.setChecked(False)
@ -285,6 +295,7 @@ class CSVTab(QWidget):
self.speaker_column_spin.setValue(0) # 0 = None
self.skip_header_cb.setChecked(False)
self.use_target_if_not_empty_cb.setChecked(False)
self.skip_if_target_translated_cb.setChecked(False)
self.write_next_column_cb.setChecked(False)
self.parse_name_tags_cb.setChecked(False)
self.parse_m_markers_cb.setChecked(False)
@ -301,6 +312,7 @@ class CSVTab(QWidget):
self.speaker_column_spin.setValue(3) # Display as 1-based (was 2, 0=None so 3=col2)
self.skip_header_cb.setChecked(False)
self.use_target_if_not_empty_cb.setChecked(False)
self.skip_if_target_translated_cb.setChecked(False)
self.write_next_column_cb.setChecked(False)
self.parse_name_tags_cb.setChecked(False)
self.parse_m_markers_cb.setChecked(False)
@ -334,6 +346,7 @@ class CSVTab(QWidget):
self.csv_delimiter_combo.currentIndexChanged.disconnect()
self.skip_header_cb.stateChanged.disconnect()
self.use_target_if_not_empty_cb.stateChanged.disconnect()
self.skip_if_target_translated_cb.stateChanged.disconnect()
self.write_next_column_cb.stateChanged.disconnect()
self.parse_name_tags_cb.stateChanged.disconnect()
self.parse_m_markers_cb.stateChanged.disconnect()
@ -350,6 +363,7 @@ class CSVTab(QWidget):
self.csv_delimiter_combo.currentIndexChanged.connect(lambda: self.apply_to_module(show_messages=False))
self.skip_header_cb.stateChanged.connect(lambda: self.apply_to_module(show_messages=False))
self.use_target_if_not_empty_cb.stateChanged.connect(lambda: self.apply_to_module(show_messages=False))
self.skip_if_target_translated_cb.stateChanged.connect(lambda: self.apply_to_module(show_messages=False))
self.write_next_column_cb.stateChanged.connect(lambda: self.apply_to_module(show_messages=False))
self.parse_name_tags_cb.stateChanged.connect(lambda: self.apply_to_module(show_messages=False))
self.parse_m_markers_cb.stateChanged.connect(lambda: self.apply_to_module(show_messages=False))
@ -376,6 +390,7 @@ class CSVTab(QWidget):
"CSV_DELIMITER": delimiter,
"SKIP_HEADER_ROW": self.skip_header_cb.isChecked(),
"USE_TARGET_IF_NOT_EMPTY": self.use_target_if_not_empty_cb.isChecked(),
"SKIP_IF_TARGET_TRANSLATED": self.skip_if_target_translated_cb.isChecked(),
"WRITE_TO_NEXT_COLUMN": self.write_next_column_cb.isChecked(),
"PARSE_NAME_TAGS": self.parse_name_tags_cb.isChecked(),
"PARSE_M_MARKERS": self.parse_m_markers_cb.isChecked(),
@ -406,6 +421,7 @@ class CSVTab(QWidget):
self.skip_header_cb.setChecked(config.get("SKIP_HEADER_ROW", True))
self.use_target_if_not_empty_cb.setChecked(config.get("USE_TARGET_IF_NOT_EMPTY", False))
self.skip_if_target_translated_cb.setChecked(config.get("SKIP_IF_TARGET_TRANSLATED", False))
self.write_next_column_cb.setChecked(config.get("WRITE_TO_NEXT_COLUMN", False))
self.parse_name_tags_cb.setChecked(config.get("PARSE_NAME_TAGS", False))
self.parse_m_markers_cb.setChecked(config.get("PARSE_M_MARKERS", False))
@ -441,6 +457,7 @@ class CSVTab(QWidget):
bool_patterns = {
"SKIP_HEADER_ROW": r'^SKIP_HEADER_ROW\s*=\s*(True|False)',
"USE_TARGET_IF_NOT_EMPTY": r'^USE_TARGET_IF_NOT_EMPTY\s*=\s*(True|False)',
"SKIP_IF_TARGET_TRANSLATED": r'^SKIP_IF_TARGET_TRANSLATED\s*=\s*(True|False)',
"WRITE_TO_NEXT_COLUMN": r'^WRITE_TO_NEXT_COLUMN\s*=\s*(True|False)',
"PARSE_NAME_TAGS": r'^PARSE_NAME_TAGS\s*=\s*(True|False)',
"PARSE_M_MARKERS": r'^PARSE_M_MARKERS\s*=\s*(True|False)',

View file

@ -48,6 +48,7 @@ TARGET_COLUMN = 3 # Which column to write translations to
SPEAKER_COLUMN = 1 # Which column has speaker names (-1 = none)
SKIP_HEADER_ROW = False # Skip the first row (header)
USE_TARGET_IF_NOT_EMPTY = False # Use target column text if not empty (T++ style)
SKIP_IF_TARGET_TRANSLATED = False # Skip batches whose targets are already translated
WRITE_TO_NEXT_COLUMN = False # Write to column after target instead of overwriting
PARSE_NAME_TAGS = False # Parse :name[] tags in text
PARSE_M_MARKERS = False # Parse \M markers in text
@ -215,6 +216,64 @@ def flush_progress_csv(writeFile, writer, rows):
traceback.print_exc()
def _actual_target_column():
"""Column index translations are written to."""
return TARGET_COLUMN + 1 if WRITE_TO_NEXT_COLUMN else TARGET_COLUMN
def _target_is_translated(row):
"""True when the write-target cell is non-empty and has no Japanese left."""
actual_target = _actual_target_column()
if len(row) <= actual_target:
return False
target = row[actual_target]
if not isinstance(target, str) or not target.strip():
return False
return not re.search(LANGREGEX, target)
def _row_source_text(row):
"""Source text for a row, honoring USE_TARGET_IF_NOT_EMPTY."""
if USE_TARGET_IF_NOT_EMPTY and len(row) > TARGET_COLUMN and row[TARGET_COLUMN]:
return row[TARGET_COLUMN]
if len(row) > SOURCE_COLUMN and row[SOURCE_COLUMN]:
return row[SOURCE_COLUMN]
return ""
def _is_candidate_row(data, i):
"""True when this row would normally be collected for translation."""
if SKIP_HEADER_ROW and i == 0:
return False
if SKIP_COMMENT_ROWS and len(data[i]) > 0 and "comment" in str(data[i][0]).lower():
return False
if len(data[i]) <= SOURCE_COLUMN:
return False
return bool(_row_source_text(data[i]))
def _collect_process_indices(data):
"""Row indices to translate.
When SKIP_IF_TARGET_TRANSLATED is on, candidates are checked in groups of
BATCHSIZE (same as the translation batch size). A group is skipped only if
every target in it is already translated; if any row still needs work, the
whole group runs.
"""
candidates = [i for i in range(len(data)) if _is_candidate_row(data, i)]
if not SKIP_IF_TARGET_TRANSLATED:
return candidates
process = []
batch_size = max(1, int(BATCHSIZE))
for start in range(0, len(candidates), batch_size):
batch = candidates[start:start + batch_size]
if all(_target_is_translated(data[i]) for i in batch):
continue
process.extend(batch)
return process
def translateCSV(data, pbar, writeFile, writer, filename, translatedList):
"""
Unified CSV translation function using configurable settings.
@ -225,6 +284,7 @@ def translateCSV(data, pbar, writeFile, writer, filename, translatedList):
- SPEAKER_COLUMN: column index for speaker names (-1 = none)
- SKIP_HEADER_ROW: whether to skip first row
- USE_TARGET_IF_NOT_EMPTY: use existing target text if present (T++ style)
- SKIP_IF_TARGET_TRANSLATED: skip fully-translated candidate batches
- WRITE_TO_NEXT_COLUMN: write to column after target
- PARSE_NAME_TAGS: parse :name[] tags
- PARSE_M_MARKERS: parse \\M markers
@ -235,42 +295,16 @@ def translateCSV(data, pbar, writeFile, writer, filename, translatedList):
PBAR = pbar
translatedText = ""
totalTokens = [0, 0]
i = 0
stringList = []
process_indices = _collect_process_indices(data)
try:
# Translate
while i < len(data):
# Skip header row if configured
if SKIP_HEADER_ROW and i == 0:
i += 1
continue
# Skip comment rows if configured
if SKIP_COMMENT_ROWS and len(data[i]) > 0 and 'comment' in str(data[i][0]).lower():
i += 1
continue
# Check if row has enough columns
if len(data[i]) <= SOURCE_COLUMN:
i += 1
continue
for i in process_indices:
# Get source text
jaString = ""
jaString = _row_source_text(data[i])
speaker = ""
# If USE_TARGET_IF_NOT_EMPTY is enabled (T++ style), check target column first
if USE_TARGET_IF_NOT_EMPTY and len(data[i]) > TARGET_COLUMN and data[i][TARGET_COLUMN]:
jaString = data[i][TARGET_COLUMN]
else:
jaString = data[i][SOURCE_COLUMN] if data[i][SOURCE_COLUMN] else ""
# Skip empty strings
if not jaString:
i += 1
continue
# Handle speaker column if configured
if SPEAKER_COLUMN >= 0 and len(data[i]) > SPEAKER_COLUMN and data[i][SPEAKER_COLUMN]:
speakerResponse = getSpeaker(data[i][SPEAKER_COLUMN])
@ -338,7 +372,7 @@ def translateCSV(data, pbar, writeFile, writer, filename, translatedList):
translatedText = translatedText.replace("\n", "\\n")
# Determine target column
actual_target = TARGET_COLUMN + 1 if WRITE_TO_NEXT_COLUMN else TARGET_COLUMN
actual_target = _actual_target_column()
# Ensure row has enough columns
while len(data[i]) <= actual_target:
@ -352,9 +386,6 @@ def translateCSV(data, pbar, writeFile, writer, filename, translatedList):
data[i][actual_target] = translatedText
flush_progress_csv(writeFile, writer, data)
# Iterate
i += 1
# EOF - Process collected strings
if len(stringList) > 0:

View file

@ -0,0 +1,207 @@
import unittest
from unittest.mock import MagicMock, patch
import modules.csv as csv_mod
class TargetTranslatedDetectionTests(unittest.TestCase):
def setUp(self):
self._orig = {
"SOURCE_COLUMN": csv_mod.SOURCE_COLUMN,
"TARGET_COLUMN": csv_mod.TARGET_COLUMN,
"WRITE_TO_NEXT_COLUMN": csv_mod.WRITE_TO_NEXT_COLUMN,
"USE_TARGET_IF_NOT_EMPTY": csv_mod.USE_TARGET_IF_NOT_EMPTY,
"SKIP_HEADER_ROW": csv_mod.SKIP_HEADER_ROW,
"SKIP_COMMENT_ROWS": csv_mod.SKIP_COMMENT_ROWS,
"SKIP_IF_TARGET_TRANSLATED": csv_mod.SKIP_IF_TARGET_TRANSLATED,
"BATCHSIZE": csv_mod.BATCHSIZE,
}
csv_mod.SOURCE_COLUMN = 0
csv_mod.TARGET_COLUMN = 1
csv_mod.WRITE_TO_NEXT_COLUMN = False
csv_mod.USE_TARGET_IF_NOT_EMPTY = False
csv_mod.SKIP_HEADER_ROW = False
csv_mod.SKIP_COMMENT_ROWS = False
def tearDown(self):
for key, value in self._orig.items():
setattr(csv_mod, key, value)
def test_empty_target_not_translated(self):
self.assertFalse(csv_mod._target_is_translated(["こんにちは", ""]))
def test_missing_target_column_not_translated(self):
self.assertFalse(csv_mod._target_is_translated(["こんにちは"]))
def test_english_target_is_translated(self):
self.assertTrue(csv_mod._target_is_translated(["こんにちは", "Hello"]))
def test_japanese_target_not_translated(self):
self.assertFalse(csv_mod._target_is_translated(["こんにちは", "こんにちは"]))
def test_write_to_next_column_checks_next(self):
csv_mod.WRITE_TO_NEXT_COLUMN = True
self.assertFalse(csv_mod._target_is_translated(["こんにちは", "Hello", ""]))
self.assertTrue(csv_mod._target_is_translated(["こんにちは", "Hello", "Hello there"]))
class SkipTranslatedBatchIndexTests(unittest.TestCase):
def setUp(self):
self._orig = {
"SOURCE_COLUMN": csv_mod.SOURCE_COLUMN,
"TARGET_COLUMN": csv_mod.TARGET_COLUMN,
"WRITE_TO_NEXT_COLUMN": csv_mod.WRITE_TO_NEXT_COLUMN,
"USE_TARGET_IF_NOT_EMPTY": csv_mod.USE_TARGET_IF_NOT_EMPTY,
"SKIP_HEADER_ROW": csv_mod.SKIP_HEADER_ROW,
"SKIP_COMMENT_ROWS": csv_mod.SKIP_COMMENT_ROWS,
"SKIP_IF_TARGET_TRANSLATED": csv_mod.SKIP_IF_TARGET_TRANSLATED,
"BATCHSIZE": csv_mod.BATCHSIZE,
}
csv_mod.SOURCE_COLUMN = 0
csv_mod.TARGET_COLUMN = 1
csv_mod.WRITE_TO_NEXT_COLUMN = False
csv_mod.USE_TARGET_IF_NOT_EMPTY = False
csv_mod.SKIP_HEADER_ROW = False
csv_mod.SKIP_COMMENT_ROWS = False
csv_mod.BATCHSIZE = 30
def tearDown(self):
for key, value in self._orig.items():
setattr(csv_mod, key, value)
def test_default_off_keeps_all_candidates(self):
csv_mod.SKIP_IF_TARGET_TRANSLATED = False
data = [
["こんにちは", "Hello"],
["さようなら", ""],
]
self.assertEqual(csv_mod._collect_process_indices(data), [0, 1])
def test_fully_translated_batch_is_skipped(self):
csv_mod.SKIP_IF_TARGET_TRANSLATED = True
csv_mod.BATCHSIZE = 2
data = [
["", "A"],
["", "I"],
["", ""],
["", ""],
]
# First batch (0,1) all translated -> skip; second batch (2,3) needs work
self.assertEqual(csv_mod._collect_process_indices(data), [2, 3])
def test_partial_batch_keeps_entire_batch(self):
csv_mod.SKIP_IF_TARGET_TRANSLATED = True
csv_mod.BATCHSIZE = 3
data = [
["", "A"],
["", ""],
["", "U"],
["", "E"],
["", "O"],
["", "Ka"],
]
# Batch [0,1,2] has untranslated row 1 -> keep all three
# Batch [3,4,5] all translated -> skip
self.assertEqual(csv_mod._collect_process_indices(data), [0, 1, 2])
def test_uses_global_batch_size(self):
csv_mod.SKIP_IF_TARGET_TRANSLATED = True
csv_mod.BATCHSIZE = 30
data = [["" + str(i), f"Line {i}"] for i in range(30)]
data.append(["未翻訳", ""])
# First 30 fully translated -> skipped; last singleton batch needs work
self.assertEqual(csv_mod._collect_process_indices(data), [30])
class SkipIfTargetTranslatedCollectTests(unittest.TestCase):
def setUp(self):
self._orig = {
"SOURCE_COLUMN": csv_mod.SOURCE_COLUMN,
"TARGET_COLUMN": csv_mod.TARGET_COLUMN,
"SPEAKER_COLUMN": csv_mod.SPEAKER_COLUMN,
"SKIP_HEADER_ROW": csv_mod.SKIP_HEADER_ROW,
"USE_TARGET_IF_NOT_EMPTY": csv_mod.USE_TARGET_IF_NOT_EMPTY,
"SKIP_IF_TARGET_TRANSLATED": csv_mod.SKIP_IF_TARGET_TRANSLATED,
"BATCHSIZE": csv_mod.BATCHSIZE,
"WRITE_TO_NEXT_COLUMN": csv_mod.WRITE_TO_NEXT_COLUMN,
"PARSE_NAME_TAGS": csv_mod.PARSE_NAME_TAGS,
"PARSE_M_MARKERS": csv_mod.PARSE_M_MARKERS,
"REMOVE_FURIGANA": csv_mod.REMOVE_FURIGANA,
"SKIP_COMMENT_ROWS": csv_mod.SKIP_COMMENT_ROWS,
"ESTIMATE": csv_mod.ESTIMATE,
}
csv_mod.SOURCE_COLUMN = 0
csv_mod.TARGET_COLUMN = 1
csv_mod.SPEAKER_COLUMN = -1
csv_mod.SKIP_HEADER_ROW = False
csv_mod.USE_TARGET_IF_NOT_EMPTY = False
csv_mod.WRITE_TO_NEXT_COLUMN = False
csv_mod.PARSE_NAME_TAGS = False
csv_mod.PARSE_M_MARKERS = False
csv_mod.REMOVE_FURIGANA = False
csv_mod.SKIP_COMMENT_ROWS = False
csv_mod.ESTIMATE = True
csv_mod.BATCHSIZE = 30
def tearDown(self):
for key, value in self._orig.items():
setattr(csv_mod, key, value)
def test_default_off_does_not_skip_translated_targets(self):
csv_mod.SKIP_IF_TARGET_TRANSLATED = False
data = [
["こんにちは", "Hello"],
["さようなら", ""],
]
pbar = MagicMock()
with patch.object(csv_mod, "translateAI", return_value=(["Hello", "Goodbye"], [0, 0])) as mock_ai:
with patch.object(csv_mod, "dazedwrap") as mock_wrap:
mock_wrap.wrapText.side_effect = lambda text, _width: text
csv_mod.translateCSV(data, pbar, None, MagicMock(), "test.csv", None)
self.assertEqual(mock_ai.call_args[0][0], ["こんにちは", "さようなら"])
def test_opt_in_skips_only_fully_translated_batches(self):
csv_mod.SKIP_IF_TARGET_TRANSLATED = True
csv_mod.BATCHSIZE = 2
data = [
["こんにちは", "Hello"],
["おはよう", "Good morning"],
["さようなら", ""],
["ありがとう", "ありがとう"],
]
pbar = MagicMock()
with patch.object(
csv_mod, "translateAI", return_value=(["Goodbye", "Thank you"], [0, 0])
) as mock_ai:
with patch.object(csv_mod, "dazedwrap") as mock_wrap:
mock_wrap.wrapText.side_effect = lambda text, _width: text
csv_mod.translateCSV(data, pbar, None, MagicMock(), "test.csv", None)
# First batch fully translated -> skipped; second batch kept whole
self.assertEqual(mock_ai.call_args[0][0], ["さようなら", "ありがとう"])
self.assertEqual(data[0][1], "Hello")
self.assertEqual(data[1][1], "Good morning")
self.assertEqual(data[2][1], "Goodbye")
self.assertEqual(data[3][1], "Thank you")
def test_partial_batch_retranslates_already_done_rows(self):
csv_mod.SKIP_IF_TARGET_TRANSLATED = True
csv_mod.BATCHSIZE = 2
data = [
["こんにちは", "Hello"],
["さようなら", ""],
]
pbar = MagicMock()
with patch.object(
csv_mod, "translateAI", return_value=(["Hi again", "Goodbye"], [0, 0])
) as mock_ai:
with patch.object(csv_mod, "dazedwrap") as mock_wrap:
mock_wrap.wrapText.side_effect = lambda text, _width: text
csv_mod.translateCSV(data, pbar, None, MagicMock(), "test.csv", None)
# Mixed batch: keep both rows, including the already-translated one
self.assertEqual(mock_ai.call_args[0][0], ["こんにちは", "さようなら"])
self.assertEqual(data[0][1], "Hi again")
self.assertEqual(data[1][1], "Goodbye")
if __name__ == "__main__":
unittest.main()