使用教程-青果网络

我们青果网络长期服务AI训练数据采集、舆情监测、广告监测这类采集强度大的业务，这两年最深的体感是：采购问的问题变了。三年前问“池有多大、单价多少”，现在越来越多团队先问“资质全不全、出了问题能不能拿出记录、不同业务能不能隔离开”。判断轴的变化，先发生在客户的会议室里，其次才发生在服务商的产品页上。监管加码，为什么不只是合规部门的事？多数技术负责人有一个默认判断：数据采集合不合规，是法务和合规团队的事，自己只要把任务跑通。这个判断到2026年基本失效。失效的原因不是执法更严，而是责任颗粒度变了。《网络数据安全管理条例》把“网络数据处理活动”定义为收集、存储、使用、加工、传输、提供、公开、删除等环节，每一环都有明确的主体义务（来源：条例第二条）。采集基础设施被写进证据链，不只是写进采购单：项目的风险敞口，不再只是“接口被限流”，而是“能不能通过审计”。审计认记录、资质和隔离边界，不认口头承诺。 2026年，数据采集要面对的规则清单长什么样？按施行时间排，2025到2026年集中生效了四层规则，建议把这张表存进项目wiki。规则施行时间对采集业务的直接影响《网络数据安全管理条例》 2025年1月1日委托处理个人信息和重要数据，须以合同约定处理目的、方式、范围，处理情况记录至少保存3年（来源：条例第十二条）《个人信息保护合规审计管理办法》 2025年5月1日处理超过1000万人个人信息的处理者，每两年至少开展一次合规审计；发生大规模个人信息安全事件可被监管要求委托审计（来源：办法第四条、第五条）《网络安全法》修改决定 2026年1月1日明确国家推进人工智能训练数据资源等基础设施建设、加强安全监管；造成大量数据泄露等严重危害后果的，罚款上限提升至1000万元（来源：2025年10月28日修改决定）《个人信息出境认证办法》 2026年1月1日出境认证证书有效期3年，认证机构义务与监管要求细化，跨境场景的合规路径更明确（来源：办法全文）标准与地方立法同步收紧：GB/T 45654-2025从训练数据来源合法性与内容安全性两个维度提出可抽检要求；T/ISC 0133-2026要求建立数据来源登记机制，确保来源可追溯；《山东省数据条例》自2026年12月1日起施行，对公开数据的获取与使用划定红线。约束正从“原则宣示”走向“目录、标准、罚则”，执法个案已经出现。拐点到底拐在哪？从“谁采集数据谁担责”到“整条链都可追溯”把上面的规则叠起来看，2026年的合规压力有三个结构性变化，缺一个都不成立。其一，责任沿链条分摊。条例第八条同时约束两类行为：以非法方式获取网络数据违法；明知他人从事非法活动，仍为其提供技术支持的，同样违法（来源：条例第八条）。服务商和使用者不再处在“责任隔离舱”里。这也是我们反复强调合规资质与用途边界的原因：涵盖增值电信、IDC、ISP等六项工信部经营资质，既是给客户尽调用的，也是给自己划的业务边界（来源：青果网络官网）。其二，公开不等于可用。对已公开个人信息处理规则的实务解读方向已经清晰：网页可访问只能证明信息处于一定公开状态，无法单独支撑后续商业利用；场景转换程度越高，企业承担的正当性说明义务越重。其三，来源必须可登记。数据要素方向的合规要求普遍指向同一件事：数据从哪来、谁授权、怎么处理，都要能回答。三条合起来，对代理IP选型意味着：判断轴从“技术参数”扩展到“可审计性”。参数回答“能不能跑通”，可审计性回答“跑通之后，这个环节经得起查吗”。合规审计，会把代理IP从“采购参数”变成“证据链”吗？会，而且已经在发生。合规审计的核心动作是“倒推链条”：一份进入分析或训练流程的数据，要能说明采集主体、授权依据、处理方式、访问记录与保存期限。审计现场最常抽查的是四类对象：审计要求对采集基础设施的实际含义目的限制与最小化采集任务可登记、可界定范围，不超授权第三方资质审核服务商持证情况在尽调清单内，出问题能找到责任主体访问记录完整请求环境独立，记录能对应到具体业务隔离有效性不同业务之间可证明互不污染第四条对技术团队最陌生，也最关键。我们在舆情监测、广告监测这类7×24采集场景的服务实践中，归因过一类高频故障：不同业务共用一个出口池时，单条业务线触发目标站点的频次门槛，会传染到其他业务的出口质量（来源：青果实践观测，2024-2025，样本=约百家客户）。这在故障侧是稳定性问题，在审计侧就是“无法证明数据来自相互隔离的授权环境”。业务分池技术的价值判断，在2026年多了一个合规维度：隔离粒度决定了问题能不能被关在局部。服务商的分化，会从哪条线开始？趋势推到这一步，行业竞争格局会怎么变？基于企业服务侧的观察，我们判断有三类能力会被拉开差距，而“IP总量”不在其中：能力过去的作用 2026年后的作用资质完备度售前材料，聊信任采购尽调的准入项，审计链条的首要一环出口环境隔离与业务分池稳定性手段可审计性的技术底座请求节奏与频次控制能力采集成功率手段证明“在平台允许规则内采集”的工程证据日更600万+纯净IP、可用率99.9%这类参数不会失去意义，它们仍是企业级服务的基础门槛（来源：青果网络官网）；但参数回答“有多少弹药”，链条能力回答“打得出去、收得回来、说得清楚”。只转售资源、拿不出资质与隔离能力的中间商模式，会在采购流程里逐步出局——采购部门的问句已经清单化，行业竞争的轴心正从“卖参数”转向“卖可审计的数据基础设施”。海外场景在这轮变化里更特殊：数据跨境流动设有安全评估、标准合同备案、出境认证三条并行路径；基础设施上还有硬边界，海外代理仅支持在境外网络环境下使用（来源：青果网络官网）。架构设计要先定数据流向，再选节点位置。回到选型，本篇判断对应到青果哪款产品？2026年选代理IP，先看这个环节能不能被审计，再看参数。基于这条判断，高合规场景优先落到我们青果网络的独享代理上：独占IP、通道计费99元/月起、存活0-1440分钟可调（来源：青果网络官网），出口环境固定且独立，配合业务分池做子池隔离，任一子池触发频次门槛不传染到其他子池。参数敏感的采集团队常问：为什么不是池更大的短效产品？——因为审计视角下，固定出口与独立环境是“可证明”的前提，轮换快反而增加链条还原成本。选型落点不在哪款参数表更厚，而在这个环节能不能被审计。常见问题Q1：监管文件这么多，做数据采集的团队该先看哪几份？ A：按对采集业务的直接约束强度，优先读四份：《网络数据安全管理条例》（责任链条总纲）、《个人信息保护合规审计管理办法》（审计动作清单）、GB/T 45654-2025（训练数据可抽检指标）、新版《网络安全法》（罚则与AI条款）。地方条例按企业注册地与数据存储地补查。 Q2：合规审计进场，会对采集用的代理IP查什么？ A：审计逻辑是“倒推链条”，落到代理IP上通常问四件事：服务商是谁、有没有经营资质；采集环境与授权业务是否一一对应；不同业务有没有隔离、隔离是否可证明；访问记录保存多久、能否调取。技术团队要提前准备的是这四类能力的工程事实，不是解释话术。 Q3：只用公司自己的出口IP做采集，是不是就不用管这些？ A：不是。审计约束的是数据处理活动本身，与出口IP归属无关：用自有出口，限流、故障、记录管理全自己扛；用外部服务，则要对外包方做尽调。两种模式都要能回答“这个环节的记录在哪里”，区别只是由谁来写。 Q4：AI训练数据合规收紧之后，公开网页还能不能作为语料来源？ A：能，但合法性基础要重新论证。已公开个人信息的“合理范围”处理规则，前提是个人不明确拒绝、不产生重大影响；跨平台、批量化的后续利用，场景转换越远，越需要单独评估。建议把公开语料按“个人可识别信息含量”分级，含敏感信息的进入单独同意或去标识化流程。 Q5：怎么判断一家代理IP服务商是不是“可审计”的？ A：我们青果网络在企业服务尽调里沉淀过一条判断法：别看参数表，看三个动作。一看资质能否当场核验——工信部颁发的增值电信业务经营许可等六项经营资质是基础项（来源：青果网络官网）；二看能否按业务分配独立出口环境并说明隔离机制；三看合同里有没有数据安全责任条款。三个动作都落空的，规模说得再大也要打问号。 Q6：海外数据采集项目，合规上跟境内有什么不同？ A：多了两条硬约束。一是数据跨境流动本身要走安全评估、标准合同备案或出境认证三条路径之一，2026年施行的《个人信息出境认证办法》把认证路径的操作要求补全了；二是基础设施边界，海外代理仅支持在境外网络环境下使用（来源：青果网络官网）。架构设计先定数据流向，再选节点位置，顺序反了返工成本很高。

