refactor(auth): implement two-pass OAuth credential deduplication

Refactors the OAuth token initialization process to proactively handle duplicate credentials based on user email across providers.

The new flow operates in two passes:
- Pass 1 (Pre-scan): Reads metadata from files to identify and skip duplicates based on existing email tags before initialization.
- Pass 2 (Post-init): Initializes unique tokens, retrieves user email via `get_user_info`, and performs a final deduplication check, updating metadata only for unique credentials.

Additionally:
- Introduce automatic refresh token usage for Gemini and Qwen providers when tokens are expired, reducing interactive login necessity.
- Disable automatic scanning of default credential directories in `credential_manager` to prefer locally defined or explicit paths.

src/proxy_app/main.py

@@ -163,42 +163,98 @@ async def lifespan(app: FastAPI):
         raise ValueError("No provider API keys or OAuth credentials found.")
 
     if not skip_oauth_init and oauth_credentials:
-        logging.info("Validating OAuth credentials and checking for duplicates...")
-        processed_emails = {} # email -> {provider: path}
+        logging.info("Starting OAuth credential validation and deduplication...")
+        processed_emails = {}  # email -> {provider: path}
+        credentials_to_initialize = {} # provider -> [paths]
+        final_oauth_credentials = {}
+
+        # --- Pass 1: Pre-initialization Scan & Deduplication ---
+        #logging.info("Pass 1: Scanning for existing metadata to find duplicates...")
         for provider, paths in oauth_credentials.items():
+            if provider not in credentials_to_initialize:
+                credentials_to_initialize[provider] = []
+            for path in paths:
+                try:
+                    with open(path, 'r') as f:
+                        data = json.load(f)
+                    metadata = data.get("_proxy_metadata", {})
+                    email = metadata.get("email")
+
+                    if email:
+                        if email not in processed_emails:
+                            processed_emails[email] = {}
+                        
+                        if provider in processed_emails[email]:
+                            original_path = processed_emails[email][provider]
+                            logging.warning(f"Duplicate for '{email}' on '{provider}' found in pre-scan: '{Path(path).name}'. Original: '{Path(original_path).name}'. Skipping.")
+                            continue
+                        else:
+                            processed_emails[email][provider] = path
+                    
+                    credentials_to_initialize[provider].append(path)
+
+                except (FileNotFoundError, json.JSONDecodeError) as e:
+                    logging.warning(f"Could not pre-read metadata from '{path}': {e}. Will process during initialization.")
+                    credentials_to_initialize[provider].append(path)
+        
+        # --- Pass 2: Initialization of Filtered Credentials & Final Check ---
+        #logging.info("Pass 2: Initializing unique credentials and performing final check...")
+        for provider, paths in credentials_to_initialize.items():
+            if not paths: continue
+
             provider_plugin_class = PROVIDER_PLUGINS.get(provider)
             if not provider_plugin_class: continue
             
             provider_instance = provider_plugin_class()
+            
             for path in paths:
                 try:
                     await provider_instance.initialize_token(path)
-                    if hasattr(provider_instance, 'get_user_info'):
+
+                    if not hasattr(provider_instance, 'get_user_info'):
+                        if provider not in final_oauth_credentials:
+                            final_oauth_credentials[provider] = []
+                        final_oauth_credentials[provider].append(path)
+                        continue
+
+                    user_info = await provider_instance.get_user_info(path)
+                    email = user_info.get("email")
+
+                    if not email:
+                        logging.warning(f"Could not retrieve email for '{path}'. Treating as unique.")
+                        if provider not in final_oauth_credentials:
+                            final_oauth_credentials[provider] = []
+                        final_oauth_credentials[provider].append(path)
+                        continue
+
+                    if email not in processed_emails:
+                        processed_emails[email] = {}
+                    
+                    if provider in processed_emails[email] and processed_emails[email][provider] != path:
+                        original_path = processed_emails[email][provider]
+                        logging.warning(f"Duplicate for '{email}' on '{provider}' found post-init: '{Path(path).name}'. Original: '{Path(original_path).name}'. Skipping.")
+                        continue
+                    else:
+                        processed_emails[email][provider] = path
+                        if provider not in final_oauth_credentials:
+                            final_oauth_credentials[provider] = []
+                        final_oauth_credentials[provider].append(path)
+
                         with open(path, 'r+') as f:
                             data = json.load(f)
                             metadata = data.get("_proxy_metadata", {})
