Cookie 管理与模拟登录

pip install requests>=2.31 lxml>=5.0 redis>=5.0 flask>=3.0

# Python 3.10+
import requests

# 不使用 Session —— 每次请求独立，Cookie 不会保留
r1 = requests.get("https://httpbin.org/cookies/set/foo/bar")
r2 = requests.get("https://httpbin.org/cookies")
print(r2.json())  # {"cookies": {}}  —— Cookie 丢失！

# 使用 Session —— 自动管理 Cookie
session = requests.Session()
r1 = session.get("https://httpbin.org/cookies/set/foo/bar")
r2 = session.get("https://httpbin.org/cookies")
print(r2.json())  # {"cookies": {"foo": "bar"}}  —— Cookie 保留！

# Python 3.10+
import json
import os
from pathlib import Path
import requests
import logging

logger = logging.getLogger(__name__)


class CookiePersistence:
    """Cookie 持久化管理 —— 支持保存到文件、从文件恢复、自动校验"""

    def __init__(self, session: requests.Session, cookie_file: str = "cookies.json"):
        self.session = session
        self.cookie_file = Path(cookie_file)

    def save(self) -> None:
        """将 Session 中的 Cookie 保存为 JSON 文件"""
        cookies: list[dict[str, str]] = []
        for cookie in self.session.cookies:
            cookies.append({
                "name": cookie.name,
                "value": cookie.value,
                "domain": cookie.domain,
                "path": cookie.path,
            })
        self.cookie_file.write_text(json.dumps(cookies, indent=2, ensure_ascii=False))
        logger.info(f"Cookie 已保存到 {self.cookie_file}（{len(cookies)} 条）")

    def load(self) -> bool:
        """从 JSON 文件恢复 Cookie 到 Session"""
        if not self.cookie_file.exists():
            logger.warning(f"Cookie 文件不存在: {self.cookie_file}")
            return False

        cookies: list[dict] = json.loads(self.cookie_file.read_text())
        for cookie in cookies:
            self.session.cookies.set(
                cookie["name"],
                cookie["value"],
                domain=cookie.get("domain"),
                path=cookie.get("path"),
            )
        logger.info(f"已从 {self.cookie_file} 恢复 {len(cookies)} 条 Cookie")
        return True

    def is_valid(self, check_url: str) -> bool:
        """向 check_url 发送请求，通过响应判断 Cookie 是否有效"""
        try:
            resp = self.session.get(check_url, allow_redirects=False, timeout=10)
            # 有效的 Cookie 不应被重定向到登录页
            return resp.status_code == 200 and "login" not in resp.url.lower()
        except requests.RequestException:
            return False

# Python 3.10+
import requests
from lxml import etree
import logging

logger = logging.getLogger(__name__)


class FormLoginClient:
    """通用表单登录客户端"""

    def __init__(self, login_page_url: str, login_action_url: str):
        self.session = requests.Session()
        self.login_page_url = login_page_url
        self.login_action_url = login_action_url
        self._setup_headers()

    def _setup_headers(self) -> None:
        self.session.headers.update({
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/120.0.0.0 Safari/537.36"
            ),
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        })

    def _extract_csrf_token(self, html: str, token_field_name: str = "authenticity_token") -> str | None:
        """
        从登录页面 HTML 中提取 CSRF Token。
        使用 XPath 为主，正则表达式为备选方案。
        """
        # 策略1：XPath 解析
        selector = etree.HTML(html)
        tokens = selector.xpath(f'//input[@name="{token_field_name}"]/@value')
        if tokens:
            return str(tokens[0])

        # 策略2：正则表达式兜底
        import re
        pattern = rf'name="{token_field_name}"\s+value="([^"]+)"'
        match = re.search(pattern, html)
        if match:
            return match.group(1)

        return None

    def login(self, username: str, password: str,
              extra_fields: dict[str, str] | None = None) -> bool:
        """
        执行表单登录。

        Args:
            username: 用户名/邮箱
            password: 密码
            extra_fields: 额外的表单字段（如 commit、utf8 等固定值）

        Returns:
            bool: 登录是否成功
        """
        # 步骤1：访问登录页面，获取初始 Cookie 和 CSRF Token
        logger.info("正在访问登录页面...")
        response = self.session.get(self.login_page_url)
        csrf_token = self._extract_csrf_token(response.text)

        if csrf_token is None:
            logger.error("无法提取 CSRF Token，请检查页面结构")
            return False
        logger.info(f"CSRF Token 获取成功: {csrf_token[:20]}...")

        # 步骤2：构造登录表单
        form_data: dict[str, str] = {
            "login": username,
            "password": password,
            "authenticity_token": csrf_token,
        }
        if extra_fields:
            form_data.update(extra_fields)

        # 步骤3：提交登录请求
        logger.info("正在提交登录表单...")
        response = self.session.post(
            self.login_action_url,
            data=form_data,
            allow_redirects=True,
        )

        # 步骤4：验证登录结果
        return self._verify_login(response)

    def _verify_login(self, response: requests.Response) -> bool:
        """通过多种方式验证登录是否成功"""
        # 方式1：检查特定 Cookie 是否出现
        if "logged_in" in self.session.cookies:
            return True
        # 方式2：检查页面内容中的登录标识
        if "Sign out" in response.text or "logout" in response.text.lower():
            return True
        return False

    def fetch_protected(self, url: str) -> str | None:
        """获取需要登录才能访问的页面"""
        resp = self.session.get(url)
        return resp.text if resp.status_code == 200 else None

