年报下载实战：3个高频面试题拆解项目搭建

刚学会 Python 语法，却面对“从 PDF 到 Excel”的需求手足无措？这是很多后端初学者的通病。

学会语法却不知怎么搭项目，是阻碍你从“写代码”进阶到“做工程”的最大鸿沟。在最近的几个高频面试题中，考察数据处理全流程的占比越来越高，尤其是像“年报下载与解析”这种典型的企业级场景。

今天我们就拿年报下载这个真实场景，从零搭建一个可运行的爬虫+解析工具。不讲虚的，直接上代码和工程化思路，把高频面试题里的难点吃透。

1. 项目目标与痛点拆解

为什么选“年报下载”做案例？因为它涵盖了 HTTP 请求、反爬处理、文件存储、数据解析四大核心模块。

很多初学者一上来就写 requests.get()，结果遇到以下问题就卡壳：

动态加载：网页数据是 JS 渲染的，静态请求拿不到。
反爬机制：频繁请求导致 IP 被封，或者返回 403。
数据清洗：下载下来的 PDF 或 HTML 表格，怎么转成结构化数据？

我们的目标不是写一个“能跑”的脚本，而是搭建一个可扩展、可维护的项目骨架。这直接对应了高频面试题中关于“系统架构设计”的考察点。

2. 工程化目录结构设计

拒绝“单文件脚本”思维。一个合格的项目，目录结构必须清晰。以下是我们推荐的目录结构：

annual_report_downloader/
├── config/
│   └── settings.py          # 全局配置：URL, 超时, 重试次数
├── core/
│   ├── spider.py            # 核心爬虫逻辑
│   ├── parser.py            # 数据解析逻辑
│   └── downloader.py        # 文件下载与存储
├── utils/
│   ├── logger.py            # 日志工具
│   └── exceptions.py        # 自定义异常
├── data/
│   └── raw/                 # 存放下载的原始文件
├── main.py                  # 程序入口
├── requirements.txt         # 依赖管理
└── README.md

为什么这样分？

解耦：爬虫、解析、存储分开，方便单独测试和替换。
配置分离：URL 和参数放在 config，换目标网站不用改代码。
日志规范：生产环境必须记录日志，方便排查问题。

3. 核心代码实现与逐行讲解

3.1 配置与日志初始化

在 config/settings.py 中：

import osBASE_URL = "https://example-annual-reports.com"
DOWNLOAD_DIR = os.path.join(os.path.dirname(__file__), "../data/raw")
REQUEST_TIMEOUT = 10
MAX_RETRIES = 3
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

在 utils/logger.py 中，我们使用标准库 logging，避免 print 满天飞：

import logging
import osdef get_logger(name: str) -> logging.Logger:log_dir = "logs"os.makedirs(log_dir, exist_ok=True)logger = logging.getLogger(name)logger.setLevel(logging.INFO)# 避免重复添加 handlerif not logger.handlers:file_handler = logging.FileHandler(f"{log_dir}/app.log", encoding="utf-8")file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))logger.addHandler(file_handler)return logger

3.2 爬虫核心：处理动态加载与重试

很多高频面试题会问：“如何处理 JS 渲染的页面？” 答案是：判断复杂度。如果页面简单，用 requests；如果复杂，用 Selenium 或 Playwright。这里我们以 requests 模拟一个接口请求场景，并加入重试机制。

core/spider.py：

import requests
from config.settings import HEADERS, REQUEST_TIMEOUT, MAX_RETRIES
from utils.logger import get_logger
from utils.exceptions import CustomSpiderErrorlogger = get_logger("Spider")class AnnualReportSpider:def __init__(self):self.session = requests.Session()self.session.headers.update(HEADERS)def fetch_page(self, url: str) -> str:"""获取页面内容，带重试机制"""for attempt in range(1, MAX_RETRIES + 1):try:logger.info(f"Fetching {url} (Attempt {attempt})")response = self.session.get(url, timeout=REQUEST_TIMEOUT)# 检查状态码if response.status_code == 200:return response.textelif response.status_code == 403:raise CustomSpiderError(f"Access Forbidden: {url}")else:raise CustomSpiderError(f"HTTP {response.status_code}: {url}")except requests.exceptions.RequestException as e:logger.warning(f"Request failed: {e}. Retrying...")if attempt == MAX_RETRIES:raise CustomSpiderError(f"Max retries reached for {url}")return ""def get_report_list(self, base_url: str) -> list:"""模拟从列表页获取年报链接实际场景中，这里可能需要解析 HTML 或使用 API"""# 假设这是一个 JSON API 返回链接列表# 实际中需用 BeautifulSoup 解析 HTMLurl = f"{base_url}/api/reports"content = self.fetch_page(url)# 简化处理：假设 content 是 JSON 字符串# 实际需 try-except 处理 JSON 解析错误try:import jsondata = json.loads(content)return [item['url'] for item in data.get('items', [])]except Exception as e:logger.error(f"Failed to parse report list: {e}")return []

关键点讲解：

Session 复用：requests.Session() 保持连接池，提升性能，减少 TCP 握手开销。
重试机制：网络不稳定是常态，必须处理 RequestException。
异常封装：自定义 CustomSpiderError，让上层代码能精确捕获业务错误。

3.3 下载与存储

core/downloader.py：

