project/spides/spiders/taobao/snapshot_taobao_login.py

import time
import random
import signal
import sys
from DrissionPage import ChromiumPage, ChromiumOptions
import socket
from commons.conn_mysql import MySQLPoolOn2
from commons.Logger import logger
from commons.config import CHROME_PATH

MAX_PAGES = 5
WAIT_BETWEEN_PAGES = (8, 15)  # 页间等待时间范围（秒）
SCROLL_DELAY = (0.3, 0.8)  # 滚动延迟范围
CLICK_DELAY = (0.5, 1.2)  # 点击延迟范围
BROWSE_TIME = (5, 10)  # 浏览时间范围
chrome_path = CHROME_PATH


class TaobaoAutoCrawl:
    def __init__(self, account_name, ip, key_word):
        self.driver = None
        self.register_signal_handler()
        self.db = MySQLPoolOn2()
        self.account_name = account_name
        self.ip = ip
        self.keyword = key_word

    @staticmethod
    def _get_free_port():
        """获取一个当前可用的本地端口，供 Chrome 调试使用。"""
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(("127.0.0.1", 0))
            return s.getsockname()[1]

    def init_drissionpage(self):
        # 避免 auto_port 在部分环境下生成异常地址（无端口）导致 ChromiumPage 初始化失败
        co = ChromiumOptions().set_browser_path(chrome_path)
        #debug_port = self._get_free_port()
        #co.set_argument(f"--remote-debugging-port={debug_port}")

        co.set_user_data_path(f"./account_cache/taobao/{'TB_103'}")
        if self.ip:
            proxy = self.ip.strip()
            if not proxy.startswith(("http://", "https://")):
                proxy = f"http://{proxy}"
            co.set_argument(f"--proxy-server={proxy}")
        self.driver = ChromiumPage(co)

    def register_signal_handler(self):
        """ 非常必要，注册信号处理，确保状态保存"""

        def signal_handler(signum, frame):
            print("\n⚠️ 收到退出信号，正在保存状态并退出...")
            if self.driver:
                self.driver.quit()
            sys.exit(0)

        signal.signal(signal.SIGINT, signal_handler)
        # Windows 上可能不支持 SIGTERM，做兼容处理
        if hasattr(signal, "SIGTERM"):
            signal.signal(signal.SIGTERM, signal_handler)

    # ==================== 人工行为模拟 ====================
    def random_wait(self, min_sec, max_sec=None):
        """随机等待"""
        if max_sec is None:
            max_sec = min_sec
        time.sleep(random.uniform(min_sec, max_sec))

    def move_mouse_to_element(self, element):
        """移动鼠标到元素"""
        if not element:
            return
        try:
            # 优先使用 DrissionPage 推荐方式
            self.driver.actions.move_to(element)
        except Exception:
            # 兼容旧逻辑：按元素中心点移动（不同版本 move 参数可能不同）
            box = element.rect
            try:
                center_x = int(box.x + box.width / 2)
                center_y = int(box.y + box.height / 2)
            except Exception:
                center_x = int(box["x"] + box["width"] / 2)
                center_y = int(box["y"] + box["height"] / 2)
            try:
                self.driver.actions.move(center_x, center_y)
            except TypeError:
                # 某些版本仅支持关键字参数
                self.driver.actions.move(offset_x=center_x, offset_y=center_y)
        self.random_wait(0.2, 0.5)

    def human_type(self, element, text):
        """模拟人类输入"""
        for char in text:
            element.send_keys(char)
            time.sleep(random.uniform(0.1, 0.3))

    def login(self, username, password):
        self.driver.get("https://login.taobao.com")
        self.random_wait(5, 8)

        # 输入账号
        login_name = self.driver.ele("xpath=//input[@name='fm-login-id']", timeout=30)
        if login_name:
            self.move_mouse_to_element(login_name)
            self.human_type(login_name, username)
            self.random_wait(1, 3)

        # 输入密码
        login_pass = self.driver.ele("xpath=//input[@name='fm-login-password']", timeout=30)
        if login_pass:
            self.move_mouse_to_element(login_pass)
            self.human_type(login_pass, password)
            self.random_wait(1, 3)

        # 点击登录
        login_button = self.driver.ele("xpath=//button[text()='登录']", timeout=30)
        if login_button:
            self.move_mouse_to_element(login_button)
            login_button.click()
            self.random_wait(1, 3)

        # 处理同意按钮
        login_agree = self.driver.ele("xpath=//button[text()='同意']", timeout=5)
        if login_agree:
            self.move_mouse_to_element(login_agree)
            login_agree.click()
            self.random_wait(1, 3)

        # 等待登录结果
        self.random_wait(10, 20)
        # 检查是否登录成功
        user_info = self.driver.ele("xpath=//a[@class='site-nav-login-info-nick']", timeout=10)
        if user_info:
            print("登录成功！")
        else:
            print("登录失败，请检查账号密码或验证码")

    def get_search(self):
        url = "https://www.taobao.com"
        self.driver.get(url, timeout=30)
        time.sleep(20)
        time.sleep(random.uniform(3, 8))

        # 刷新一次，否则可能未找到登录状态
        self.driver.refresh()
        self.random_wait(5, 10)
        # login_name = self.driver.ele("xpath=//input[@name='fm-login-id']")
        # if login_name:
        #     self.login("aqwwer","wewetrv")
        #     self.driver.refresh()
        #     self.random_wait(5, 10)

        ele_iframe = self.driver.ele("xpath=//iframe[@id='baxia-dialog-content']")


        ele = self.driver.ele('xpath=//*[contains(@class,"site-nav-login-info-nick")]', timeout=30)

        if ele:
            cookies_list = self.driver.cookies()
            cookies_dict = {c['name']: c['value'] for c in cookies_list}
            timestamp = int(time.time())
            # 保存 cookie 到文件
            print(f"{self.account_name},获取 cookie 成功！")
            logger.info(f"{self.account_name},获取 cookie 成功！")

            self.random_wait(3, 5)
            return True
        else:
            return False

    def run(self):
        bool_login = False
        try:
            self.init_drissionpage()
            bool_login = self.get_search()
        except Exception as e:
            logger.exception(f"{self.account_name} 获取 cookie 异常: {e}")
        finally:
            if self.driver:
                return self.driver
        return bool(bool_login)


if __name__ == '__main__':
    account_name = "tb_account10"
    ip = ""
    keyword = "手机"
    taobao_crawl = TaobaoAutoCrawl(account_name, ip, keyword)
    taobao_crawl.run()