mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 19:37:54 +08:00
e0642eafa7
* chore(types): clear AniList endpoint mypy debt * chore(types): clear Douban endpoint mypy debt * chore(types): clear recommendation endpoint mypy debt * chore(types): clear subscription notification mypy debt * chore(types): clear SDK string compatibility mypy debt * chore(types): ratchet v3 host mypy across remaining files * chore(types): reduce v3 host mypy debt and retire ledger * chore(types): reconcile host snapshot with upstream HTTP pool * test(types): cover host paths exposed by mypy cleanup
788 lines
33 KiB
Python
788 lines
33 KiB
Python
"""
|
||
通过注入浏览器、HTTP 和识别端口完成站点登录与 Cookie 获取。
|
||
"""
|
||
|
||
import time
|
||
from typing import Any, Callable, Optional, Protocol, Tuple
|
||
from urllib.parse import urljoin, urlparse
|
||
|
||
from lxml import etree
|
||
|
||
from app.application.security.twofactor import TwoFactorAuth
|
||
from app.domain.site import SiteUtils
|
||
from app.runtime.log import logger
|
||
|
||
CookieResult = Tuple[Optional[str], Optional[str], str]
|
||
|
||
|
||
class CookiePageContentPort(Protocol):
|
||
"""
|
||
声明登录页源码读取和加载等待所需的页面能力。
|
||
"""
|
||
|
||
def wait_for_load_state(self, state: str, *, timeout: int) -> None:
|
||
"""
|
||
等待页面进入指定加载状态。
|
||
|
||
:param state: 目标加载状态
|
||
:param timeout: 最长等待时间，单位为毫秒
|
||
"""
|
||
...
|
||
|
||
def content(self) -> str:
|
||
"""
|
||
读取当前页面源码，导航期间的异常由实现方抛出。
|
||
|
||
:return: 页面 HTML 源码
|
||
"""
|
||
...
|
||
|
||
|
||
class CookieBrowserPort(Protocol):
|
||
"""
|
||
声明站点登录用例所需的受控浏览器执行能力。
|
||
"""
|
||
|
||
def action(self, *, url: str, callback: Callable[[Any], CookieResult],
|
||
proxies: Optional[dict[str, Any]], timeout: Optional[int]) -> CookieResult:
|
||
"""
|
||
在受控页面会话内执行登录回调并返回兼容三元组。
|
||
|
||
页面及浏览器资源的获取与释放由实现方负责，回调不得把页面留给会话外使用。
|
||
|
||
:param url: 站点登录页面地址
|
||
:param callback: 在已创建页面会话内执行的登录回调，返回 Cookie 结果三元组
|
||
:param proxies: 浏览器代理配置；None 表示不指定代理
|
||
:param timeout: 浏览器操作超时，单位为秒；None 使用实现默认值
|
||
:return: 回调产生的 Cookie、User-Agent、错误说明三元组；浏览器未完成回调时沿用底层 None 回退
|
||
"""
|
||
|
||
|
||
class CaptchaHttpPort(Protocol):
|
||
"""
|
||
声明验证码图片下载能力。
|
||
"""
|
||
|
||
def fetch(self, *, url: str, cookie: str, ua: str) -> Optional[bytes]:
|
||
"""
|
||
下载验证码图片字节。
|
||
|
||
:param url: 验证码图片的绝对地址
|
||
:param cookie: 当前站点会话的 Cookie 请求头文本
|
||
:param ua: 当前浏览器会话使用的 User-Agent
|
||
:return: 下载的原始图片字节；未取得有效响应时返回 None
|
||
"""
|
||
|
||
|
||
class CaptchaOcrPort(Protocol):
|
||
"""
|
||
声明验证码图片识别能力。
|
||
"""
|
||
|
||
def recognize(self, image_data: bytes) -> str:
|
||
"""
|
||
识别原始验证码图片字节，避免在应用层重复 Base64 编码。
|
||
|
||
:param image_data: 未做 Base64 编码的验证码图片字节
|
||
:return: 识别出的验证码文本；未取得识别结果时返回空字符串
|
||
"""
|
||
|
||
|
||
_cookie_browser_port: Optional[CookieBrowserPort] = None
|
||
_captcha_http_port: Optional[CaptchaHttpPort] = None
|
||
_captcha_ocr_port: Optional[CaptchaOcrPort] = None
|
||
|
||
|
||
def configure_cookie_ports(*, browser: CookieBrowserPort, http: CaptchaHttpPort,
|
||
ocr: CaptchaOcrPort) -> None:
|
||
"""
|
||
由组合根装配站点登录与验证码端口。
|
||
"""
|
||
global _cookie_browser_port, _captcha_http_port, _captcha_ocr_port
|
||
_cookie_browser_port = browser
|
||
_captcha_http_port = http
|
||
_captcha_ocr_port = ocr
|
||
|
||
|
||
def reset_cookie_ports() -> None:
|
||
"""
|
||
清除站点登录端口，避免跨生命周期保留 Adapter。
|
||
"""
|
||
global _cookie_browser_port, _captcha_http_port, _captcha_ocr_port
|
||
_cookie_browser_port = None
|
||
_captcha_http_port = None
|
||
_captcha_ocr_port = None
|
||
|
||
|
||
def _require_cookie_ports() -> Tuple[CookieBrowserPort, CaptchaHttpPort, CaptchaOcrPort]:
|
||
"""
|
||
返回已装配端口，缺失时明确拒绝隐式构造 Adapter。
|
||
"""
|
||
if _cookie_browser_port is None or _captcha_http_port is None or _captcha_ocr_port is None:
|
||
raise RuntimeError("站点登录端口尚未由启动组合根装配")
|
||
return _cookie_browser_port, _captcha_http_port, _captcha_ocr_port
|
||
|
||
|
||
class CookieLoginFormMixin:
|
||
"""
|
||
提供跨站点登录表单、验证码和页面错误识别的共享能力。
|
||
"""
|
||
|
||
# 站点登录界面元素XPATH
|
||
_SITE_LOGIN_XPATH = {
|
||
"username": [
|
||
'//input[@name="username"]',
|
||
'//input[@name="user"]',
|
||
'//input[@name="user_email"]',
|
||
'//input[@name="txt_user"]',
|
||
'//input[@name="txt_email"]',
|
||
'//input[@name="email"]',
|
||
'//input[@id="form_item_username"]',
|
||
'//input[@id="username"]',
|
||
'//input[@id="user"]',
|
||
'//input[@id="email"]',
|
||
'//input[contains(@placeholder,"用户名")]',
|
||
'//input[contains(@placeholder,"邮箱")]',
|
||
(
|
||
'//input[not(translate(@type,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz")="hidden") and '
|
||
'(contains(translate(@name,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"user") or '
|
||
'contains(translate(@name,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"email") or '
|
||
'contains(translate(@id,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"user") or '
|
||
'contains(translate(@id,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"email"))]'
|
||
),
|
||
],
|
||
"password": [
|
||
'//input[@name="password"]',
|
||
'//input[@id="form_item_password"]',
|
||
'//input[@id="password"]',
|
||
'//input[@type="password"]',
|
||
'//form[.//input[@type="password"]][1]//input[@type="password"][1]',
|
||
],
|
||
"captcha": [
|
||
'//input[@name="imagestring"]',
|
||
'//input[@name="captcha"]',
|
||
'//input[@name="captcha_code"]',
|
||
'//input[@name="verifycode"]',
|
||
'//input[@name="verification_code"]',
|
||
'//input[@name="security_code"]',
|
||
'//input[@name="imagecode"]',
|
||
'//input[@id="form_item_captcha"]',
|
||
'//input[@placeholder="驗證碼"]',
|
||
'//input[contains(@placeholder,"验证码")]',
|
||
(
|
||
'//input[not(translate(@type,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz")="hidden") and '
|
||
'(contains(translate(@name,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"captcha") or '
|
||
'contains(translate(@name,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"verify") or '
|
||
'contains(translate(@id,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"captcha") or '
|
||
'contains(translate(@id,"ABCDEFGHIJKLMNOPQRSTUVWXYZ",'
|
||
'"abcdefghijklmnopqrstuvwxyz"),"verify"))]'
|
||
),
|
||
],
|
||
"captcha_img": [
|
||
'//img[@alt="captcha"]/@src',
|
||
'//img[@alt="CAPTCHA"]/@src',
|
||
'//img[@alt="SECURITY CODE"]/@src',
|
||
'//img[@id="LAY-user-get-vercode"]/@src',
|
||
'//img[contains(@src,"/api/getCaptcha")]/@src',
|
||
(
|
||
'//img[contains(translate(concat(@alt," ",@title," ",@class," ",@src),'
|
||
'"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz"),"captcha")]/@src'
|
||
),
|
||
(
|
||
'//img[contains(translate(concat(@alt," ",@title," ",@class," ",@src),'
|
||
'"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz"),"verify")]/@src'
|
||
),
|
||
],
|
||
"submit": [
|
||
'//input[@type="submit"]',
|
||
'//button[@type="submit"]',
|
||
'//button[@lay-filter="login"]',
|
||
'//button[@lay-filter="formLogin"]',
|
||
'//input[@type="button"][@value="登录"]',
|
||
'//input[@id="submit-btn"]',
|
||
'//button[contains(translate(normalize-space(.),"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz"),"login")]',
|
||
'//button[contains(normalize-space(.),"登录") or contains(normalize-space(.),"登入") or contains(normalize-space(.),"提交")]',
|
||
'//input[contains(translate(@value,"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz"),"login")]',
|
||
],
|
||
"error": [
|
||
"//table[@class='main']//td[@class='text']/text()",
|
||
'//*[@role="alert"]//text()',
|
||
'//*[contains(translate(@class,"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz"),"error")]//text()',
|
||
'//*[contains(translate(@class,"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz"),"alert")]//text()',
|
||
'//*[contains(translate(@class,"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz"),"message")]//text()',
|
||
],
|
||
"remember": [
|
||
'//input[@type="checkbox"][contains(@name,"remember") or contains(@id,"remember")]',
|
||
'//*[@role="checkbox"][contains(.,"保持登录") or contains(.,"记住我") or contains(.,"自动登录")]',
|
||
],
|
||
"twostep": [
|
||
'//input[@name="two_step_code"]',
|
||
'//input[@name="2fa_secret"]',
|
||
'//input[@name="otp"]',
|
||
]
|
||
}
|
||
|
||
@classmethod
|
||
def _first_xpath(cls, html: etree._Element, key: str) -> Optional[str]:
|
||
"""
|
||
返回页面上第一个命中的字段 XPath。
|
||
"""
|
||
for xpath in cls._SITE_LOGIN_XPATH.get(key, []):
|
||
if html.xpath(xpath):
|
||
return xpath
|
||
return None
|
||
|
||
@classmethod
|
||
def _find_login_fields(
|
||
cls,
|
||
html: etree._Element,
|
||
) -> tuple[Optional[str], Optional[str]]:
|
||
"""
|
||
按常见命名、表单结构和输入类型推断用户名与密码字段。
|
||
"""
|
||
username_xpath = cls._first_xpath(html, "username")
|
||
password_xpath = cls._first_xpath(html, "password")
|
||
form_xpath = "//form[.//input[translate(@type,\"ABCDEFGHIJKLMNOPQRSTUVWXYZ\",\"abcdefghijklmnopqrstuvwxyz\")=\"password\"]][1]"
|
||
form_nodes = html.xpath(form_xpath)
|
||
if form_nodes:
|
||
login_form = form_nodes[0]
|
||
|
||
def belongs_to_login_form(xpath: Optional[str]) -> bool:
|
||
"""
|
||
判断全局候选是否属于含密码字段的登录表单。
|
||
"""
|
||
if not xpath:
|
||
return False
|
||
nodes = html.xpath(xpath)
|
||
return bool(nodes) and nodes[0] in login_form.iter()
|
||
|
||
if not belongs_to_login_form(username_xpath):
|
||
username_xpath = None
|
||
if not belongs_to_login_form(password_xpath):
|
||
password_xpath = None
|
||
|
||
if not password_xpath:
|
||
password_xpath = f'{form_xpath}//input[@type="password"][1]'
|
||
if not username_xpath:
|
||
scoped_candidates: tuple[str, ...] = (
|
||
f'{form_xpath}//input[@name="username"][1]',
|
||
f'{form_xpath}//input[@name="user"][1]',
|
||
f'{form_xpath}//input[@name="user_email"][1]',
|
||
f'{form_xpath}//input[@name="txt_user"][1]',
|
||
f'{form_xpath}//input[@name="txt_email"][1]',
|
||
f'{form_xpath}//input[@name="email"][1]',
|
||
f'{form_xpath}//input[@id="username"][1]',
|
||
f'{form_xpath}//input[@id="user"][1]',
|
||
f'{form_xpath}//input[@id="email"][1]',
|
||
f'{form_xpath}//input[@type="email"][1]',
|
||
f'{form_xpath}//input[@type="text"][1]',
|
||
f'{form_xpath}//input[not(@type) and not(@name="password")][1]',
|
||
)
|
||
username_xpath = next(
|
||
(candidate for candidate in scoped_candidates if html.xpath(candidate)),
|
||
None,
|
||
)
|
||
|
||
if password_xpath and username_xpath:
|
||
return username_xpath, password_xpath
|
||
|
||
if not password_xpath and html.xpath(f"{form_xpath}//input[@type='password']"):
|
||
password_xpath = f"{form_xpath}//input[@type='password'][1]"
|
||
if not username_xpath:
|
||
scoped_candidates = (
|
||
f"{form_xpath}//input[@type='email'][1]",
|
||
f"{form_xpath}//input[@type='text'][1]",
|
||
f"{form_xpath}//input[not(@type) and not(@name='password')][1]",
|
||
)
|
||
username_xpath = next(
|
||
(candidate for candidate in scoped_candidates if html.xpath(candidate)),
|
||
None,
|
||
)
|
||
if not password_xpath and html.xpath("//input[@type='password'][1]"):
|
||
password_xpath = "//input[@type='password'][1]"
|
||
if not username_xpath:
|
||
fallback_candidates = (
|
||
"//input[@type='email'][1]",
|
||
"//input[@type='text'][1]",
|
||
)
|
||
username_xpath = next(
|
||
(candidate for candidate in fallback_candidates if html.xpath(candidate)),
|
||
None,
|
||
)
|
||
return username_xpath, password_xpath
|
||
|
||
@classmethod
|
||
def _find_captcha_source(
|
||
cls,
|
||
html: etree._Element,
|
||
) -> tuple[Optional[str], Optional[str]]:
|
||
"""
|
||
查找验证码输入字段及图片地址，兼容常见命名和站点自定义 class。
|
||
"""
|
||
captcha_xpath = cls._first_xpath(html, "captcha")
|
||
if not captcha_xpath:
|
||
return None, None
|
||
for image_xpath in cls._SITE_LOGIN_XPATH.get("captcha_img", []):
|
||
values = html.xpath(image_xpath)
|
||
if values and isinstance(values[0], str):
|
||
return captcha_xpath, values[0]
|
||
return captcha_xpath, None
|
||
|
||
@classmethod
|
||
def _page_issue(cls, html_text: str, page_url: str) -> Optional[str]:
|
||
"""
|
||
把站点不可用和人机挑战归类为 Agent 可采取行动的提示。
|
||
"""
|
||
text = " ".join((html_text or "").split())
|
||
lowered = text.lower()
|
||
url_lowered = (page_url or "").lower()
|
||
if any(
|
||
marker in lowered or marker in url_lowered
|
||
for marker in (
|
||
"cf-chl-",
|
||
"cloudflare",
|
||
"cf-turnstile",
|
||
"turnstile",
|
||
"challenge-platform",
|
||
"captcha verification token is missing",
|
||
)
|
||
):
|
||
return "站点需要完成 Cloudflare/人机验证，无法自动登录，请手动登录后提供 Cookie"
|
||
if any(
|
||
marker in lowered or marker in url_lowered
|
||
for marker in (
|
||
"404 not found",
|
||
"502 bad gateway",
|
||
"503 service unavailable",
|
||
"site not found",
|
||
"没有找到站点",
|
||
"站点不存在",
|
||
"域名未绑定",
|
||
"无法连接到站点",
|
||
)
|
||
):
|
||
return "站点不可用或域名未绑定源站，请先确认站点地址和网络状态"
|
||
return None
|
||
|
||
@classmethod
|
||
def _error_message(cls, html_text: str) -> str:
|
||
"""
|
||
提取登录页可见错误，并去除重复或过长的 HTML 文本。
|
||
"""
|
||
html = etree.HTML(html_text or "")
|
||
if html is None:
|
||
return ""
|
||
messages: list[str] = []
|
||
for xpath in cls._SITE_LOGIN_XPATH.get("error", []):
|
||
for value in html.xpath(xpath):
|
||
text = " ".join(str(value).split())
|
||
if text and text not in messages:
|
||
messages.append(text)
|
||
return "；".join(messages)[:500]
|
||
|
||
@staticmethod
|
||
def _page_user_agent(page: Any) -> str:
|
||
"""
|
||
读取当前页面 User-Agent，浏览器实现不支持时返回空字符串。
|
||
"""
|
||
try:
|
||
return str(page.evaluate("() => window.navigator.userAgent") or "")
|
||
except Exception:
|
||
return ""
|
||
|
||
@staticmethod
|
||
def _click_with_fallback(page: Any, selector: str, timeout: int = 5000) -> None:
|
||
"""
|
||
依次尝试普通、强制、元素和 XPath JavaScript 点击。
|
||
"""
|
||
first_error: Optional[Exception] = None
|
||
try:
|
||
page.click(selector)
|
||
return
|
||
except Exception as error:
|
||
first_error = error
|
||
try:
|
||
page.click(selector, timeout=timeout, force=True)
|
||
return
|
||
except Exception:
|
||
pass
|
||
try:
|
||
element = page.query_selector(selector)
|
||
if element is not None and hasattr(element, "click"):
|
||
element.click(timeout=timeout, force=True)
|
||
return
|
||
except Exception:
|
||
pass
|
||
try:
|
||
clicked = page.evaluate(
|
||
"""
|
||
(xpath) => {
|
||
const node = document.evaluate(
|
||
xpath, document, null, XPathResult.FIRST_ORDERED_NODE_TYPE, null
|
||
).singleNodeValue;
|
||
if (!node) return false;
|
||
node.click();
|
||
return true;
|
||
}
|
||
""",
|
||
selector,
|
||
)
|
||
if clicked is True or clicked is None:
|
||
return
|
||
except Exception:
|
||
pass
|
||
raise first_error or RuntimeError("无法点击页面元素")
|
||
|
||
@classmethod
|
||
def _refresh_captcha(cls, page: Any, html: etree._Element) -> bool:
|
||
"""
|
||
点击验证码图片或调用页面 reload，尽量获取下一张验证码。
|
||
"""
|
||
for image_xpath in cls._SITE_LOGIN_XPATH.get("captcha_img", []):
|
||
if not html.xpath(image_xpath):
|
||
continue
|
||
selector = image_xpath[:-5] if image_xpath.endswith("/@src") else image_xpath
|
||
try:
|
||
cls._click_with_fallback(page, selector, timeout=3000)
|
||
return True
|
||
except Exception:
|
||
continue
|
||
reload_page = getattr(page, "reload", None)
|
||
if callable(reload_page):
|
||
try:
|
||
reload_page(wait_until="domcontentloaded", timeout=10000)
|
||
return True
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
class CookieHelper(CookieLoginFormMixin):
|
||
"""
|
||
处理站点登录表单、验证码和 Cookie 获取流程。
|
||
"""
|
||
|
||
_MAX_CAPTCHA_ATTEMPTS = 3
|
||
|
||
@staticmethod
|
||
def get_page_content(page: CookiePageContentPort, retries: int = 3,
|
||
interval: float = 1.0) -> Optional[str]:
|
||
"""
|
||
获取页面源码，页面跳转中（如登录前后的重定向）会导致 page.content() 抛出
|
||
"Unable to retrieve content because the page is navigating" 异常，等待加载完成后重试。
|
||
|
||
:param page: 浏览器页面
|
||
:param retries: 最大重试次数
|
||
:param interval: 重试间隔（秒）
|
||
:return: 页面源码
|
||
"""
|
||
for i in range(retries):
|
||
# 等待加载失败不代表源码不可读取，最后一次等待失败时仍尝试直接获取源码
|
||
try:
|
||
page.wait_for_load_state("domcontentloaded", timeout=10 * 1000)
|
||
except Exception as e:
|
||
if i < retries - 1:
|
||
logger.warning(f"等待页面加载完成失败：{str(e)}，{interval}秒后重试 ({i + 1}/{retries - 1})")
|
||
time.sleep(interval)
|
||
continue
|
||
logger.warning(f"等待页面加载完成失败：{str(e)}，尝试直接获取源码")
|
||
try:
|
||
return page.content()
|
||
except Exception as e:
|
||
if i >= retries - 1:
|
||
logger.error(f"获取页面源码失败：{str(e)}")
|
||
return None
|
||
logger.warning(f"获取页面源码失败：{str(e)}，{interval}秒后重试 ({i + 1}/{retries - 1})")
|
||
time.sleep(interval)
|
||
return None
|
||
|
||
@staticmethod
|
||
def parse_cookies(cookies: list[object]) -> str:
|
||
"""
|
||
将浏览器 Cookie 列表转成请求头字符串，并忽略不完整条目。
|
||
|
||
:param cookies: 浏览器返回的 Cookie 条目；非字典或缺少名称、值的条目会跳过
|
||
:return: 用分号分隔的 Cookie 请求头；没有有效条目时返回空字符串
|
||
"""
|
||
if not cookies:
|
||
return ""
|
||
values: list[str] = []
|
||
for cookie in cookies:
|
||
if not isinstance(cookie, dict):
|
||
continue
|
||
name = cookie.get("name")
|
||
value = cookie.get("value")
|
||
if name is not None and value is not None:
|
||
values.append(f"{name}={value}")
|
||
return "; ".join(values) + ("; " if values else "")
|
||
|
||
@staticmethod
|
||
def _find_login_page_url(html: etree._Element, current_url: str) -> Optional[str]:
|
||
"""
|
||
从首页查找同源登录入口，避免把账号密码提交到跨域页面。
|
||
"""
|
||
login_hrefs = html.xpath(
|
||
"//a["
|
||
"contains(translate(@href, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'login')"
|
||
" or contains(translate(@href, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'signin')"
|
||
"]/@href"
|
||
)
|
||
current = urlparse(current_url)
|
||
for href in login_hrefs:
|
||
login_url = urljoin(current_url, href)
|
||
target = urlparse(login_url)
|
||
if target.scheme in ("http", "https") and \
|
||
target.scheme == current.scheme and target.netloc == current.netloc:
|
||
return str(login_url)
|
||
return None
|
||
|
||
def get_site_cookie_ua(self,
|
||
url: str,
|
||
username: str,
|
||
password: str,
|
||
two_step_code: Optional[str] = None,
|
||
proxies: Optional[dict[str, object]] = None,
|
||
timeout: Optional[int] = None) -> Tuple[Optional[str], Optional[str], str]:
|
||
"""
|
||
获取站点 Cookie、User-Agent 和兼容错误消息。
|
||
|
||
:param url: 站点登录地址
|
||
:param username: 登录用户名
|
||
:param password: 登录密码
|
||
:param two_step_code: 可选的二步验证码或密钥
|
||
:param proxies: 浏览器代理配置；None 表示不指定代理
|
||
:param timeout: 浏览器超时时间，单位为秒；None 使用实现默认值
|
||
:return: Cookie、User-Agent 和错误说明；失败时保留原有空值与错误文本
|
||
"""
|
||
return self._get_site_cookie_ua_impl(
|
||
url=url,
|
||
username=username,
|
||
password=password,
|
||
two_step_code=two_step_code,
|
||
proxies=proxies,
|
||
timeout=timeout,
|
||
)
|
||
|
||
def _get_site_cookie_ua_impl(self,
|
||
url: str,
|
||
username: str,
|
||
password: str,
|
||
two_step_code: Optional[str] = None,
|
||
proxies: Optional[dict[str, object]] = None,
|
||
timeout: Optional[int] = None) -> Tuple[Optional[str], Optional[str], str]:
|
||
"""
|
||
获取站点cookie和ua。
|
||
|
||
:param url: 站点地址
|
||
:param username: 用户名
|
||
:param password: 密码
|
||
:param two_step_code: 二步验证码或密钥
|
||
:param proxies: 代理
|
||
:param timeout: 超时时间
|
||
:return: Cookie、User-Agent 与错误说明组成的三元组
|
||
"""
|
||
|
||
def __page_handler(page: Any) -> CookieResult:
|
||
"""
|
||
在受控浏览器页面内完成登录并返回当前会话凭据。
|
||
"""
|
||
html_text = self.get_page_content(page)
|
||
if not html_text:
|
||
return None, None, "获取源码失败"
|
||
html = etree.HTML(html_text)
|
||
if html is None:
|
||
return None, None, "解析网页源码失败"
|
||
issue = self._page_issue(html_text, getattr(page, "url", "") or url)
|
||
if issue:
|
||
return None, None, issue
|
||
|
||
username_xpath, password_xpath = self._find_login_fields(html)
|
||
if not username_xpath or not password_xpath:
|
||
login_url = self._find_login_page_url(html, getattr(page, "url", "") or url)
|
||
if login_url:
|
||
try:
|
||
page.goto(
|
||
login_url,
|
||
wait_until="domcontentloaded",
|
||
timeout=(timeout or 60) * 1000,
|
||
)
|
||
except Exception as error:
|
||
return None, None, f"打开登录页面失败：{str(error)}"
|
||
html_text = self.get_page_content(page)
|
||
html = etree.HTML(html_text) if html_text else None
|
||
if html is None:
|
||
return None, None, "解析网页源码失败"
|
||
issue = self._page_issue(html_text or "", getattr(page, "url", "") or url)
|
||
if issue:
|
||
return None, None, issue
|
||
username_xpath, password_xpath = self._find_login_fields(html)
|
||
if not username_xpath or not password_xpath:
|
||
try:
|
||
page.wait_for_selector("xpath=//input[@type='password']", timeout=5000)
|
||
except Exception:
|
||
pass
|
||
latest_text = self.get_page_content(page)
|
||
html = etree.HTML(latest_text) if latest_text else None
|
||
if html is not None:
|
||
html_text = latest_text or html_text
|
||
username_xpath, password_xpath = self._find_login_fields(html)
|
||
if not username_xpath:
|
||
return None, None, "未找到用户名输入框，登录表单字段无法识别"
|
||
if not password_xpath:
|
||
return None, None, "未找到密码输入框，登录表单字段无法识别"
|
||
|
||
otp_code = TwoFactorAuth(two_step_code).get_code()
|
||
twostep_xpath = self._first_xpath(html, "twostep") if otp_code else None
|
||
captcha_xpath, captcha_img_url = self._find_captcha_source(html)
|
||
if captcha_xpath and not captcha_img_url:
|
||
return None, None, "检测到验证码输入框，但未找到验证码图片"
|
||
submit_xpath = self._first_xpath(html, "submit")
|
||
if not submit_xpath:
|
||
return None, None, "未找到登录按钮，请在登录页提供可提交的按钮"
|
||
|
||
try:
|
||
page.wait_for_selector(submit_xpath)
|
||
page.fill(username_xpath, username)
|
||
page.fill(password_xpath, password)
|
||
for xpath in self._SITE_LOGIN_XPATH.get("remember", []):
|
||
remember_element = page.query_selector(xpath)
|
||
if not remember_element:
|
||
continue
|
||
try:
|
||
checked = remember_element.get_attribute("aria-checked")
|
||
if checked is None:
|
||
checked = "true" if remember_element.is_checked() else "false"
|
||
if checked != "true":
|
||
remember_element.click(timeout=3000)
|
||
break
|
||
except Exception as error:
|
||
logger.warning(f"勾选记住登录选项失败：{str(error)}，尝试下一候选")
|
||
if twostep_xpath:
|
||
page.fill(twostep_xpath, otp_code)
|
||
|
||
if captcha_xpath and captcha_img_url:
|
||
captcha_element = page.query_selector(captcha_xpath)
|
||
if captcha_element is None or captcha_element.is_visible():
|
||
captcha = ""
|
||
for attempt in range(self._MAX_CAPTCHA_ATTEMPTS):
|
||
current_text = self.get_page_content(page)
|
||
current_html = etree.HTML(current_text) if current_text else html
|
||
if current_html is None:
|
||
current_html = html
|
||
if current_html is None:
|
||
continue
|
||
current_xpath, current_image = self._find_captcha_source(current_html)
|
||
captcha_xpath = current_xpath or captcha_xpath
|
||
captcha_img_url = current_image or captcha_img_url
|
||
code_url = self.__get_captcha_url(
|
||
getattr(page, "url", "") or url,
|
||
captcha_img_url,
|
||
)
|
||
cookie = self.parse_cookies(page.context.cookies())
|
||
ua = self._page_user_agent(page)
|
||
captcha = self.__get_captcha_text(
|
||
cookie=cookie,
|
||
ua=ua,
|
||
code_url=code_url,
|
||
)
|
||
if captcha:
|
||
logger.info("验证码已完成识别，第 %s/%s 次尝试", attempt + 1, self._MAX_CAPTCHA_ATTEMPTS)
|
||
break
|
||
if attempt < self._MAX_CAPTCHA_ATTEMPTS - 1:
|
||
self._refresh_captcha(page, current_html)
|
||
time.sleep(0.5)
|
||
if not captcha:
|
||
return None, None, f"验证码识别失败，已尝试 {self._MAX_CAPTCHA_ATTEMPTS} 次，请手动刷新验证码后重试"
|
||
page.fill(captcha_xpath, captcha)
|
||
|
||
self._click_with_fallback(page, submit_xpath)
|
||
page.wait_for_load_state("networkidle", timeout=30 * 1000)
|
||
except Exception as error:
|
||
logger.error(f"仿真登录失败：{str(error)}")
|
||
return None, None, f"仿真登录失败：{str(error)}"
|
||
|
||
current_url = (getattr(page, "url", "") or "").lower()
|
||
if "verify" in current_url:
|
||
if not otp_code:
|
||
return None, None, "站点要求二次验证码，请提供二步验证码或密钥"
|
||
html_text = self.get_page_content(page)
|
||
html = etree.HTML(html_text) if html_text else None
|
||
verify_xpath = self._first_xpath(html, "twostep") if html is not None else None
|
||
if verify_xpath:
|
||
try:
|
||
page.fill(verify_xpath, TwoFactorAuth(two_step_code).get_code())
|
||
self._click_with_fallback(page, submit_xpath)
|
||
page.wait_for_load_state("networkidle", timeout=30 * 1000)
|
||
except Exception as error:
|
||
logger.error(f"二次验证码输入失败：{str(error)}")
|
||
return None, None, f"二次验证码输入失败：{str(error)}"
|
||
|
||
first_failure_html: Optional[str] = None
|
||
for index in range(3):
|
||
if index:
|
||
time.sleep(2)
|
||
latest_text = self.get_page_content(page)
|
||
if not latest_text:
|
||
continue
|
||
if SiteUtils.is_logged_in(latest_text):
|
||
return self.parse_cookies(page.context.cookies()), self._page_user_agent(page), ""
|
||
if first_failure_html is None:
|
||
first_failure_html = latest_text
|
||
failure_issue = self._page_issue(
|
||
latest_text,
|
||
getattr(page, "url", "") or url,
|
||
)
|
||
if failure_issue or self._error_message(latest_text):
|
||
first_failure_html = latest_text
|
||
break
|
||
if not first_failure_html:
|
||
return None, None, "获取登录结果源码失败"
|
||
failure_issue = self._page_issue(
|
||
first_failure_html,
|
||
getattr(page, "url", "") or url,
|
||
)
|
||
if failure_issue:
|
||
return None, None, failure_issue
|
||
error_message = self._error_message(first_failure_html)
|
||
return None, None, f"登录失败：{error_message}" if error_message else "登录失败：页面未提供具体原因"
|
||
|
||
if not url or not username or not password:
|
||
return None, None, "参数错误"
|
||
|
||
browser_port, _, _ = _require_cookie_ports()
|
||
return browser_port.action(url=url,
|
||
callback=__page_handler,
|
||
proxies=proxies,
|
||
timeout=timeout)
|
||
|
||
@staticmethod
|
||
def __get_captcha_text(cookie: str, ua: str, code_url: str) -> str:
|
||
"""
|
||
识别验证码图片的内容。
|
||
"""
|
||
if not code_url:
|
||
return ""
|
||
_, http_port, ocr_port = _require_cookie_ports()
|
||
content = http_port.fetch(url=code_url, cookie=cookie, ua=ua)
|
||
if not content:
|
||
return ""
|
||
return ocr_port.recognize(content)
|
||
|
||
@staticmethod
|
||
def __get_captcha_url(siteurl: str, imageurl: str) -> str:
|
||
"""
|
||
按页面地址解析验证码图片地址，兼容绝对、根相对和路径相对 URL。
|
||
"""
|
||
if not siteurl or not imageurl:
|
||
return ""
|
||
return urljoin(siteurl, imageurl)
|