企业官网建设流程全解析

简介：本资源是一套面向Python数据采集初学者与金融信息分析实践者的爬虫项目实战方案，聚焦于东方财富网上市公司财务报表的结构化获取，解决金融数据自动化采集与合规处理的实际问题。压缩包共19个文件，含2个核心爬虫脚本（eastmoney_crawler.py与eastmoney_crawler2.py）、10个CSV格式财务报表样本（覆盖资产负债表、利润表、现金流量表等7类报表）、3个备份文件及README、LICENSE等工程文档，总大小37.96MB；其中Requests方案脚本性能显著优于Selenium方案，实测效率提升百倍，且输出严格遵循企业会计准则规范。目前已有47人学习下载，读者可直接复用命令行参数配置逻辑（支持年度、财季、报表类型、页码区间四维控制），获得开箱即用的数据采集能力，并通过对比代码深入理解反爬适配、HTTP会话管理与结构化存储设计。

1. 为什么财务数据爬取不能只靠 Requests？——Selenium 与 Requests 协同抓取东方财富网财报的实战逻辑

你试过用纯requests去拿东方财富网（eastmoney.com）的「600519 贵州茅台」2023年年报资产负债表吗？大概率返回的是空表格、<div id="content">里只有 loading 动画，或者干脆是 403 + 一段混淆 JS。这不是你 headers 写得不够像浏览器，而是该页面的财务报表数据根本不随 HTML 一次性下发：它由前端 JS 动态调用/api/stock/f10/GetFinanceInfo类接口拉取，且请求头带有时效性 token、Referer 校验、甚至前端生成的加密参数（如_时间戳、v随机数、sign签名）。纯 requests 模拟成本高、维护脆、易失效——这正是本实践选择Selenium 与 Requests 协同分工的底层动因：Selenium 负责“登录态维持”和“真实环境触发”，Requests 负责“高频、批量、低开销的数据接口直取”。我们不是在造轮子，而是在给爬虫装上离合器——该用引擎（浏览器）时挂 D 挡，该省油（直连 API）时切 N 挡。适合两类人：一是需要稳定获取 A 股上市公司近 5 年完整财报结构化数据（资产负债表 / 利润表 / 现金流量表 / 财务指标）用于基本面分析、因子回测或教学演示的开发者；二是已踩过纯 Selenium 全页渲染导致速度慢、内存爆、超时多等坑，正寻找可落地提速方案的工程实践者。

2. 理清数据链路：从网页 DOM 到真实 API 的三步逆向定位法

东方财富网财报页看似静态，实则暗藏三层数据加载逻辑：HTML 骨架 → JS 初始化 → 异步 API 请求。若跳过逆向直接写脚本，90% 的失败源于没摸清真实数据源。下面以「贵州茅台（600519）2023年年报利润表」为例，手把手还原完整链路。

2.1 第一步：用浏览器开发者工具锁定目标表格的 DOM 容器与加载特征

打开 https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?type=web&code=sh600519
按 F12 → 切到 Elements 面板 → Ctrl+F 搜索 “利润表” → 找到<div class="table-container" id="lrb">
右键该 div → “Break on” → “subtree modifications”
刷新页面，断点会停在 JS 插入表格内容的瞬间。此时切到 Network 面板，筛选 XHR，观察断点触发前后新增的请求：你会看到类似/api/stock/f10/GetFinanceInfo?code=600519&type=lrb&year=2023&season=4的请求（注意type=lrb表示利润表，season=4表示年报，非季度报）。

提示：东方财富网对season参数校验严格——1=一季报，2=中报，3=三季报，4=年报。传0或空值将返回空数据，且无明确错误提示，极易误判为网络问题。

2.2 第二步：提取关键请求头与动态参数生成逻辑

点击上述 XHR 请求 → Headers → 查看 Request Headers。重点关注三项：

• Referer: 必须为对应股票的财报页 URL（如https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?type=web&code=sh600519），否则返回 403
• User-Agent: 需与当前 Selenium 启动的浏览器一致（后续会复用）
• Cookie: 包含em_hq_sessionid（会话 ID）、em_deviceid（设备指纹）等，必须由 Selenium 登录/访问首页后自动携带，不可硬编码

