fix(collect-data): 品牌检测查询失败不再写入无效品牌表，避免故障期误杀永久拉黑

线上 task-27265 采集结果 0 行的根因：.env 缺 export 导致 Java 请求
16890 不带 X-Token 全线 422 -> 整批 queryFailed，而 filterByBrandCheck
把 queryFailed 与 rejected 一并写入 biz_invalid_asin_data，故障品牌被
永久拉黑，后续任务同品牌商品全部被 invalidFiltered 误杀（9-01 起所有
采集任务均 0 行）。改为仅真正判定无效的 rejected 才写表，rejected 为空
不触发写入；新增 2 个单测固化契约。

backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java

@@ -803,10 +803,15 @@ public class CollectDataService {
         CollectDataBrandBatchFilter.BrandBatchOutcome outcome = brandBatchFilter.filter(rows);
         stats.brandRejectedCount += outcome.rejected().size();
         stats.brandQueryFailedCount += outcome.queryFailed().size();
-        List<CollectDataResultRowVo> invalidRows = new ArrayList<>(outcome.rejected().size() + outcome.queryFailed().size());
-        invalidRows.addAll(outcome.rejected());
-        invalidRows.addAll(outcome.queryFailed());
-        invalidAsinBatchWriter.writeBatch(invalidRows);
+        // 只有被服务端判定为无效品牌（rejected）的行才写入无效品牌表；
+        // queryFailed 是品牌检测服务端到端失败（如缺 X-Token 422/超时/限流），
+        // 一并写入会把故障期间被误伤的品牌永久拉黑，后续同品牌商品全部被
+        // invalidFiltered 过滤（task-27265 实测：通过 Python 过滤的 21 行中
+        // 8 行查询失败被写表后误杀，最终结果 Excel 0 行）。rejected 为空时
+        // 不触发任何写入，避免空批次无意义调用。
+        if (!outcome.rejected().isEmpty()) {
+            invalidAsinBatchWriter.writeBatch(outcome.rejected());
+        }
         return outcome.accepted();
     }
 

backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataServiceTxBoundaryTest.java

@@ -208,6 +208,30 @@ class CollectDataServiceTxBoundaryTest {
         verify(invalidAsinBatchWriter).writeBatch(any());
     }
 
+    @Test
+    void brandQueryFailedRowsNotWrittenToInvalidWriter() {
+        CollectDataResultRowVo queryFailed = new CollectDataResultRowVo();
+        queryFailed.setAsin("B0QUERYFAIL");
+        when(brandBatchFilter.filter(any())).thenReturn(
+                new CollectDataBrandBatchFilter.BrandBatchOutcome(List.of(), List.of(queryFailed), List.of()));
+
+        service.submitResult(TASK_ID, submitRequest());
+
+        verify(invalidAsinBatchWriter, never()).writeBatch(any());
+    }
+
+    @Test
+    void emptyRejectedOutcomeSkipsInvalidWriter() {
+        CollectDataResultRowVo accepted = new CollectDataResultRowVo();
+        accepted.setAsin("B0OK");
+        when(brandBatchFilter.filter(any())).thenReturn(
+                new CollectDataBrandBatchFilter.BrandBatchOutcome(List.of(), List.of(), List.of(accepted)));
+
+        service.submitResult(TASK_ID, submitRequest());
+
+        verify(invalidAsinBatchWriter, never()).writeBatch(any());
+    }
+
     @Test
     void rustfsFailureAbortsWithoutPartialChunk() {
         when(transientPayloadStorageService.extractPointer(anyString())).thenReturn("local:/tmp/x");