# Python 3.10+
client = FormLoginClient(
    login_page_url="https://github.com/login",
    login_action_url="https://github.com/session",
)

success = client.login(
    username="your_email@example.com",
    password="your_password",
    extra_fields={"commit": "Sign in", "utf8": "✓"},
)

if success:
    print("登录成功！")
    # 爬取登录后才能访问的个人设置页
    profile_html = client.fetch_protected("https://github.com/settings/profile")
else:
    print("登录失败，请检查用户名/密码或分析登录接口变化")

# Python 3.10+
import requests

def parse_cookie_string(cookie_str: str) -> dict[str, str]:
    """
    将浏览器复制的 Cookie 字符串解析为字典。
    输入示例: "name1=value1; name2=value2; name3=value3"
    """
    cookies: dict[str, str] = {}
    for item in cookie_str.split(";"):
        item = item.strip()
        if "=" in item:
            key, _, value = item.partition("=")
            cookies[key.strip()] = value.strip()
    return cookies


# 从浏览器复制的 Cookie 字符串
cookie_string = "logged_in=yes; dotcom_user=zhang; user_session=xyz789"

session = requests.Session()
session.headers.update({
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
})

# 方式1：通过 cookies 参数传递
cookies_dict = parse_cookie_string(cookie_string)
resp = session.get("https://github.com/settings/profile", cookies=cookies_dict)

# 方式2：直接注入到 Session（推荐，后续请求自动携带）
session.cookies.update(cookies_dict)
resp = session.get("https://github.com/settings/profile")
print(f"状态码: {resp.status_code}，页面长度: {len(resp.text)}")

# Python 3.10+
import requests

# === JWT Token 认证 ===
session = requests.Session()
session.headers.update({
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
    "User-Agent": "Mozilla/5.0 ...",
})

# 直接请求 API，无需 Cookie
resp = session.get("https://api.example.com/user/profile")
print(resp.json())

# === GitHub Personal Access Token 认证 ===
# 在 GitHub Settings > Developer settings > Personal access tokens 创建
session = requests.Session()
session.headers.update({
    "Authorization": "token ghp_xxxxxxxxxxxxxxxxxxxx",
    "Accept": "application/vnd.github+json",
})
resp = session.get("https://api.github.com/user")
print(resp.json())

# Python 3.10+
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import requests


def login_with_selenium_and_export_cookies(
    login_url: str, username: str, password: str
) -> dict[str, str]:
    """
    使用 Selenium 完成登录（包括验证码处理），然后导出 Cookie 供 requests 使用。
    典型应用场景：requests 无法处理的验证码，由 Selenium 人工或半自动处理。
    """
    options = webdriver.ChromeOptions()
    options.add_argument("--disable-blink-features=AutomationControlled")
    driver = webdriver.Chrome(options=options)
    wait = WebDriverWait(driver, 15)

    try:
        driver.get(login_url)

        # 填写表单
        username_input = wait.until(EC.presence_of_element_located((By.NAME, "username")))
        username_input.send_keys(username)
        driver.find_element(By.NAME, "password").send_keys(password)
        driver.find_element(By.CSS_SELECTOR, "button[type='submit']").click()

        # 如果有验证码，手动处理
        input("请在浏览器中完成验证码，然后按 Enter 继续...")

        # 导出 Cookie
        return {c["name"]: c["value"] for c in driver.get_cookies()}
    finally:
        driver.quit()