再切到 Preview 或 Response，确认返回 JSON 结构是否含data字段及report数组——这是有效数据体。若返回{"status":0,"msg":"参数错误"}，说明type/season/year组合非法，需回查网页 URL 中实际参数。

2.3 第三步：验证 API 可直连性并识别签名机制（关键！）

在终端执行最简测试：

curl -H "Referer: https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?type=web&code=sh600519" \ -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" \ "https://emweb.securities.eastmoney.com/api/stock/f10/GetFinanceInfo?code=600519&type=lrb&year=2023&season=4"

若返回 403 或空数据，说明缺失 Cookie 或存在额外校验。此时回到 Network 面板，勾选 “Preserve log”，重新加载页面，找到该请求的Initiator列（显示哪段 JS 发起请求）。点击跳转至 Sources 面板，定位到类似financeApi.getFinanceData(...)的调用处。搜索关键词sign=、_=、v=，你会发现如下典型代码片段（经混淆，但逻辑可还原）：

// 伪代码：实际为压缩后的 webpack bundle const t = Date.now(); const n = Math.random().toString(36).substr(2, 8); const sign = md5(`code=600519&type=lrb&year=2023&season=4&_=${t}&v=${n}&key=EM_STOCK_F10_SECRET`);

即：sign是对查询字符串 + 时间戳_+ 随机串v+ 固定密钥拼接后 MD5。该密钥EM_STOCK_F10_SECRET在前端 JS 中明文存在（搜索EM_STOCK_F10_SECRET或key=即可定位），无需逆向算法，只需提取。

注意：此密钥并非全局固定，不同财报模块（如zcfzb资产负债表、xjllb现金流）可能使用不同密钥，需分别提取。常见密钥有EM_STOCK_F10_LRB、EM_STOCK_F10_ZCFZB等，均位于同一 JS 文件中。

3. 构建协同工作流：Selenium 负责“身份入场”，Requests 负责“数据收割”

协同的核心在于：Selenium 不渲染整页，只做最小必要动作——访问首页、触发 JS 初始化、提取 Cookie 和 User-Agent，然后交棒给 Requests。这样既规避了 Selenium 渲染开销，又确保了请求合法性。整个流程分四阶段：初始化浏览器 → 获取合法会话 → 构造签名参数 → 批量调用 API。

3.1 初始化无头 Chrome 并复用 User-Agent

我们禁用图片、JS（部分）、GPU 加速，并显式设置 UA，确保 Requests 复用时 header 一致：

from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from webdriver_manager.chrome import ChromeDriverManager def init_selenium_driver(): chrome_options = Options() chrome_options.add_argument("--headless") # 无头模式 chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--disable-gpu") chrome_options.add_argument("--disable-images") # 关键：加速加载 chrome_options.add_argument("--disable-javascript") # 关键：防止干扰 chrome_options.add_argument("--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") # 启动驱动（自动管理版本） driver = webdriver.Chrome( service=Service(ChromeDriverManager().install()), options=chrome_options ) return driver driver = init_selenium_driver()

逻辑说明：--disable-javascript是关键取舍——它让页面无法执行动态请求，但能完整加载 HTML 和内联 JS（含密钥），同时极大提升启动速度。我们不需要 Selenium 执行请求，只需要它加载出含密钥的 JS 上下文。

3.2 访问首页并提取 Cookie 与密钥

访问任意股票财报页（如贵州茅台），等待 DOM 加载，然后从页面源码中提取密钥，并从 driver 获取当前 Cookie：

