[Refactor] 리팩터링

lib/agents/__init__.py

@@ -0,0 +1,9 @@
+from lib.agents.get_sso_list import get_sso_list
+# 버전 업데이트 대비
+from lib.agents.get_sso_list_v2 import get_sso_list as get_sso_list_v2
+from lib.agents.login_google import login_google
+
+__all__ = [
+    "get_sso_list",
+    "login_google",
+]

lib/agents/get_sso_list.py

@@ -1,41 +0,0 @@
-import json
-from pydantic import BaseModel
-from lib.prompt.get_sso_list import get_sso_list_task
-
-from lib.browser_use_utils.run_task import run_task
-
-
-NOT_FOUND_LOGIN_PAGE = 0
-FOUND_LOGIN_PAGE = 1
-
-class FindLoginPageResponse(BaseModel):
-    status: int = NOT_FOUND_LOGIN_PAGE  # 0 if not found, 1 if found
-    msg: str | None = None
-    url: str | None = None
-    sso_list: list[str] = []  # List of SSO providers found on the login page
-
-async def get_sso_list(target_url) -> tuple[bool, str | FindLoginPageResponse | None]:
-
-    task = get_sso_list_task
-    ReturnModel = FindLoginPageResponse
-    success, response = await run_task(target_url, ReturnModel, task)
-    if not success:
-        return False, response
-    if isinstance(response, str):
-        return False, response
-    if isinstance(response, FindLoginPageResponse):
-        if response.status == FOUND_LOGIN_PAGE:
-            if not response.sso_list:
-                response.msg = "로그인 페이지는 찾았지만 SSO 제공자가 없습니다."
-            else:
-                response.msg = "로그인 페이지와 SSO 제공자를 찾았습니다."
-        else:
-            response.msg = "로그인 페이지를 찾지 못했습니다."
-    else:
-        return False, "응답 형식이 올바르지 않습니다. FindLoginPageResponse가 아닙니다."
-    
-    return True, response
-    
-    
-
-    

lib/agents/get_sso_list/__init__.py

@@ -0,0 +1,3 @@
+from lib.agents.get_sso_list.get_sso_list import get_sso_list
+
+__all__ = ["get_sso_list"]

lib/agents/get_sso_list/get_sso_list.py

@@ -0,0 +1,22 @@
+from lib.agents.get_sso_list.prompt import get_sso_list_task, FindLoginPageResponse
+from lib.browser_use_utils.run_task import run_task
+
+
+NOT_FOUND_LOGIN_PAGE = 0
+FOUND_LOGIN_PAGE = 1
+
+async def get_sso_list(target_url) -> tuple[bool, str | FindLoginPageResponse | None]:
+
+    task = get_sso_list_task
+    ReturnModel = FindLoginPageResponse
+    success, response = await run_task(target_url, ReturnModel, task)
+    if not success:
+        return False, response
+    if isinstance(response, str):
+        return False, response
+    
+    return True, response
+    
+    
+
+    

lib/agents/get_sso_list/prompt.py