import os
import requests
from config.settings import DOWNLOAD_DIR, HEADERS, REQUEST_TIMEOUT
from utils.logger import get_loggerlogger = get_logger("Downloader")class ReportDownloader:def __init__(self):os.makedirs(DOWNLOAD_DIR, exist_ok=True)def download_file(self, url: str, filename: str) -> str:"""下载文件并保存到本地返回文件路径"""file_path = os.path.join(DOWNLOAD_DIR, filename)# 如果文件已存在，跳过下载（幂等性设计）if os.path.exists(file_path):logger.info(f"File already exists: {filename}")return file_pathtry:logger.info(f"Downloading {url} to {filename}")response = requests.get(url, headers=HEADERS, stream=True, timeout=REQUEST_TIMEOUT)response.raise_for_status()  # 抛出 HTTP 错误with open(file_path, 'wb') as f:for chunk in response.iter_content(chunk_size=8192):if chunk:f.write(chunk)logger.info(f"Downloaded successfully: {filename}")return file_pathexcept requests.exceptions.RequestException as e:logger.error(f"Download failed for {url}: {e}")# 删除不完整的文件if os.path.exists(file_path):os.remove(file_path)raise

避坑指南：

stream=True：大文件必须流式下载，否则内存溢出。
raise_for_status()：默认 requests 不会抛出 4xx/5xx 异常，必须手动检查。
幂等性：检查文件是否存在，避免重复下载，这在分布式任务中尤为重要。

3.4 主流程串联

main.py：

from core.spider import AnnualReportSpider
from core.downloader import ReportDownloader
from utils.logger import get_logger
import hashlib
import timelogger = get_logger("Main")def generate_filename(url: str, extension: str = "pdf") -> str:"""根据 URL 生成唯一文件名，避免覆盖"""# 使用 URL 的 MD5 前 8 位作为文件名unique_id = hashlib.md5(url.encode()).hexdigest()[:8]timestamp = int(time.time())return f"{unique_id}_{timestamp}.{extension}"def main():spider = AnnualReportSpider()downloader = ReportDownloader()base_url = "https://example-annual-reports.com"try:# 1. 获取链接列表report_urls = spider.get_report_list(base_url)if not report_urls:logger.warning("No reports found.")returnlogger.info(f"Found {len(report_urls)} reports.")# 2. 逐个下载for i, url in enumerate(report_urls):try:# 模拟文件名生成，实际需从 URL 提取filename = generate_filename(url)file_path = downloader.download_file(url, filename)logger.info(f"Processed {i+1}/{len(report_urls)}: {filename}")except Exception as e:logger.error(f"Failed to process {url}: {e}")# 继续处理下一个，不中断整个流程# 添加延时，避免对服务器造成压力time.sleep(1)except Exception as e:logger.critical(f"Critical error: {e}", exc_info=True)if __name__ == "__main__":main()

4. 运行与测试策略

很多初学者写完代码直接 python main.py，这是大忌。

4.1 单元测试

使用 pytest 对核心模块进行隔离测试。

tests/test_spider.py：

import pytest
from core.spider import AnnualReportSpider
from unittest.mock import patch, MagicMockclass TestAnnualReportSpider:@patch('requests.Session.get')def test_fetch_page_success(self, mock_get):# 模拟返回成功响应mock_response = MagicMock()mock_response.status_code = 200mock_response.text = "<html>Success</html>"mock_get.return_value = mock_responsespider = AnnualReportSpider()result = spider.fetch_page("http://test.com")assert result == "<html>Success</html>"assert mock_get.call_count == 1@patch('requests.Session.get')def test_fetch_page_retry(self, mock_get):# 模拟前两次失败，第三次成功mock_fail = MagicMock()mock_fail.side_effect = Exception("Network Error")mock_success = MagicMock()mock_success.status_code = 200mock_success.text = "Success"mock_get.side_effect = [mock_fail, mock_fail, mock_success]spider = AnnualReportSpider()result = spider.fetch_page("http://test.com")assert result == "Success"assert mock_get.call_count == 3

4.2 集成测试

在 data/raw 目录下准备几个真实的 PDF 样本，运行 main.py，观察：

日志是否记录了下载进度？
重复运行时，是否跳过了已存在的文件？
网络断开时，是否优雅报错并继续处理下一个？

5. 优化扩展与进阶技巧

当项目跑通后，如何让它更“专业”？这也是高频面试题中“性能优化”部分的考点。

5.1 异步并发

requests 是同步阻塞的。如果年报有 1000 份，串行下载效率极低。 解决方案：使用 aiohttp + asyncio。

# 伪代码示意
import aiohttp
import asyncioasync def fetch_async(session, url):async with session.get(url) as response:return await response.text()async def main():connector = aiohttp.TCPConnector(limit=10)  # 限制并发连接数async with aiohttp.ClientSession(connector=connector) as session:tasks = [fetch_async(session, url) for url in urls]results = await asyncio.gather(*tasks)

5.2 数据解析层

下载只是第一步。通常需要将 PDF 中的表格提取出来。 工具推荐：pdfplumber 或 camelot。

import pdfplumberdef extract_table_from_pdf(file_path: str):with pdfplumber.open(file_path) as pdf:for page in pdf.pages:table = page.extract_table()if table:# 将表格转换为 DataFrame 或 CSVimport pandas as pddf = pd.DataFrame(table[1:], columns=table[0])return dfreturn None