import re import json from urllib.parse import urlparse, parse_qs def extract_finance_keys_and_cookies(driver, stock_code="600519"): url = f"https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?type=web&code=sh{stock_code}" driver.get(url) # 等待页面基础元素出现（非数据，仅 DOM） from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.CLASS_NAME, "table-container")) ) # 1. 提取密钥：搜索页面源码中的 EM_STOCK_F10_* 字符串 page_source = driver.page_source lrb_key_match = re.search(r'EM_STOCK_F10_LRB\s*[:\s]*["\']([^"\']+)["\']', page_source) zcfzb_key_match = re.search(r'EM_STOCK_F10_ZCFZB\s*[:\s]*["\']([^"\']+)["\']', page_source) keys = { "lrb": lrb_key_match.group(1) if lrb_key_match else "default_lrb_key", "zcfzb": zcfzb_key_match.group(1) if zcfzb_key_match else "default_zcfzb_key", "xjllb": re.search(r'EM_STOCK_F10_XJLLB\s*[:\s]*["\']([^"\']+)["\']', page_source).group(1) or "default_xjllb_key" } # 2. 提取 Cookie 字符串（供 requests 复用） cookies = driver.get_cookies() cookie_str = "; ".join([f"{c['name']}={c['value']}" for c in cookies]) # 3. 获取当前 UA（与 driver 一致） user_agent = driver.execute_script("return navigator.userAgent;") return keys, cookie_str, user_agent keys, cookie_str, ua = extract_finance_keys_and_cookies(driver, "600519") print("Extracted keys:", keys) print("Cookie string:", cookie_str[:50] + "...") print("User-Agent:", ua)

参数说明：driver.get_cookies()返回字典列表，需拼成name=value; name2=value2格式供 requests 使用；execute_script("return navigator.userAgent")确保 UA 与浏览器实际一致，避免因手动设置 UA 导致 Referer 校验失败。

3.3 构造带签名的 Requests 请求函数

封装一个通用函数，输入股票代码、报表类型、年份、季度，自动计算sign并发起请求：

import hashlib import time import random import requests def build_signed_url(code, report_type, year, season, keys, cookie_str, ua): """ report_type: 'lrb'(利润表), 'zcfzb'(资产负债表), 'xjllb'(现金流量表) keys: 从 extract_finance_keys_and_cookies 返回的密钥字典 """ base_params = { "code": code, "type": report_type, "year": str(year), "season": str(season) } # 生成动态参数 t = int(time.time() * 1000) # 毫秒时间戳 v = ''.join(random.choices('abcdefghijklmnopqrstuvwxyz0123456789', k=8)) # 拼接签名原文：按字典序排序参数 + _ + v + 密钥 sorted_items = sorted(base_params.items()) param_str = "&".join([f"{k}={v}" for k, v in sorted_items]) sign_input = f"{param_str}&_={t}&v={v}&key={keys[report_type]}" # 计算 MD5 sign = hashlib.md5(sign_input.encode()).hexdigest() # 构造最终 URL url = f"https://emweb.securities.eastmoney.com/api/stock/f10/GetFinanceInfo?{param_str}&_={t}&v={v}&sign={sign}" return url def fetch_finance_data(code, report_type, year, season, keys, cookie_str, ua): url = build_signed_url(code, report_type, year, season, keys, cookie_str, ua) headers = { "Referer": f"https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?type=web&code=sh{code}", "User-Agent": ua, "Cookie": cookie_str, "Accept": "application/json, text/plain, */*", "X-Requested-With": "XMLHttpRequest" } try: resp = requests.get(url, headers=headers, timeout=15) resp.raise_for_status() data = resp.json() # 东方财富返回结构：{"status":1, "data": {...}} 或 {"status":0, "msg":"xxx"} if data.get("status") != 1: print(f"API error for {code} {report_type} {year}Q{season}: {data.get('msg', 'Unknown')}") return None return data["data"] except Exception as e: print(f"Request failed for {code} {report_type} {year}Q{season}: {e}") return None # 示例调用 data = fetch_finance_data("600519", "lrb", 2023, 4, keys, cookie_str, ua) if data: print("Profit data fetched:", len(data.get("report", [])), "rows")

逻辑说明：build_signed_url严格按前端 JS 逻辑还原签名过程；fetch_finance_data增加了 status 校验和异常捕获，避免因单次失败中断整个批次；timeout=15防止卡死，比 Selenium 默认 60s 更可控。

