Python 爬虫实战：用 LokiProxy 动态住宅代理采集图书列表数据

一、爬取目标项目说明目标网站http://books.toscrape.com采集字段书名、价格、库存状态、详情页链接采集范围全站 50 页图书列表输出格式books.csv点击体验https://www.lokiproxy.com/该站点专为爬虫练习设计无登录、无验证码页面结构稳定。选它是为了把精力放在动态代理配置和采集逻辑上而不是和反爬较劲。二、动态住宅和短效 IP 有什么不同LokiProxy 里有多类产品动态住宅和短效住宅 IP不要混用配置入口和代码写法都不一样。短效住宅 IP从仪表盘批量提取多个IP:端口自己维护PROXY_POOL哪个失效了手动换。动态住宅只配一个网关地址每次请求由服务端自动换出口 IP代码里固定一个PROXY_URL即可不用建 IP 池。三、实现效果全站约 1000 条50 页 × 每页 20 本。单线程加随机延时本地三五分钟能跑完。四、准备工作4.1 安装依赖pip install requests lxml4.2 在仪表盘配置动态住宅登录 https://dashboard.lokiproxy.com/左侧选择动态住宅Rotating Residential确认可用流量充足认证方式二选一方式 1APIIP 白名单打开管理 IP 白名单加入本机公网 IP代理设置里国家/地区选Global或目标地区点击获取 API URL复制网关地址形如http://host:port白名单下通常无需账密方式 2用户名 密码切换到用户名:密码标签在代理设置里选国家、主机拼接为http://用户名:密码主机:端口五、为什么要用代理5.1 数据采集为什么需要代理5.2 为什么选择 LokiProxy100% 真实全球住宅 IP。 LokiProxy 强调 IP 来自真实家庭宽带不是机房段。对目标站来说更接近普通用户匿名性和通过率更好。代理池持续更新。 住宅 IP 会动态补充和轮换降低「一批 IP 很快全失效」的问题长期使用更省心。无限并发请求。 轮换住宅支持高并发后续做多线程、多任务采集时不必过早被「并发上限」卡住。六、实战代码6.1 代理与全局参数import requests import csv import time import random from lxml import etree # ---------- LokiProxy 动态住宅单一网关自动换 IP---------- # 方式 A用户名密码 LOKI_USER your_username LOKI_PASS your_password LOKI_HOST gate.lokiproxy.com # 以「获取 API URL」为准 LOKI_PORT 10000 PROXY_URL fhttp://{LOKI_USER}:{LOKI_PASS}{LOKI_HOST}:{LOKI_PORT} # 方式 BAPI 白名单IP 已加白名单按 API URL 填写无账密 # PROXY_URL http://gate.lokiproxy.com:10000 PROXIES { http: PROXY_URL, https: PROXY_URL, } BASE_URL http://books.toscrape.com TOTAL_PAGE 50 OUTPUT_FILE books.csv TIMEOUT 15 RETRY 3 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 ) }动态住宅不需要PROXY_POOL。一个网关地址贯穿全程出口 IP 由服务端轮换。6.2 验证代理与动态是否生效连发几次 httpbin看 IP 是否变化def check_proxy(): print([代理验证] 动态住宅出口检测) try: ips [] for i in range(3): resp requests.get( http://httpbin.org/ip, proxiesPROXIES, timeoutTIMEOUT ) ip resp.json()[origin] ips.append(ip) print(f 第 {i 1} 次 → {ip}) time.sleep(1) if len(set(ips)) 1: print( 出口 IP 已变化动态代理工作正常) else: print( 三次 IP 相同可能尚未轮换或样本太少可继续采集观察) return True except Exception as e: print(f[代理验证失败] {e}) print( 请检查白名单是否加本机 IP / 账密是否正确 / 流量是否用尽) return False6.3 带重试的请求网关偶发超时仍可能发生保留重试def fetch(url): for attempt in range(1, RETRY 1): try: resp requests.get( url, headersHEADERS, proxiesPROXIES, timeoutTIMEOUT ) resp.raise_for_status() resp.encoding utf-8 return resp.text except Exception as e: print(f[重试 {attempt}/{RETRY}] {e}) time.sleep(2 * attempt) return None每次fetch走同一PROXIES网关会在背后换 IP无需手动轮换。6.4 解析单页def parse_page(html): tree etree.HTML(html) books [] for item in tree.xpath(//article[classproduct_pod]): title item.xpath(.//h3/a/title) price item.xpath(.//p[classprice_color]/text()) stock item.xpath(.//p[classinstock availability]/text()) link item.xpath(.//h3/a/href) books.append({ title: title[0].strip() if title else , price: price[0].strip() if price else , stock: stock[-1].strip() if stock else , link: BASE_URL /catalogue/ link[0].lstrip(../) if link else , }) return books6.5 翻页采集第 1 页/index.html第 2 页起/catalogue/page-{n}.htmldef crawl_all(): all_books [] for page in range(1, TOTAL_PAGE 1): url BASE_URL /index.html if page 1 else BASE_URL f/catalogue/page-{page}.html print(f[采集] 第 {page}/{TOTAL_PAGE} 页) html fetch(url) if html is None: print(f[跳过] 第 {page} 页) continue books parse_page(html) all_books.extend(books) print(f 本页 {len(books)} 条累计 {len(all_books)} 条) time.sleep(random.uniform(0.5, 1.5)) return all_books页间 0.51.5 秒随机延时降低请求规律性也节省流量。6.6 写入 CSVdef save_csv(books, filename): with open(filename, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnames[title, price, stock, link]) writer.writeheader() writer.writerows(books) print(f[完成] 共 {len(books)} 条 → {filename})6.7 主入口def main(): print( * 50) print( LokiProxy 动态住宅 · 图书采集) print( * 50) if not check_proxy(): return books crawl_all() save_csv(books, OUTPUT_FILE) if __name__ __main__: main()七、完整源码import requests import csv import time import random from lxml import etree LOKI_USER your_username LOKI_PASS your_password LOKI_HOST gate.lokiproxy.com LOKI_PORT 10000 PROXY_URL fhttp://{LOKI_USER}:{LOKI_PASS}{LOKI_HOST}:{LOKI_PORT} PROXIES {http: PROXY_URL, https: PROXY_URL} BASE_URL http://books.toscrape.com TOTAL_PAGE 50 OUTPUT_FILE books.csv TIMEOUT 15 RETRY 3 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 ) } def check_proxy(): print([代理验证]) try: ips [] for i in range(3): r requests.get(http://httpbin.org/ip, proxiesPROXIES, timeoutTIMEOUT) ip r.json()[origin] ips.append(ip) print(f 第 {i 1} 次 → {ip}) time.sleep(1) print( 动态正常 if len(set(ips)) 1 else IP 未变化请继续观察) return True except Exception as e: print(f 失败: {e}) return False def fetch(url): for n in range(1, RETRY 1): try: r requests.get(url, headersHEADERS, proxiesPROXIES, timeoutTIMEOUT) r.raise_for_status() r.encoding utf-8 return r.text except Exception as e: print(f 重试 {n}/{RETRY}: {e}) time.sleep(2 * n) return None def parse_page(html): tree etree.HTML(html) books [] for item in tree.xpath(//article[classproduct_pod]): title item.xpath(.//h3/a/title) price item.xpath(.//p[classprice_color]/text()) stock item.xpath(.//p[classinstock availability]/text()) link item.xpath(.//h3/a/href) books.append({ title: title[0].strip() if title else , price: price[0].strip() if price else , stock: stock[-1].strip() if stock else , link: BASE_URL /catalogue/ link[0].lstrip(../) if link else , }) return books def crawl_all(): all_books [] for page in range(1, TOTAL_PAGE 1): url BASE_URL /index.html if page 1 else BASE_URL f/catalogue/page-{page}.html print(f[采集] 第 {page}/{TOTAL_PAGE} 页) html fetch(url) if html is None: continue books parse_page(html) all_books.extend(books) print(f 累计 {len(all_books)} 条) time.sleep(random.uniform(0.5, 1.5)) return all_books def save_csv(books, filename): with open(filename, w, newline, encodingutf-8-sig) as f: w csv.DictWriter(f, fieldnames[title, price, stock, link]) w.writeheader() w.writerows(books) print(f[完成] {len(books)} 条 → {filename}) def main(): if not check_proxy(): return save_csv(crawl_all(), OUTPUT_FILE) if __name__ __main__: main()复制后改LOKI_USER、LOKI_PASS、LOKI_HOST、LOKI_PORT或白名单方式的PROXY_URL即可运行。运行起来也非常简单直接在cmd中使用python xxx.py即可运行。查看采集的CSV结果结语动态住宅用起来很直接在 LokiProxy 仪表盘拿到一个网关地址填进requests的proxies后面换 IP 的事交给平台就行。 不用像短效 IP 那样批量提取、自己维护池子对只有动态住宅流量套餐的情况尤其合适——配置少、上手快把时间留给解析和采集逻辑。