多城市租房采集：地域定向代理实测与防漂移落地代码

import requests
import time
import random

PROXY_URL = "http://用户名:密码@代理host:端口"
TARGET_CITY = "上海"

HEADERS = {
    "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",
    "Referer": "https://fang.com/"
}

PROXIES = {
    "http": PROXY_URL,
    "https": PROXY_URL
}

session = requests.Session()
session.headers.update(HEADERS)

_last_check_ts = 0
CHECK_EVERY_SECONDS = 60  # 或按请求数，比如每30个请求查一次
_req_count = 0


def get_ip_city():
    """查当前代理IP归属城市。ip-api免费版有限速，别查太勤。"""
    try:
        r = session.get(
            "http://ip-api.com/json/?lang=zh-CN",
            proxies=PROXIES,
            timeout=5
        )
        data = r.json()
        if data.get("status") != "success":
            return False, data.get("message", "查询失败")
        city = data.get("city") or ""
        return TARGET_CITY in city, city
    except Exception as e:
        return False, str(e)


def switch_proxy():
    """
    这里必须接代理商的换IP接口，或者用支持粘性会话的代理参数。
    不换IP的话，重试大概率还是同一个节点。
    """
    # 示例思路：
    # new_ip = requests.get("代理商的提取API").text
    # 更新全局 PROXIES
    pass


def need_check(force=False):
    global _last_check_ts, _req_count
    now = time.time()

    if force or now - _last_check_ts > CHECK_EVERY_SECONDS:
        _last_check_ts = now
        _req_count = 0
        return True

    _req_count += 1
    return False


def is_valid_rent_html(html):
    """
    这里别只判断200。按你页面结构加：
    租金、户型、发布时间至少两个存在。
    """
    if not html:
        return False
    # 示例：return "租金" in html and "户型" in html
    return True


def fetch_rent(url, max_retry=3):
    for attempt in range(max_retry):
        if need_check(force=(attempt == 0)):
            ok, city = get_ip_city()
            if not ok:
                print(f"IP地域不对，当前：{city}，目标：{TARGET_CITY}，准备换IP")
                switch_proxy()
                time.sleep(random.uniform(1, 2))
                continue

        try:
            time.sleep(random.uniform(0.8, 2.0))
            resp = session.get(url, proxies=PROXIES, timeout=8)

            if resp.status_code in (403, 429, 502):
                print(f"触发风控或限流：{resp.status_code}，换IP重试")
                switch_proxy()
                time.sleep(random.uniform(2, 4))
                continue

            resp.raise_for_status()

            if not is_valid_rent_html(resp.text):
                print("页面字段不完整，当失败处理")
                switch_proxy()
                time.sleep(random.uniform(1, 3))
                continue

            return resp.text

        except Exception as e:
            print(f"请求异常：{e}，重试")
            switch_proxy()
            time.sleep(random.uniform(1, 3))

    return None


if __name__ == "__main__":
    rent_url = "目标租房房源链接"
    html = fetch_rent(rent_url)

    if html:
        print("采集成功，可以解析户型、租金、发布时间等字段")
    else:
        print("采集失败，检查代理城市、稳定性或换节点")