From 50478790731e52413023979d9ab56decf4bb98d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ant=C3=98n=20Ustin=C3=98ff?= Date: Mon, 13 Jul 2026 12:05:09 +0400 Subject: [PATCH 1/3] feat: batch sheet updates and add local localize task --- CHANGELOG.md | 5 + bin/localize.dart | 259 ++++++++++++++++++++++++++-------------------- pubspec.yaml | 2 +- 3 files changed, 153 insertions(+), 113 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1906dda..7da1698 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +## 0.4.3 + +- **CHANGED**: `localize` now writes translated cells to Google Sheets via batch updates per row instead of one request per cell. +- **CHANGED**: `--workers` now controls real concurrent OpenAI requests during localization. + ## 0.4.2 - **ADDED**: Configurable language batch size via `--batch` (`-b`) CLI parameter to control how many languages are translated per single API call (default: 3). diff --git a/bin/localize.dart b/bin/localize.dart index 0eccd3c..532ff1c 100644 --- a/bin/localize.dart +++ b/bin/localize.dart @@ -112,7 +112,8 @@ void main(List? $arguments) => runZonedGuarded( // Create Google Sheets API client $log('Creating Google Sheets API client...'); - final sheetsApi = await createSheetsApiClient( + final (api: sheetsApi, client: sheetsClient) = + await createSheetsApiClient( credentialsPath, ); @@ -125,6 +126,7 @@ void main(List? $arguments) => runZonedGuarded( ).toList(); if (sheets.isEmpty) { + sheetsClient.close(); $err('No valid sheets found to process.'); io.exit(1); } @@ -139,29 +141,37 @@ void main(List? $arguments) => runZonedGuarded( // Process each sheet $log('Processing ${sheets.length} sheets...'); - for (final (:sheet, :values) in sheets) { - final title = sheet.properties?.title ?? 'Unknown'; - $log('Processing sheet: $title'); - final rows = await extractEmptyCells( - sheet: sheet, - values: values, - ); - if (rows.isEmpty) continue; - - $log('Found ${rows.length} rows to localize in sheet: $title'); - await for (final row in localizeRows( - rows: rows, - client: client, - cellsPerBatch: batch.clamp(1, 20), - )) { - if (row.isEmpty) continue; - await updateSheet( - api: sheetsApi, - sheetId: sheetId, - sheetTitle: title, - row: row, + try { + for (final (:sheet, :values) in sheets) { + final title = sheet.properties?.title ?? 'Unknown'; + $log('Processing sheet: $title'); + final rows = await extractEmptyCells( + sheet: sheet, + values: values, ); + if (rows.isEmpty) continue; + + $log('Found ${rows.length} rows to localize in sheet: $title'); + await for (final row in localizeRows( + rows: rows, + client: client, + cellsPerBatch: batch.clamp(1, 20), + )) { + if (row.isEmpty) continue; + await updateSheet( + api: sheetsApi, + sheetId: sheetId, + sheetTitle: title, + row: row, + ); + } } + $log('Localization completed successfully ' + 'for ${sheets.length} sheets.'); + } finally { + // Close the auth client so its token-refresh timer and keep-alive + // sockets are released and the process can exit. + sheetsClient.close(); } }, (error, stackTrace) { @@ -326,7 +336,8 @@ Usage: dart run bin/localize.dart [options] '''; /// Create Google Sheets API client -Future createSheetsApiClient( +Future<({SheetsApi api, AutoRefreshingAuthClient client})> + createSheetsApiClient( String credentialsPath, ) async { $log('Credentials path: $credentialsPath'); @@ -351,18 +362,16 @@ Future createSheetsApiClient( } $log('Creating Google Sheets API client...'); - SheetsApi sheetsApi; try { final client = await clientViaServiceAccount(credentials, [ // SheetsApi.spreadsheetsReadonlyScope, SheetsApi.spreadsheetsScope, ]); - sheetsApi = SheetsApi(client); + return (api: SheetsApi(client), client: client); } on Object catch (e) { $err('Error creating Google Sheets API client: $e'); io.exit(1); } - return sheetsApi; } /// Fetch spreadsheets from Google Sheets API @@ -702,13 +711,16 @@ Stream localizeRows({ required List rows, required OpenAIClient client, int cellsPerBatch = 3, -}) async* { - for (final row in rows) { +}) { + // Localize a single row: split its target locales into batches and call the + // OpenAI API for each batch. Batches within a row run sequentially, but many + // rows are dispatched concurrently below — the client's internal semaphore + // caps the number of simultaneous requests at [OpenAIClient.workers]. + Future localizeOne(LocalizeRow row) async { final cells = row.cells.map((e) => e.code).toList(growable: false); - for (var i = 0; i <= cells.length; i += cellsPerBatch) { + for (var i = 0; i < cells.length; i += cellsPerBatch) { try { // Get the next batch of languages to process - if (i >= cells.length) break; final languages = cells.skip(i).take(cellsPerBatch).toList(growable: false); if (languages.isEmpty) break; @@ -751,8 +763,24 @@ Stream localizeRows({ continue; } } - yield row; } + + // Dispatch every row concurrently and emit each one as soon as it finishes. + final controller = StreamController(); + if (rows.isEmpty) { + controller.close(); + return controller.stream; + } + + var pending = rows.length; + for (final row in rows) { + localizeOne(row).whenComplete(() { + controller.add(row); + if (--pending == 0) controller.close(); + }); + } + + return controller.stream; } class OpenAIClient { @@ -762,7 +790,7 @@ class OpenAIClient { this.workers = 6, this.retries = 3, this.systemPrompt, - }); + }) : _available = workers < 1 ? 1 : workers; final String apiKey; final String model; @@ -770,11 +798,31 @@ class OpenAIClient { final int retries; final String? systemPrompt; - final Queue Function()> _taskQueue = - Queue Function()>(); + /// Counting semaphore limiting the number of concurrent in-flight + /// OpenAI requests to [workers]. + int _available; + final Queue> _waiters = Queue>(); + + /// Acquire a slot before performing a request, waiting if [workers] requests + /// are already in flight. + Future _acquire() { + if (_available > 0) { + _available--; + return Future.value(); + } + final completer = Completer(); + _waiters.add(completer); + return completer.future; + } - bool _isProcessing = false; - bool get isProcessing => _isProcessing; + /// Release a slot, waking the next waiter if any. + void _release() { + if (_waiters.isNotEmpty) { + _waiters.removeFirst().complete(); + } else { + _available++; + } + } Future<({String label, Map localization})> _request({ required io.HttpClient client, @@ -862,61 +910,37 @@ class OpenAIClient { throw Exception('Invalid response format from OpenAI API'); } - void _processQueue() { - if (_isProcessing) return; - if (_taskQueue.isEmpty) return; - _isProcessing = true; - Future(() async { - while (_taskQueue.isNotEmpty) { - try { - final task = _taskQueue.removeFirst(); - await task(); - } on Object { - // Ignore errors in the queue processing - } - } - _isProcessing = false; - }).ignore(); - } - Future<({String label, Map localization})> call({ required String prompt, required Map schema, }) async { - final completer = - Completer<({String label, Map localization})>(); - _taskQueue.add(() async { - io.HttpClient client = io.HttpClient(); - try { - for (var i = 0; i < retries; i++) { - try { - final response = await _request( - client: client, - prompt: prompt, - schema: schema, - ); - if (!completer.isCompleted) completer.complete(response); - return; - } on FormatException catch (e) { - if (i == retries - 1) rethrow; - $err('OpenAI API returned invalid JSON ' - '(attempt ${i + 1}/$retries): $e'); - await Future.delayed(const Duration(milliseconds: 250)); - } on Object catch (e) { - if (i == retries - 1) rethrow; - $err('OpenAI API call failed ' '(attempt ${i + 1}/$retries): $e'); - await Future.delayed(const Duration(milliseconds: 250)); - } + // Throttle to at most [workers] concurrent in-flight requests. + await _acquire(); + final client = io.HttpClient(); + try { + for (var i = 0; i < retries; i++) { + try { + return await _request( + client: client, + prompt: prompt, + schema: schema, + ); + } on FormatException catch (e) { + if (i == retries - 1) rethrow; + $err('OpenAI API returned invalid JSON ' + '(attempt ${i + 1}/$retries): $e'); + await Future.delayed(const Duration(milliseconds: 250)); + } on Object catch (e) { + if (i == retries - 1) rethrow; + $err('OpenAI API call failed ' '(attempt ${i + 1}/$retries): $e'); + await Future.delayed(const Duration(milliseconds: 250)); } - } on Object catch (e, s) { - if (!completer.isCompleted) completer.completeError(e, s); - $err('OpenAI API call failed: $e'); - } finally { - client.close(); } - }); - _processQueue(); - return completer.future; + throw Exception('OpenAI API call failed after $retries attempts'); + } finally { + client.close(); + _release(); + } } } @@ -970,40 +994,51 @@ Future updateSheet({ }) async { if (row.isEmpty) return; + // Collect every non-empty cell into a single batch so the whole row is + // written in ONE API request instead of one request per cell. This keeps us + // well under the Google Sheets write quota (60 requests/min). + final data = []; for (final cell in row.cells) { if (cell.isEmpty) continue; - final text = cell.text; - const attempts = 3; + data.add( + ValueRange( + range: '$sheetTitle!${columnFromIndex(cell.column)}${row.row + 1}', + values: [ + [cell.text] + ], + ), + ); + } + if (data.isEmpty) return; - for (var attempt = 1; attempt <= attempts; attempt++) { - try { - // Wait for rate limiter before making API call - await _sheetsRateLimiter.waitIfNeeded(); - - await api.spreadsheets.values.update( - ValueRange(values: [ - [text] - ]), - sheetId, - '$sheetTitle!${columnFromIndex(cell.column)}${row.row + 1}', + const attempts = 3; + for (var attempt = 1; attempt <= attempts; attempt++) { + try { + // Wait for rate limiter before making API call + await _sheetsRateLimiter.waitIfNeeded(); + + await api.spreadsheets.values.batchUpdate( + BatchUpdateValuesRequest( valueInputOption: 'RAW', - ); - break; // Success, exit retry loop - } on Object catch (e) { - if (attempt == attempts) { - $err( - 'Error updating sheet "$sheetTitle" ' - 'cell [${columnFromIndex(cell.column)}${row.row + 1}]: $e', - ); - rethrow; - } + data: data, + ), + sheetId, + ); + break; // Success, exit retry loop + } on Object catch (e) { + if (attempt == attempts) { $err( - 'Retrying update for sheet "$sheetTitle" ' - 'cell [${columnFromIndex(cell.column)}${row.row + 1}] ' - '(attempt $attempt/$attempts) due to error: $e', + 'Error updating sheet "$sheetTitle" ' + 'row [${row.row + 1}]: $e', ); - await Future.delayed(const Duration(seconds: 30)); + rethrow; } + $err( + 'Retrying update for sheet "$sheetTitle" ' + 'row [${row.row + 1}] ' + '(attempt $attempt/$attempts) due to error: $e', + ); + await Future.delayed(const Duration(seconds: 30)); } } } diff --git a/pubspec.yaml b/pubspec.yaml index ac5b2e0..b073629 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -9,7 +9,7 @@ description: > It uses the Google Sheets API to fetch translations and generates Dart localization files for use in Flutter applications. -version: 0.4.2 +version: 0.4.3 homepage: https://github.com/DoctorinaAI/sheety_localization repository: https://github.com/DoctorinaAI/sheety_localization From ef3ada607b9567a775744c562eefb788ca1f4632 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ant=C3=98n=20Ustin=C3=98ff?= Date: Mon, 13 Jul 2026 13:29:57 +0400 Subject: [PATCH 2/3] fix: stream controller initialization --- bin/localize.dart | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/bin/localize.dart b/bin/localize.dart index 532ff1c..3a82a7a 100644 --- a/bin/localize.dart +++ b/bin/localize.dart @@ -766,11 +766,8 @@ Stream localizeRows({ } // Dispatch every row concurrently and emit each one as soon as it finishes. + if (rows.isEmpty) return const Stream.empty(); final controller = StreamController(); - if (rows.isEmpty) { - controller.close(); - return controller.stream; - } var pending = rows.length; for (final row in rows) { From 916c3e5126fa7642df9244234adf62160b6e2e38 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ant=C3=98n=20Ustin=C3=98ff?= Date: Mon, 13 Jul 2026 13:32:26 +0400 Subject: [PATCH 3/3] fix: stream controller initialization --- bin/localize.dart | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/bin/localize.dart b/bin/localize.dart index 3a82a7a..06b2c0b 100644 --- a/bin/localize.dart +++ b/bin/localize.dart @@ -766,17 +766,21 @@ Stream localizeRows({ } // Dispatch every row concurrently and emit each one as soon as it finishes. - if (rows.isEmpty) return const Stream.empty(); + if (rows.isEmpty) return const Stream.empty(); final controller = StreamController(); - var pending = rows.length; for (final row in rows) { - localizeOne(row).whenComplete(() { - controller.add(row); - if (--pending == 0) controller.close(); + Future(() async { + try { + await localizeOne(row); + } on Object catch (e, _) { + $err('Error localizing row: $e'); + } finally { + pending--; + if (pending == 0) controller.close().ignore(); + } }); } - return controller.stream; }