# 使用 Selenium 登录，导出 Cookie 后切换到 requests
cookies = login_with_selenium_and_export_cookies(
    "https://example.com/login", "username", "password"
)

session = requests.Session()
session.cookies.update(cookies)
# 后续全部使用 requests 高效爬取
resp = session.get("https://example.com/dashboard")

# Python 3.10+
import redis
import json
import random
import time
import logging

logger = logging.getLogger(__name__)


class CookieStore:
    """
    Cookie 存储管理 —— 基于 Redis Hash。

    数据结构:
        Key:    cookies:{website}
        Field:  {username}
        Value:  {"cookies": {...}, "status": "valid/invalid", "updated": 1700000000}
    """

    def __init__(self, host: str = "localhost", port: int = 6379,
                 db: int = 0, password: str | None = None):
        self.redis = redis.StrictRedis(
            host=host, port=port, db=db, password=password,
            decode_responses=True,
        )
        logger.info(f"Redis 连接成功: {host}:{port}")

    def _key(self, website: str) -> str:
        return f"cookies:{website}"

    def set(self, website: str, username: str, cookies: dict, status: str = "valid") -> None:
        """存储 Cookie"""
        value = json.dumps({
            "cookies": cookies,
            "status": status,
            "updated": int(time.time()),
        }, ensure_ascii=False)
        self.redis.hset(self._key(website), username, value)
        logger.info(f"Cookie 已存储: {website}:{username} ({status})")

    def get_random(self, website: str) -> dict | None:
        """随机获取一个有效 Cookie"""
        all_data = self.redis.hgetall(self._key(website))
        valid_accounts = [
            (username, json.loads(data))
            for username, data in all_data.items()
            if json.loads(data).get("status") == "valid"
        ]
        if not valid_accounts:
            logger.warning(f"{website} 无可用 Cookie")
            return None
        username, data = random.choice(valid_accounts)
        logger.info(f"随机获取: {website}:{username}")
        return data.get("cookies")

    def set_status(self, website: str, username: str, status: str) -> None:
        """更新 Cookie 状态"""
        data = self.get(website, username)
        if data:
            self.set(website, username, data["cookies"], status)

    def get(self, website: str, username: str) -> dict | None:
        """获取指定账号的 Cookie"""
        value = self.redis.hget(self._key(website), username)
        return json.loads(value) if value else None

    def delete(self, website: str, username: str) -> None:
        """删除指定账号的 Cookie"""
        self.redis.hdel(self._key(website), username)

    def count(self, website: str, status: str | None = None) -> int:
        """统计 Cookie 数量"""
        all_data = self.redis.hgetall(self._key(website))
        if status is None:
            return len(all_data)
        return sum(1 for v in all_data.values() if json.loads(v).get("status") == status)

    def get_all(self, website: str, status: str | None = None) -> dict[str, dict]:
        """获取指定网站的所有 Cookie"""
        all_data = self.redis.hgetall(self._key(website))
        result = {u: json.loads(v) for u, v in all_data.items()}
        if status:
            return {u: d for u, d in result.items() if d.get("status") == status}
        return result

# Python 3.10+
import requests
from abc import ABC, abstractmethod


class BaseChecker(ABC):
    """Cookie 检测器基类"""

    def __init__(self, website: str):
        self.website = website

    @abstractmethod
    def check(self, cookies: dict) -> bool:
        """检测 Cookie 是否有效，返回 True/False"""
        ...


class GithubChecker(BaseChecker):
    """GitHub Cookie 检测器 —— 请求个人设置页验证"""

    def __init__(self):
        super().__init__("github")
        self.check_url = "https://github.com/settings/profile"

    def check(self, cookies: dict) -> bool:
        try:
            resp = requests.get(
                self.check_url,
                cookies=cookies,
                allow_redirects=False,  # 不跟随重定向，通过状态码判断
                headers={"User-Agent": "Mozilla/5.0 ..."},
                timeout=10,
            )
            # 200 = 有效，302 = 被重定向到登录页 = 失效
            return resp.status_code == 200
        except requests.RequestException:
            return False


# 检测器注册表（可扩展更多网站）
CHECKER_MAP: dict[str, type[BaseChecker]] = {
    "github": GithubChecker,
}