fix(import): fail auto import on drive auth failure

services/google_drive_service.py

@@ -38,6 +38,16 @@ class GoogleDriveService:
         """初始化 Google Drive 服務"""
         self.service = None
         self.credentials = None
+        self.last_error = None
+        self.last_error_kind = None
+
+    def _clear_error(self) -> None:
+        self.last_error = None
+        self.last_error_kind = None
+
+    def _set_error(self, kind: str, message: str) -> None:
+        self.last_error_kind = kind
+        self.last_error = str(message)[:500]
 
     @staticmethod
     def _escape_query_value(value: str) -> str:
@@ -51,6 +61,7 @@ class GoogleDriveService:
             bool: 認證是否成功
         """
         try:
+            self._clear_error()
             # 舊版 pickle token 遷移提示（不自動刪除舊檔）
             if os.path.exists(_LEGACY_PICKLE_FILE) and not os.path.exists(TOKEN_FILE):
                 logger.warning(
@@ -73,7 +84,9 @@ class GoogleDriveService:
                 else:
                     # 需要重新認證
                     if not os.path.exists(CREDENTIALS_FILE):
-                        logger.error(f"找不到認證檔案: {CREDENTIALS_FILE}")
+                        error_message = f"找不到認證檔案: {CREDENTIALS_FILE}"
+                        self._set_error("authentication_failed", error_message)
+                        logger.error(error_message)
                         return False
 
                     logger.info("進行 Google Drive 認證...")
@@ -90,10 +103,12 @@ class GoogleDriveService:
 
             # 建立 Drive API 服務
             self.service = build('drive', 'v3', credentials=self.credentials)
+            self._clear_error()
             logger.info("Google Drive 服務已連接")
             return True
 
         except Exception as e:
+            self._set_error("authentication_failed", str(e))
             logger.error(f"Google Drive 認證失敗: {str(e)}")
             return False
 
@@ -111,6 +126,8 @@ class GoogleDriveService:
         if not self.service:
             if not self.authenticate():
                 return []
+        else:
+            self._clear_error()
 
         try:
             # 首先找到資料夾 ID
@@ -138,11 +155,13 @@ class GoogleDriveService:
             ).execute()
 
             files = results.get('files', [])
+            self._clear_error()
             logger.info(f"在 {folder_path} 找到 {len(files)} 個 Excel 檔案")
 
             return files
 
         except HttpError as error:
+            self._set_error("drive_api_failed", str(error))
             logger.error(f"列出檔案時發生錯誤: {error}")
             return []
 

services/import_service.py

@@ -982,6 +982,24 @@ class ImportService:
 
             # 列出檔案
             files = drive_service.list_files_in_folder(folder_path, file_pattern)
+            drive_error_kind = getattr(drive_service, "last_error_kind", None)
+            drive_error = getattr(drive_service, "last_error", None)
+
+            if drive_error_kind:
+                message = (
+                    "Google Drive 連線或認證失敗，未能確認來源資料夾是否有新檔案："
+                    f"{drive_error or drive_error_kind}"
+                )
+                logger.error(message)
+                return {
+                    'success': False,
+                    'message': message,
+                    'file_count': 0,
+                    'imported_count': 0,
+                    'failed_count': 1,
+                    'connection_error': True,
+                    'error_kind': drive_error_kind,
+                }
 
             if not files:
                 logger.info("沒有找到待匯入的檔案")
@@ -1201,14 +1219,15 @@ class ImportService:
             is_connection_error = any(err.lower() in error_msg.lower() for err in connection_errors)
 
             if is_connection_error:
-                # 連線錯誤：返回成功但無檔案（避免發送告警）
-                logger.warning(f"Google Drive 連線問題，跳過本次匯入檢查: {error_msg}")
+                # Drive 連線 / 認證錯誤不是「無檔案」，必須 fail-closed 才能觸發告警與人工補件。
+                logger.error(f"Google Drive 連線問題，無法確認待匯入檔案: {error_msg}")
                 return {
-                    'success': True,  # 標記為成功避免告警
-                    'message': f'Google Drive 連線問題，跳過本次檢查',
+                    'success': False,
+                    'message': f'Google Drive 連線問題，無法確認待匯入檔案: {error_msg}',
                     'file_count': 0,
                     'imported_count': 0,
-                    'connection_error': True  # 標記為連線錯誤供日誌記錄
+                    'failed_count': 1,
+                    'connection_error': True
                 }
             else:
                 # 真正的匯入錯誤：返回失敗

tests/test_auto_import_failure_boundaries.py

@@ -143,3 +143,50 @@ def test_auto_import_does_not_move_drive_file_when_import_fails(monkeypatch, tmp
     assert result["success"] is False
     assert result["failed_count"] == 1
     assert fake_drive.moved_files == []
+
+
+def test_auto_import_fails_closed_when_drive_auth_fails(monkeypatch, tmp_path):
+    import_service = _load_import_service(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}")
+    import_service.Base.metadata.create_all(import_service.engine)
+
+    class FakeDriveService:
+        last_error_kind = "authentication_failed"
+        last_error = "could not locate runnable browser"
+
+        def list_files_in_folder(self, folder_path, file_pattern):
+            return []
+
+    monkeypatch.setattr(import_service, "drive_service", FakeDriveService())
+
+    service = import_service.ImportService()
+    result = service.auto_import_from_drive()
+
+    assert result["success"] is False
+    assert result["file_count"] == 0
+    assert result["imported_count"] == 0
+    assert result["failed_count"] == 1
+    assert result["connection_error"] is True
+    assert result["error_kind"] == "authentication_failed"
+    assert "Google Drive" in result["message"]
+    assert "could not locate runnable browser" in result["message"]
+
+
+def test_auto_import_empty_drive_folder_remains_success(monkeypatch, tmp_path):
+    import_service = _load_import_service(monkeypatch, f"sqlite:///{tmp_path / 'momo.db'}")
+    import_service.Base.metadata.create_all(import_service.engine)
+
+    class FakeDriveService:
+        last_error_kind = None
+        last_error = None
+
+        def list_files_in_folder(self, folder_path, file_pattern):
+            return []
+
+    monkeypatch.setattr(import_service, "drive_service", FakeDriveService())
+
+    service = import_service.ImportService()
+    result = service.auto_import_from_drive()
+
+    assert result["success"] is True
+    assert result["file_count"] == 0
+    assert result["message"] == "沒有找到待匯入的檔案"