4. 规避反爬与稳定性陷阱：5 条血泪经验总结

爬取东方财富网财报数据，最大的成本不在代码，而在调试时间。以下 5 条是某跨平台金融数据系统在 3 年迭代中反复验证的避坑清单，每一条都对应一次线上翻车事故：

4.1 现象：请求返回{"status":0,"msg":"参数错误"}，但code/type/year/season明明正确

原因：season参数被前端 JS 自动修正。例如，2023 年年报实际对应season=4，但若传season=2023（误以为是年份），或season=0（默认值），接口静默返回错误。更隐蔽的是：某些股票（如 ST 类）2023 年未发布年报，则season=4也无效，需先查GetFinanceSummary接口确认可用 season。
解决：在调用主报表 API 前，先请求摘要接口https://emweb.securities.eastmoney.com/api/stock/f10/GetFinanceSummary?code=600519，解析其reportList字段，提取season和year的合法组合，再逐个请求。

4.2 现象：首次成功，后续请求全部 403，Cookie 未过期

原因：em_hq_sessionid会话 ID 与 IP + User-Agent 强绑定。若 Selenium 启动时未固定 UA，或服务器端做了设备指纹校验（em_deviceid），会导致 Cookie 失效。单纯复用driver.get_cookies()不够，必须确保每次 requests 的 UA 与 driver 启动时完全一致。
解决：在init_selenium_driver()中硬编码 UA 字符串，并在fetch_finance_data()中严格复用，禁止用driver.execute_script("return navigator.userAgent")动态获取（因--disable-javascript下该脚本返回空）。

4.3 现象：数据字段错位，如“营业收入”列实际是“营业成本”

原因：东方财富网对不同股票、不同年份的报表字段顺序不保证一致。其返回 JSON 中report是二维数组，header字段定义列名，但header本身可能缺失或为空，此时需 fallback 到column字段（若存在）或解析report[0]的 key。
解决：统一用data.get("header") or data.get("column") or list(data.get("report", [{}])[0].keys())获取列名，并用pandas.DataFrame(report, columns=header)构造 DataFrame，避免硬索引。

4.4 现象：Selenium 启动极慢（>30s），或频繁崩溃

原因：webdriver-manager自动下载的 ChromeDriver 版本与系统 Chrome 不匹配；或未禁用 GPU/沙箱导致容器内运行失败。
解决：改用chromedriver-py库（pip install chromedriver-py），它提供预编译二进制且版本精准匹配；或在 Docker 中显式指定CHROMEDRIVER_VERSION环境变量。

4.5 现象：批量爬取时，部分请求返回空report，但状态码 200

原因：接口存在隐式限频，非 429 错误码，而是返回{status:1,data:{report:[]}}。单 IP 每分钟请求超过 20 次即触发。
解决：在fetch_finance_data()外层加time.sleep(0.5)，或使用requests.adapters.HTTPAdapter设置连接池和重试策略，对空report自动重试 2 次（间隔 1s）。

5. 工程化落地：构建可配置、可扩展、可监控的财报爬取管道

当你要为 5000+ A 股公司、10 年财报、3 类报表建立稳定数据源时，脚本必须升级为管道。这里不讲 Airflow 或 Kubeflow，只聚焦一线工程师真正用得上的 4 个落地技巧：配置驱动、增量更新、结构化存储、轻量监控。

5.1 用 YAML 配置替代硬编码：让股票列表、年份范围、报表类型可外部管理

创建config.yaml：

stocks: - code: "600519" name: "贵州茅台" - code: "000858" name: "五粮液" - code: "601318" name: "中国平安" years: [2023, 2022, 2021] reports: - type: "lrb" name: "利润表" - type: "zcfzb" name: "资产负债表" - type: "xjllb" name: "现金流量表" output_dir: "./data/finance"

Python 加载逻辑：

import yaml from pathlib import Path def load_config(config_path="config.yaml"): with open(config_path, encoding="utf-8") as f: return yaml.safe_load(f) config = load_config() for stock in config["stocks"]: for year in config["years"]: for rpt in config["reports"]: # 构造任务：(stock["code"], rpt["type"], year, 4) # 年报 season=4 pass