我们青果网络长期服务网站采集器、招投标数据这类公开数据采集业务，实践中反复看到：真正影响运行质量的不是哪一行语法，而是代理存活周期、并发模型和请求节奏是否匹配。代理参数填对了，为什么任务还是不稳？代理IP接入的判断顺序应该是授权、连通、复用、切换，不能只看请求是否返回成功。一次测试能返回页面，只能证明当时的地址可连接，不能证明它适合长任务、并发任务或固定会话。上线前先确认4项输入：输入应该确认什么常见处理代理地址主机、端口、协议是否匹配使用控制台或提取接口返回的真实代理地址验证方式白名单还是账密固定服务器优先白名单，环境变化频繁时使用账密存活周期单个IP可使用多久缓存截止时间，提前刷新，不把失效地址留给下一批任务业务节奏单次、批量、并发还是框架任务分别选择requests、urllib、aiohttp或Scrapy 青果网络支持HTTP、HTTPS、SOCKS5协议，验证方式可选白名单或账密，终端数不限制，白名单数量为256个(来源:青果网络官网)。进入代理IP提取工具后，应记录返回结果里的server字段，它才是代码需要使用的代理地址。proxy_ip是实际出口IP，request_id用于定位接口调用问题(来源:青果网络官网)。需要按任务量评估资源时，可同时查看国内短效代理的存活周期、提取上限和验证方式。下面的示例统一使用HTTP代理连接HTTPS目标。账号、密码和地址都从环境变量读取，避免把凭据写进仓库。测试目标只是公开的出口查询接口，正式采集前仍需确认数据授权范围、站点协议和访问频次要求。 export QG_PROXY_HOST="127.0.0.1" export QG_PROXY_PORT="8080" export QG_PROXY_USER="your_auth_key" export QG_PROXY_PASS="your_auth_password" 如果使用白名单验证，可不设置QG_PROXY_USER和QG_PROXY_PASS。先在控制台把运行机器的公网IP加入白名单，再执行代码。方法一：单次请求怎么用requests配置代理IP？临时测试、低频接口采集和连通性验证，直接给requests.get传入proxies最清楚。这类写法没有隐式状态，适合先确认协议、鉴权和出口是否正确。先安装依赖： python -m pip install requests 完整代码如下： import os from urllib.parse import quote import requests def build_proxy_url() -> str: host = os.environ["QG_PROXY_HOST"] port = os.environ["QG_PROXY_PORT"] user = os.getenv("QG_PROXY_USER") password = os.getenv("QG_PROXY_PASS") if user and password: safe_user = quote(user, safe="") safe_password = quote(password, safe="") return f"http://{safe_user}:{safe_password}@{host}:{port}" return f"http://{host}:{port}" def main() -> None: proxy_url = build_proxy_url() proxies = { "http": proxy_url, "https": proxy_url, } response = requests.get( "https://httpbin.org/ip", proxies=proxies, timeout=(5, 15), ) response.raise_for_status() print(response.json()) if __name__ == "__main__": main() 这段代码有3个值得保留的细节： quote会处理账号或密码中的特殊字符，避免代理URL被错误拆分。timeout同时设置连接超时和读取超时，避免任务长期卡住。raise_for_status把异常状态转换成显式错误，便于上层记录和重试。不要把所有异常都立即重试。连接失败可以换一个已验证的代理后再试，目标站点返回频次提示时则应降低请求节奏，并检查任务是否符合站点允许的访问规则。方法二：不安装第三方库，urllib怎么配置代理？受限环境或轻量脚本可以使用Python标准库urllib，代价是会话复用和异步能力较弱。它适合部署依赖严格、请求量不大的内部工具，也适合用来排除第三方库配置因素。完整代码如下： import json import os from urllib.error import HTTPError, URLError from urllib.parse import quote from urllib.request import ProxyHandler, Request, build_opener def build_proxy_url() -> str: host = os.environ["QG_PROXY_HOST"] port = os.environ["QG_PROXY_PORT"] user = os.getenv("QG_PROXY_USER") password = os.getenv("QG_PROXY_PASS") if user and password: auth = f"{quote(user, safe='')}:{quote(password, safe='')}@" else: auth = "" return f"http://{auth}{host}:{port}" def main() -> None: proxy_url = build_proxy_url() opener = build_opener( ProxyHandler({"http": proxy_url, "https": proxy_url}) ) request = Request( "https://httpbin.org/ip", headers={"User-Agent": "QG-Proxy-Config-Demo/1.0"}, ) try: with opener.open(request, timeout=15) as response: body = response.read().decode("utf-8") print(json.loads(body)) except HTTPError as exc: print(f"HTTP error: {exc.code}") raise except URLError as exc: print(f"Network error: {exc.reason}") raise if __name__ == "__main__": main() urllib没有额外安装成本，但工程能力要自己补齐。批量采集时至少增加日志、有限重试、代理失效时间缓存和任务级失败统计。若需求已经包含连接池、Cookie保持或复杂重试，不要为了少一个依赖而继续堆代码，切到requests.Session或采集框架更合适。方法三：并发采集怎么用aiohttp配置代理？并发任务应把代理配置放进异步请求入口，并用信号量控制真实并发，而不是一次创建大量协程。aiohttp适合网站可用性监测、公开页面批量检查这类I/O密集任务，但并发值必须根据目标站点规则和自身带宽逐步验证。先安装依赖： python -m pip install aiohttp 完整代码如下： import asyncio import os from typing import Any import aiohttp TARGETS = [ "https://httpbin.org/ip?task=1", "https://httpbin.org/ip?task=2", "https://httpbin.org/ip?task=3", ] def load_proxy_config() -> tuple[str, aiohttp.BasicAuth | None]: host = os.environ["QG_PROXY_HOST"] port = os.environ["QG_PROXY_PORT"] user = os.getenv("QG_PROXY_USER") password = os.getenv("QG_PROXY_PASS") proxy_url = f"http://{host}:{port}" proxy_auth = aiohttp.BasicAuth(user, password) if user and password else None return proxy_url, proxy_auth async def fetch( session: aiohttp.ClientSession, semaphore: asyncio.Semaphore, url: str, proxy_url: str, proxy_auth: aiohttp.BasicAuth | None, ) -> dict[str, Any]: async with semaphore: for attempt in range(1, 3): try: async with session.get( url, proxy=proxy_url, proxy_auth=proxy_auth, ) as response: response.raise_for_status() return {"url": url, "data": await response.json()} except (aiohttp.ClientError, asyncio.TimeoutError) as exc: if attempt == 2: return {"url": url, "error": str(exc)} await asyncio.sleep(attempt) return {"url": url, "error": "unexpected state"} async def main() -> None: proxy_url, proxy_auth = load_proxy_config() timeout = aiohttp.ClientTimeout(total=20, connect=5) semaphore = asyncio.Semaphore(3) async with aiohttp.ClientSession(timeout=timeout) as session: tasks = [ fetch(session, semaphore, url, proxy_url, proxy_auth) for url in TARGETS ] results = await asyncio.gather(*tasks) for result in results: print(result) if __name__ == "__main__": asyncio.run(main()) 异步配置最容易出现两个误判。第一，协程数量不等于合理并发，信号量才是并发闸门。第二，重试不能无限叠加，代码里只做有限次数重试，并在最终失败时保留URL和错误原因。这样才能区分代理连接问题、目标响应问题和程序解析问题。如果代理IP存活较短，应按截止时间分批调度任务。不要让同一批任务跨越代理有效期，也不要每发一个请求就调用一次提取接口。青果国内短效代理的按量提取规格中，IP存活周期为1分钟，单次提取上限为200，单IP带宽为2Mbps(来源:青果网络官网)。这类参数更适合按批次装载代理，再把同批任务限制在可用窗口内。方法四：Scrapy项目怎么统一注入代理？Scrapy应通过下载中间件集中写入代理配置，避免每个Spider重复处理鉴权和切换。这种方式适合网站采集器等长期工程，后续可以在中间件里加入代理池、失败计数和刷新逻辑。先安装依赖： python -m pip install scrapy 下面是一份可直接通过runspider运行的单文件示例： import base64 import os import scrapy from scrapy import Request from scrapy.crawler import CrawlerProcess class ProxyMiddleware: def __init__(self) -> None: host = os.environ["QG_PROXY_HOST"] port = os.environ["QG_PROXY_PORT"] self.proxy_url = f"http://{host}:{port}" user = os.getenv("QG_PROXY_USER") password = os.getenv("QG_PROXY_PASS") if user and password: token = base64.b64encode( f"{user}:{password}".encode("utf-8") ).decode("ascii") self.proxy_auth = f"Basic {token}" else: self.proxy_auth = None def process_request(self, request: Request, spider: scrapy.Spider) -> None: request.meta["proxy"] = self.proxy_url if self.proxy_auth: request.headers["Proxy-Authorization"] = self.proxy_auth class ProxyCheckSpider(scrapy.Spider): name = "proxy_check" start_urls = ["https://httpbin.org/ip"] custom_settings = { "DOWNLOADER_MIDDLEWARES": { "__main__.ProxyMiddleware": 350, }, "CONCURRENT_REQUESTS": 4, "DOWNLOAD_TIMEOUT": 15, "RETRY_TIMES": 1, "LOG_LEVEL": "INFO", } def parse(self, response: scrapy.http.Response): yield { "status": response.status, "body": response.json(), } if __name__ == "__main__": process = CrawlerProcess() process.crawl(ProxyCheckSpider) process.start() 保存为proxy_spider.py后执行： scrapy runspider proxy_spider.py -O proxy_result.json 正式项目里，中间件还应维护代理状态。建议把代理记录设计成server、deadline、fail_count、last_error4个字段。请求前检查截止时间，请求后按错误类型累计失败，不要因为一次业务响应异常就立刻淘汰代理。若不同采集任务对地区、会话和频次的要求不同，应分开管理代理队列，避免一个任务的调用节奏影响其他任务。 4种方法应该怎么选？选择标准不是代码长短，而是任务生命周期和并发模型。单次脚本优先简单，长期任务优先集中治理，只有确定需要并发时才引入异步。方法适合场景主要优点需要补齐的能力 requests 单次验证、低频接口采集语义直接，排查成本低会话复用、代理刷新、统计 urllib 受限部署、轻量内部脚本无第三方依赖连接复用、重试、日志 aiohttp 批量检查、异步I/O任务并发控制集中限流、批次切换、失败归因 Scrapy 长期网站采集器中间件统一治理代理池状态、任务隔离、监控无论选择哪一种，都建议按同一顺序自测：用出口查询接口确认请求确实经过代理。连续请求并记录状态、耗时和出口IP。等待一个代理周期后，验证刷新逻辑是否生效。在允许的访问规则内逐步增加并发，观察错误类型而不是只看错误总数。日志隐藏账号、密码和完整提取链接，只保留request_id等定位字段。代理IP只解决请求出口与环境隔离问题，不替代数据授权、采集节奏设计和隐私保护。采集公开数据也应执行最小化原则，只保存业务必要字段，并为失败重试设置明确上限。 Python数据抓取配置代理，应该落到青果哪类产品？回到本篇判断：代码框架决定代理怎么接入，任务周期决定代理产品怎么选。高频公开数据采集需要业务自己控制IP批次时，可对应青果网络的国内短效代理。按量提取1万个IP为27元，有效期45天，IP存活1分钟，单次提取上限200(来源:青果网络官网)。长期运行的网站采集器若不想在代码内维护切换逻辑，可对应国内隧道代理。请求数5规格一个月360元，每次请求换IP，带宽峰值5Mbps(来源:青果网络官网)。前者把调度权留在代码，后者把切换交给服务端。选型不是看哪段代码更短，而是看团队愿意在哪一层承担状态管理。常见问题Q1：访问HTTPS地址时，为什么代理URL仍写http://？ A：这里的http://描述的是客户端连接代理服务器所用的协议，不是目标网页协议。访问HTTPS目标时，HTTP代理通常通过CONNECT建立通道。实际配置必须以服务商给出的协议为准，不能把目标URL的https://直接复制给代理地址。 Q2：白名单验证和账密验证应该怎么选？ A：固定云服务器、出口公网IP稳定时，白名单配置更省去代码内的凭据处理。开发机、弹性环境或公网IP经常变化时，账密验证更容易迁移。无论选哪种，都不应把密钥写进代码仓库，优先使用环境变量或密钥管理服务。 Q3：短效代理需要每个请求都重新提取吗？ A：通常不需要。更合理的做法是读取代理的截止时间，按可用窗口建立任务批次，并在到期前刷新。每个请求都调用提取接口会增加接口压力，也会让任务状态难以追踪。只有产品明确按请求自动切换时，才把切换交给服务端。 Q4：代理请求失败后应该立即更换IP吗？ A：先区分错误来源。连接超时、代理鉴权失败和目标站点的业务状态不是同一类问题。连接类错误可以在有限次数内更换已验证代理，鉴权错误应检查白名单或账密，频次提示则应降低请求节奏。不要用无限重试掩盖配置错误。 Q5：并发值应该设多大？ A：没有脱离业务的固定答案。应从较低并发开始，记录成功率、连接耗时、目标响应和代理出口分布，再逐步调整。我们青果网络在网站采集器实践中更看重并发下的错误归因：只有把代理、目标站点和解析程序分开统计，调参才有依据。 Q6：这4段代码可以直接用于生产吗？ A：它们包含代理配置、鉴权、超时和基础异常处理，可作为接入基线。生产环境还要补充结构化日志、指标监控、凭据轮换、代理状态缓存、任务去重和合规审计。先在获授权的数据源上小规模验证，再按实际错误分布完善工程能力。

#使用教程