-                            last_check = metadata.get("last_check_timestamp", 0)
-                            if time.time() - last_check > 86400:
-                                user_info = await provider_instance.get_user_info(path)
-                                metadata["email"] = user_info.get("email")
-                                metadata["last_check_timestamp"] = time.time()
-                                data["_proxy_metadata"] = metadata
-                                f.seek(0)
-                                json.dump(data, f, indent=2)
-                                f.truncate()
-                            
-                            email = metadata.get("email")
-                            if email:
-                                if email not in processed_emails:
-                                    processed_emails[email] = {}
-                                if provider in processed_emails[email]:
-                                    original_path = processed_emails[email][provider]
-                                    logging.warning(f"Duplicate credential for user '{email}' on provider '{provider}' found at '{Path(path).name}'. Original at '{Path(original_path).name}'.")
-                                else:
-                                    processed_emails[email][provider] = path
+                            metadata["email"] = email
+                            metadata["last_check_timestamp"] = time.time()
+                            data["_proxy_metadata"] = metadata
+                            f.seek(0)
+                            json.dump(data, f, indent=2)
+                            f.truncate()
+
                 except Exception as e:
                     logging.error(f"Failed to process OAuth token for {provider} at '{path}': {e}")
+
         logging.info("OAuth credential processing complete.")
+        oauth_credentials = final_oauth_credentials
 
     # [NEW] Load provider-specific params
     litellm_provider_params = {

src/rotator_library/credential_manager.py

@@ -56,9 +56,10 @@ class CredentialManager:
                     discovered_paths.add(path)
             
             # 2. If no overrides are provided via .env, scan the default directory
-            if not discovered_paths and default_dir.exists():
-                for json_file in default_dir.glob('*.json'):
-                    discovered_paths.add(json_file)
+            # [MODIFIED] This logic is now disabled to prefer local-first credential management.
+            # if not discovered_paths and default_dir.exists():
+            #     for json_file in default_dir.glob('*.json'):
+            #         discovered_paths.add(json_file)
             
             if not discovered_paths:
                 lib_logger.debug(f"No credential files found for provider: {provider}")

src/rotator_library/providers/gemini_auth_base.py

@@ -122,6 +122,12 @@ class GeminiAuthBase:
                 reason = "token is expired"
 
             if reason:
+                if reason == "token is expired" and creds.get("refresh_token"):
+                    try:
+                        return await self._refresh_token(path, creds)
+                    except Exception as e:
+                        lib_logger.warning(f"Automatic token refresh for '{file_name}' failed: {e}. Proceeding to interactive login.")
+
                 lib_logger.warning(f"Gemini OAuth token for '{file_name}' needs setup: {reason}.")
                 auth_code_future = asyncio.get_event_loop().create_future()
                 server = None

src/rotator_library/providers/qwen_auth_base.py

@@ -141,6 +141,12 @@ class QwenAuthBase:
                 reason = "token is expired"
 
             if reason:
+                if reason == "token is expired" and creds.get("refresh_token"):
+                    try:
+                        return await self._refresh_token(path)
+                    except Exception as e:
+                        lib_logger.warning(f"Automatic token refresh for '{file_name}' failed: {e}. Proceeding to interactive login.")
+                
                 lib_logger.warning(f"Qwen OAuth token for '{file_name}' needs setup: {reason}.")
                 code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode('utf-8').rstrip('=')
                 code_challenge = base64.urlsafe_b64encode(