价值点：运维人员无需改 Python 代码，只需编辑 YAML 即可增删股票、调整年份，配置即文档，降低协作成本。

5.2 实现增量更新：用文件时间戳 + 数据哈希双保险避免重复抓取

为每个报表生成唯一文件名：{code}_{report_type}_{year}_Q{season}.json。抓取前检查文件是否存在且非空，再读取其内容哈希（如sha256(json.dumps(data, sort_keys=True))），与本次请求结果哈希比对。仅当哈希不同才写入新文件：

import hashlib import json from pathlib import Path def save_if_updated(code, report_type, year, season, data, output_dir="./data/finance"): filename = f"{code}_{report_type}_{year}_Q{season}.json" filepath = Path(output_dir) / filename # 若文件存在，计算现有哈希 old_hash = None if filepath.exists() and filepath.stat().st_size > 0: try: with open(filepath, encoding="utf-8") as f: old_data = json.load(f) old_hash = hashlib.sha256(json.dumps(old_data, sort_keys=True).encode()).hexdigest() except: pass # 计算新数据哈希 new_hash = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest() if old_hash != new_hash: filepath.parent.mkdir(exist_ok=True) with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"Updated: {filename}") return True else: print(f"Skipped (no change): {filename}") return False # 调用 if data: save_if_updated("600519", "lrb", 2023, 4, data)

优势：避免因网络抖动导致的重复写入，同时天然支持“数据变更即触发下游处理”的事件驱动架构。

5.3 结构化导出为 Parquet：兼顾查询性能与存储效率

JSON 适合传输，但不适合分析。用pyarrow直接导出为 Parquet，体积减少 60%，Pandas 读取速度快 3 倍：

import pandas as pd import pyarrow as pa import pyarrow.parquet as pq def export_to_parquet(code, report_type, year, season, data, output_dir="./data/finance"): if not data or not data.get("report"): return # 构造 DataFrame（自动处理 header/column/report） header = data.get("header") or data.get("column") or list(data["report"][0].keys()) if data["report"] else [] df = pd.DataFrame(data["report"], columns=header) # 添加元数据列 df["stock_code"] = code df["report_type"] = report_type df["year"] = year df["season"] = season # 导出为 Parquet filename = f"{code}_{report_type}_{year}_Q{season}.parquet" filepath = Path(output_dir) / filename table = pa.Table.from_pandas(df) pq.write_table(table, filepath, compression="snappy") print(f"Exported to Parquet: {filepath}") # 调用 if data: export_to_parquet("600519", "lrb", 2023, 4, data)

场景价值：后续用 DuckDB 或 Polars 直接SELECT * FROM 'data/finance/*.parquet' WHERE stock_code='600519'，毫秒级响应，无需预加载全量数据。

5.4 轻量监控：用日志 + 简单统计实现故障自检

在主循环中加入计数器和异常日志：

from collections import defaultdict import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.FileHandler("finance_crawl.log"), logging.StreamHandler()] ) stats = defaultdict(int) def crawl_task(code, report_type, year, season, keys, cookie_str, ua): global stats try: data = fetch_finance_data(code, report_type, year, season, keys, cookie_str, ua) if data and data.get("report"): stats["success"] += 1 save_if_updated(code, report_type, year, season, data) else: stats["empty_data"] += 1 logging.warning(f"Empty data: {code} {report_type} {year}Q{season}") except Exception as e: stats["error"] += 1 logging.error(f"Failed task {code} {report_type} {year}Q{season}: {e}") # 主循环结束后打印统计 print("\n=== Crawl Summary ===") for k, v in stats.items(): print(f"{k}: {v}")

我的习惯：每天早上用grep "ERROR" finance_crawl.log | tail -20快速扫一眼昨夜失败项，配合cat finance_crawl.log | grep "Empty data" | wc -l看空数据比例。若空数据 >5%，立即检查摘要接口是否异常——这比等告警邮件快 10 分钟。

希望帮到你。

本文还有配套的精品资源，点击获取