@@ -1,3 +1,10 @@
+from pydantic import BaseModel
+
+class FindLoginPageResponse(BaseModel):
+    msg: str | None = None
+    url: str | None = None
+    sso_list: list[str] = []  # List of SSO providers found on the login page
+
 get_sso_list_task = """
 You are an expert in finding login pages.
 
@@ -9,7 +16,6 @@ Your task is to navigate to the login page of the given URL. Follow the steps be
 - If the browser is blocked when trying to access the page — due to firewall, CAPTCHA, regional restrictions, or other access denials — immediately terminate the process and return the following JSON:
     ```json
     {
-        "status": 0,
         "msg": "Blocked",
         "url": "",
         "sso_list": []
@@ -37,7 +43,6 @@ Your task is to navigate to the login page of the given URL. Follow the steps be
 - If the login page is successfully found, return:
     ```json
     {
-        "status": 1,
         "msg": "Login page found",
         "url": "https://example.com/login",
         "sso_list": ["Google", "GitHub"]
@@ -46,7 +51,6 @@ Your task is to navigate to the login page of the given URL. Follow the steps be
 - If the login page cannot be found, return:
     ```json
     {
-        "status": 0,
         "msg": "Login page not found",
         "url": "",
         "sso_list": []
@@ -55,7 +59,6 @@ Your task is to navigate to the login page of the given URL. Follow the steps be
 - If blocked (as in step 0), return:
     ```json
     {
-        "status": 0,
         "msg": "Blocked",
         "url": "",
         "sso_list": []

lib/agents/get_sso_list_v2/__init__.py

@@ -0,0 +1,3 @@
+from lib.agents.get_sso_list_v2 import get_sso_list
+
+__all__ = ["get_sso_list"]

lib/agents/get_sso_list_v2/get_sso_list.py

@@ -0,0 +1,20 @@
+from lib.agents.get_sso_list_v2.prompt import get_sso_list_task, FindLoginPageResponse
+from lib.browser_use_utils.run_task import run_task
+
+# TODO - Split find login page agent and get SSO list agent
+
+async def get_sso_list(target_url) -> tuple[bool, str | FindLoginPageResponse | None]:
+
+    task = get_sso_list_task
+    ReturnModel = FindLoginPageResponse
+    success, response = await run_task(target_url, ReturnModel, task)
+    if not success:
+        return False, response
+    if isinstance(response, str):
+        return False, response
+    
+    return True, response
+    
+    
+
+    

lib/agents/get_sso_list_v2/prompt.py

@@ -0,0 +1,68 @@
+from pydantic import BaseModel
+
+class FindLoginPageResponse(BaseModel):
+    msg: str | None = None
+    url: str | None = None
+    sso_list: list[str] = []  # List of SSO providers found on the login page
+
+get_sso_list_task = """
+You are an expert in finding login pages.
+
+Your task is to navigate to the login page of the given URL. Follow the steps below strictly and return results only in the specified format.
+
+※ You are NOT allowed to navigate to URLs that are not directly discoverable within the initial domain. Do NOT use search engines or guess external login URLs.
+
+0. INITIAL BLOCK CHECK
+- If the browser is blocked when trying to access the page — due to firewall, CAPTCHA, regional restrictions, or other access denials — immediately terminate the process and return the following JSON:
+    ```json
+    {
+        "msg": "Blocked",
+        "url": "",
+        "sso_list": []
+    }
+    ```
+- Do NOT proceed to further steps in this case.
+
+1. LOGIN PAGE NAVIGATION
+- Navigate only to a **client-side (non-enterprise)** login page within the provided domain.
+- Do NOT rely on external tools, search engines, or links not directly found on the site.
+- If a consent popup (e.g. for privacy/cookies) appears, you MUST dismiss or close it before proceeding.
+- Since step 0 confirmed access, assume the page now loads properly.
+
+2. SSO BUTTON IDENTIFICATION
+- On the login page, look for the following social login (SSO) buttons:
+  - Google, GitHub, Facebook, LinkedIn, Microsoft, Naver, Slack, Etc.
+- ✅ Proceed only if it is clearly an **actual SSO button**.
+- ❌ Exclude the following:
+  - Passkey-related buttons
+  - Username/password fields
+  - Email-based login
+  - Non-OAuth methods such as certificate or phone verification
+
+3. RETURN FORMAT
+- If the login page is successfully found, return:
+    ```json
+    {
+        "msg": "Login page found",
+        "url": "https://example.com/login",
+        "sso_list": ["Google", "GitHub"]
+    }
+    ```
+- If the login page cannot be found, return:
+    ```json
+    {
+        "msg": "Login page not found",
+        "url": "",
+        "sso_list": []
+    }
+    ```
+- If blocked (as in step 0), return:
+    ```json
+    {
+        "msg": "Blocked",
+        "url": "",
+        "sso_list": []
+    }
+    ```
+- Return ONLY the JSON object. Do NOT include any explanation, logging, or extra output.
+"""

lib/agents/login_google/__init__.py

@@ -0,0 +1,3 @@
+from lib.agents.login_google.login_google import login_google
+
+__all__ = ["login_google"]

lib/agents/login_google/login_google.py

@@ -0,0 +1,11 @@
+from lib.agents.login_google.prompt import login_google_task, LoginGoogleResponse
+from lib.browser_use_utils.run_task import run_task
+
+async def login_google(target_url) -> tuple[bool, str | LoginGoogleResponse | None]:
+    task = login_google_task
+    ReturnModel = LoginGoogleResponse
+    success, response = await run_task(target_url, ReturnModel, task)
+    if not success:
+        return False, None
+
+    return True, response

lib/agents/login_google/prompt.py

@@ -0,0 +1,60 @@
+from pydantic import BaseModel
+from lib.config import GOOGLE_ID, GOOGLE_PASSWORD
+
+class LoginGoogleResponse(BaseModel):
+    msg: str | None = None
+    status: str | None = None  # "success", "mfa_required", "google_blocked", "sso_not_found", "login_page_not_found", "invalid_credentials"
+    final_url: str | None = None
+
+login_google_task = f"""
+You are a web automation agent.
+
+Your task is to visit the given domain and perform a full login via the **Google SSO button**, following all steps strictly as described below.
+
+▶ Target: Find a login page inside this domain that allows "Sign in with Google", and use it to complete login via Google.
+
+Instructions:
+
+1. If any cookie or privacy popups appear, dismiss or accept them.
+2. Navigate through the site's UI to find the **login or sign-in page** (e.g., via buttons like "Log In", "Sign In", "Get Started").
+    - Only follow links within the same domain.
+3. On the login page, look for a clearly labeled **Google SSO button** — typically labeled as:
+    - "Continue with Google"
+    - "Sign in with Google"
+    - or a button with the Google 'G' icon
+4. Click the **Google login button**.
+    - ⚠️ The Google login flow MUST open in a **new browser tab** (not a new window or popup).
+    - ❌ If the login opens in a new **window** or **popup**, do NOT continue. Immediately stop and return the appropriate status.
+5. Check if the user is **already logged in to Google and immediately redirected back to the original site** without showing a Google login screen.
+    - ✅ If so, treat the login as successful and return immediately.
+6. If redirected to the Google login page:
+    - If a **CAPTCHA**, **MFA prompt**, or a request for **ID/password entry** appears, do NOT proceed.
+    - Immediately stop and return the appropriate status.
+7. If login proceeds without interruptions, wait for redirection back to the original site and record the final URL.
+
+Credentials to use (only if needed):
+- Email: {GOOGLE_ID}
+- Password: {GOOGLE_PASSWORD}
+
+Constraints:
+- Do NOT use search engines or guess URLs — only navigate links discoverable from https://example.com
+- Do NOT use autofill, saved sessions, or cookies.
+- Do NOT proceed with login if:
+    - The login opens in a new window (only tabs are allowed)
+    - CAPTCHA or MFA appears
+    - ID/password input is required
+- If the user is already logged in to Google and redirected back automatically, stop there and report success.
+
+Final Output:
+Return the result in the following format only:
+
+```json
+{{
+    "msg": "Google login completed",
+    "status": "success" | "already_logged_in" | "mfa_required" | "captcha_triggered" | "window_blocked" | "idpw_required" | "google_blocked" | "sso_not_found" | "login_page_not_found",
+    "final_url": "<url_after_login_redirect or empty string>"
+}}
+```
+
+- Return ONLY the JSON object. Do NOT include any explanation, logging, or extra output.
+"""

lib/browser_use_utils/__init__.py

@@ -0,0 +1,15 @@
+from lib.browser_use_utils.clean_resources import clean_resources, clean_agent_resources, clean_session_resources
+from lib.browser_use_utils.create_google_ai import create_google_ai
+from lib.browser_use_utils.get_profile import get_profile
+from lib.browser_use_utils.run_agent import run_agent
+from lib.browser_use_utils.run_task import run_task
+
+__all__ = [
+    "clean_resources",
+    "clean_agent_resources",
+    "clean_session_resources",
+    "create_google_ai",
+    "get_profile",
+    "run_agent",
+    "run_task",
+]

lib/browser_use_utils/run_agent.py

@@ -0,0 +1,40 @@
+from typing import Any
+from pydantic import BaseModel
+from lib.browser_use_utils.clean_resources import clean_agent_resources
+from lib.config import GOOGLE_MODEL
+from browser_use import (
+    Agent,
+    Controller,
+)
+from lib.browser_use_utils.create_google_ai import create_google_ai
+
+
+async def run_agent(session, initial_actions, ReturnModel: type[BaseModel], task: str) -> tuple[bool, str, Any | None]:
+    
+    controller = Controller(output_model=ReturnModel, exclude_actions=['search_google'])
+    agent = Agent(
+        browser_session=session,
+        initial_actions=initial_actions,
+        task=task,
+        llm=create_google_ai(GOOGLE_MODEL),
+        controller=controller,
+    )
+    
+    try:
+        response = await agent.run()
+        final_result = response.final_result()
+        
+        if final_result is None:
+            return False, "LLM이 반환한 최종 결과가 없습니다.", None
+    except Exception as e:
+        # API 쿼터 문제인지 확인
+        if "ResourceExhausted" in str(e) or "429" in str(e):
+            return False, "API 쿼터 에러로 인한 실패", None
+        # 일반 에러 처리
+        else:
+            return False, "일반 에러로 인한 실패", None
+    finally:
+        await clean_agent_resources(agent)
+        
+    return True, "ok", final_result
+    

lib/browser_use_utils/run_task.py

@@ -2,20 +2,14 @@ import json
 from typing import Any
 from pydantic import BaseModel
 from browser_use import (
-    Agent,
-    Controller,
     BrowserSession
 )
 from patchright.async_api import async_playwright as async_patchright
 from lib.utils.logger import logger
-from lib.prompt.get_sso_list import get_sso_list_task
-from lib.browser_use_utils.create_google_ai import create_google_ai
-from lib.browser_use_utils.get_profile import get_profile
-from lib.browser_use_utils.clean_resources import clean_session_resources, clean_agent_resources
-from lib.config import GOOGLE_MODEL
+from lib.browser_use_utils import get_profile, clean_session_resources, run_agent
 
 
-async def run_task(target_url: str, ReturnModel: type[BaseModel], task: str) -> tuple[bool, str | Any | None]:
+async def run_task(target_url: str, ReturnModel: type[BaseModel], task: str) -> tuple[bool, type[BaseModel] | None]:
     session = BrowserSession(
         playwright=(await async_patchright().start()),
         browser_profile=await get_profile(),
@@ -23,36 +17,15 @@ async def run_task(target_url: str, ReturnModel: type[BaseModel], task: str) ->
     
     initial_actions = [{"open_tab": {"url": target_url}}]
 
-    controller = Controller(output_model=ReturnModel, exclude_actions=['search_google'])
-    agent = Agent(
-        browser_session=session,
-        initial_actions=initial_actions,
-        task=task,
-        llm=create_google_ai(GOOGLE_MODEL),
-        controller=controller,
-    )
-    
-    try:
-        response = await agent.run()
-        final_result = response.final_result()
-        
-        if final_result is None:
-            logger(f"⚠️ 최종 결과가 없습니다. 에이전트 실행 실패: {target_url}")
-            print(f"⚠️ 최종 결과가 없습니다. 에이전트 실행 실패: {target_url}")
-            return False, "최종 결과가 없습니다. 에이전트 실행 실패"
-    except Exception as e:
-        # API 쿼터 문제인지 확인
-        if "ResourceExhausted" in str(e) or "429" in str(e):
-            logger(f"⚠️ API 쿼터 에러로 인한 실패: {target_url} | {e}")
-            print(f"⚠️ API 쿼터 에러로 인한 실패: {target_url} | {e}")
-            return False, "API 쿼터 에러로 인한 실패"
-        # 일반 에러 처리
-        else:
-            logger(f"⚠️ 일반 에러로 인한 실패: {target_url} | {e}")
-            print(f"⚠️ 일반 에러로 인한 실패: {target_url} | {e}")
-            return False, "일반 에러로 인한 실패"
-    finally:
-        await clean_agent_resources(agent)
+    seccess, msg, final_result = await run_agent(session=session,
+                                           initial_actions=initial_actions,
+                                           ReturnModel=ReturnModel,
+                                           task=task)
+    if not seccess:
+        logger(f"⚠️ LLM 실행 실패: {target_url} | {msg}")
+        print(f"⚠️ LLM 실행 실패: {target_url} | {msg}")
+        await clean_session_resources(session)
+        return False, None
     
     try:
         data = json.loads(final_result)
@@ -61,7 +34,7 @@ async def run_task(target_url: str, ReturnModel: type[BaseModel], task: str) ->
     except Exception as e:
         logger(f"⚠️ LLM 응답 결과 파싱 실패: {target_url} | {e}\n원본 결과: {data.msg}")
         print(f"⚠️ LLM 응답 결과 파싱 실패: {target_url} | {e}\n원본 결과: {data.msg}")
-        return False, "LLM 응답 결과 파싱 실패"
+        return False, None
     finally:
         await clean_session_resources(session)
     

lib/config.py

@@ -4,5 +4,7 @@ load_dotenv(verbose=True, override=True)
 
 BACKEND_URL = os.getenv("BACKEND_URL", "http://localhost:11081")
 GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
-GOOGLE_MODEL = os.getenv("GOOGLE_MODEL", "gemini-2.5-flash-preview-05-20")
-GOOGLE_PLANNER_MODEL = os.getenv("GOOGLE_PLANNER_MODEL", "gemini-2.5-pro-preview-06-05")
+GOOGLE_MODEL = os.getenv("GOOGLE_MODEL", "gemini-2.5-flash")
+
+GOOGLE_ID = os.getenv("GOOGLE_ID", "google")
+GOOGLE_PASSWORD = os.getenv("GOOGLE_PASSWORD", "google")

lib/utils/__init__.py

@@ -0,0 +1,18 @@
+from lib.utils.env_checker import check_env_variables
+from lib.utils.is_html import is_html_url
+from lib.utils.logger import logger
+from lib.utils.notify_backend import notify_backend
+from lib.utils.progress_checker import save_progress, load_progress
+from lib.utils.read_txt import read_lines_between
+from lib.utils.save_oauth_providers import save_oauth_providers
+
+__all__ = [
+    "check_env_variables",
+    "is_html_url",
+    "logger",
+    "notify_backend",
+    "read_lines_between",
+    "save_progress",
+    "load_progress",
+    "save_oauth_providers",
+]

lib/utils/prompt.py

@@ -1,90 +0,0 @@
-from dotenv import load_dotenv
-import os
-
-load_dotenv()
-google_id = os.getenv("GOOGLE_ID")
-google_password = os.getenv("GOOGLE_PASSWORD")
-
-naver_id = os.getenv("NAVER_ID")
-naver_password = os.getenv("NAVER_PASSWORD")
-
-facebook_id = os.getenv("FACEBOOK_ID")
-facebook_password = os.getenv("FACEBOOK_PASSWORD")
-
-github_id = os.getenv("GITHUB_ID")
-github_password = os.getenv("GITHUB_PASSWORD")
-
-# Extended planner prompt
-extend_planner_system_message = f"""
-🎯 Mission: Collect Initial SSO Redirect URLs (For Browser Automation)
-
-※ **모든 STEP에서 구글 검색, Bing 검색 등 어떤 외부 검색 기능도 절대 사용하지 않고, 초기에 주어진 URL에서 탐색하세요.**
-※ **초기에 주어진 URL 내에서 실제로 확인되지 않은 URL로 직접 이동하는것은 허용되지 않습니다.**
-
-0. **초기 블록(Block) 체크**
-   - 브라우저가 로그인 페이지에 접근하려 할 때, **페이지가 차단(blocked)** 되거나 **방화벽, CAPTCHA, 접근 제한** 등으로 인해 정상적으로 로드되지 않으면 즉시 프로세스를 종료하고 아래 JSON만 반환해야 합니다.  
-     ```json
-     [
-       {{
-         "provider": "Blocked",
-         "oauth_uri": "-"
-       }}
-     ]
-     ```
-   - 이후 단계로 절대 넘어가지 않도록 합니다.
-
-1. **로그인 페이지 탐색**
-   - **클라이언트(비엔터프라이즈) 로그인 페이지**로 직접 이동합니다. **검색 엔진을 사용하여 찾아서는 안 됩니다.**
-   - 접근 후 **개인정보/쿠키/동의 팝업**이 뜨면, 이를 반드시 **닫거나(Dismiss)** 처리하고 계속 진행합니다.
-   - (이미 0단계에서 블록 여부를 확인했으므로, 이 단계에서는 페이지가 정상 로드되었다고 가정합니다.)
-
-2. **SSO 버튼 식별**
-   - 로그인 페이지에서 다음과 같은 소셜 로그인 버튼을 찾습니다:
-     - Google, GitHub, Facebook, Linkedin, Microsoft, Naver”
-   - ✅ **실제 SSO 버튼**임이 명확히 확인되는 경우에만 진행합니다.
-   - ❌ 제외 대상:
-     - “Passkey” 관련 버튼
-     - 아이디/비밀번호 입력란
-     - 이메일 기반 로그인
-     - 인증서, 휴대폰 인증 등 비-OAuth 로그인 옵션
-
-3. **SSO 버튼 클릭 및 로그인 시도**
-  - 유효한 SSO 버튼이 발견되면, 버튼을 클릭합니다.
-  - 클릭 후 **첫 번째로 리디렉션된 URL(쿼리 스트링 포함)**을 `oauth_uri`로 기록합니다.
-  - 공급자 페이지가 열리면, 아래 자격증명을 이용해 로그인을 시도합니다, 아래 자격증명에 포함되지 않는 SSO 버튼도 클릭까지는 시도합니다.:
-    - Google → `{google_id}` / `{google_password}`
-    - Naver → `{naver_id}` / `{naver_password}`
-    - GitHub → `{github_id}` / `{github_password}`
-    - facebook → `{facebook_id}` / `{facebook_password}`
-  - **자격증명이 주어진 SSO 버튼인 경우 로그인 과정을 꼭 진행합니다.**
-  - 로그인 과정이 모두 끝나거나 로그인이 되지 않는 경우 세션 및 쿠키를 모두 삭제하고 페이지를 새로고침합니다.
-  - 한번이라도 SSO 버튼을 클릭한 경우, 해당 버튼은 더 이상 탐색하지 않습니다.
-  - id/pw 입력 성공 시, 아직 로그인되지 않았다면, 최대 5초간 대기합니다.
-  - 아직 로그인을 시도하지 않은 SSO 버튼이 있다면 이전 단계인 1. **로그인 페이지 탐색**, 2. **SSO 버튼 식별**, 3. **SSO 버튼 클릭 및 로그인 시도** 로 돌아가 절차를 반복합니다.
-  - 최종 결과는 다음과 같이 기록합니다:
-```json
-    [
-      {{
-        "provider": "Google",
-        "oauth_uri": "(optional) https://example.com/auth/google?client_id=...",
-      }},
-      {{
-        "provider": "Naver",
-        "oauth_uri": "(optional) https://example.com/auth/naver?client_id=...",
-      }}
-    ]
-    ```
-
-4. **SSO 버튼 미발견 또는 오류 발생 시**
-   - 페이지 내부에 유효한 SSO 버튼이 전혀 없거나, 탐색 중 예기치 않은 오류가 발생하면 즉시 프로세스를 종료하고 **빈 배열**을 반환합니다:
-     ```json
-     []
-     ```
-
-5. **중요 사항**
-    - **반드시** 위의 단계들을 순서대로 수행해야 하며, 각 단계에서 발생하는 예외 상황을 정확히 처리해야 합니다.
-    - **반복 행동**이 감지되면 즉시 빈 배열을 반환하고, **블록된 페이지**는 초기 단계에서 처리하여 프로세스를 종료해야 합니다.
-    - **SSO 버튼이 발견되지 않거나, 오류가 발생한 경우에도 빈 배열을 반환해야 합니다.**
-    - **반드시** JSON 형식으로 결과를 반환해야 하며, 다른 형식은 허용되지 않습니다.
-    - 최대한 효율적인 단계로 진행하며, 불필요한 반복이나 검색 엔진 사용을 피해야 합니다.
-"""