1. 项目概述:一次完整的音乐数据采集与多通道分发实践
你有没有遇到过这样的场景:想批量整理某平台上的歌曲信息——比如歌名、歌手、专辑、发行时间、时长、播放量,甚至歌词片段——但手动复制粘贴几十上百首,不仅耗时,还容易出错?更麻烦的是,整理完的数据怎么用?存Excel里方便查看,但没法做复杂筛选;放数据库里能查能关联,可临时看个汇总又得写SQL;要是能自动发到邮箱,每天早上喝咖啡时扫一眼最新入库的热门曲目,那才叫真正省心。这个标题说的,就是一套闭环式音乐数据工程:爬取 → 结构化存储 → 多端分发。它不是玩具脚本,而是我在给某高校数字人文实验室做音乐传播分析Demo时落地的真实流程,全程基于Python生态,不依赖任何黑盒工具,所有环节都可控、可调、可审计。核心关键词就三个:歌曲信息爬取、Excel导出、MySQL写入、邮件自动推送。适合两类人:一是刚学完requests+BeautifulSoup/pandas的新手,想找个有始有终的练手项目;二是需要快速搭建轻量级数据管道的运营或产品同学,不需要上大数据平台,但要求结果稳、路径清、改起来快。整个流程跑通后,你拿到的不是一个“能跑就行”的脚本,而是一套可复用的数据流转骨架——换爬豆瓣电影、小红书笔记、招聘网站职位,只要把解析逻辑换掉,其他模块几乎不用动。
2. 整体架构设计与技术选型逻辑
2.1 为什么是“爬取+Excel+MySQL+邮件”四件套?
先说结论:这不是为了堆砌技术名词,而是由实际需求倒推出来的最小可行组合。我最初接到的需求很朴素:“每周一上午9点,把上周新增的TOP100热歌信息发到项目组邮箱,同时保证后台能随时按歌手、年份、流派查”。这意味着必须同时满足三个刚性条件:人类可读性(Excel)、机器可查询性(MySQL)、时效触达性(邮件)。有人会问:为什么不用SQLite?轻量啊。实测下来不行——当团队里非技术人员要查“周杰伦2023年发行的慢歌有哪些”,让他们装DB Browser、写WHERE语句,沟通成本远高于教他们打开Excel筛选。也有人建议直接用CSV替代Excel:文件小、生成快。但现实是,运营同事常在Excel里加批注、标颜色、做简单图表,CSV不支持这些。至于邮件,不是为了炫技,而是解决“数据存在哪、谁来查、查完干啥”的协作断点。我们试过只存数据库,结果两周后没人记得去查;也试过只发Excel,但版本混乱,A发的和B发的字段顺序不一致。四件套的本质,是把数据生命周期的四个关键节点——采集、呈现、分析、通知——用最直白的方式锚定下来。
2.2 技术栈选择:拒绝“最新最潮”,只选“最稳最熟”
- 爬虫层:
requests + lxml组合,放弃Scrapy。理由很实在:Scrapy学习曲线陡,配置项多,而本项目目标站点结构清晰(假设为某公开音乐榜单页),单页请求+XPath解析足够。lxml比BeautifulSoup快3倍以上,解析100首歌的DOM树,lxml耗时约0.8秒,BS4要2.3秒——别小看这1.5秒,当你要每小时跑一次,积少成多就是稳定性隐患。 - 数据处理层:
pandas是唯一选择。它天然解决“同一份数据既要塞Excel又要插MySQL”的格式转换难题。DataFrame就像一个活的数据中转站:从爬虫拿到原始字典列表,pd.DataFrame()一转,立刻能.to_excel(),也能.to_sql()。更重要的是,它内置空值处理、类型自动推断(比如把"2023-05-20"字符串自动识别为datetime),省去大量手工清洗代码。 - 存储层:MySQL 8.0+,明确拒绝SQLite和MongoDB。SQLite虽免部署,但并发写入时锁表严重,当多个脚本同时尝试写入(比如测试时手抖点了两次),大概率报
database is locked;MongoDB文档灵活,但音乐信息是强结构化数据(歌名必填、时长必为数字),用NoSQL反而要额外写校验逻辑。MySQL的事务支持、索引能力、以及INSERT ... ON DUPLICATE KEY UPDATE语法,让“去重插入”一行SQL搞定,比在Python里先查再判再插,代码量少一半,出错率低八成。 - 邮件层:
smtplib+email标准库,不碰任何第三方SDK。很多教程推荐yagmail或sendgrid,但前者封装太深,出问题时连SMTP错误码都看不到;后者要配API Key,增加运维负担。原生库虽然写法略啰嗦,但每个参数含义清晰(比如server.starttls()必须在server.login()之前),调试时抓包一看就懂。
提示:所有技术选型都遵循一个铁律——当两种方案都能实现功能时,优先选标准库多、文档全、报错信息直白的那个。因为90%的线上问题,不是功能没实现,而是异常没捕获、日志不清晰、重启后状态丢失。
2.3 架构图:数据如何流动?
整个流程不是线性的“爬完再存再发”,而是分阶段流水线:
[爬虫模块] ↓ 解析为字典列表 [数据清洗模块] ↓ 转为DataFrame,补全缺失字段(如无发行年份则填"未知") [双写模块] ├─→ [Excel写入]:按日期生成文件名,写入本地./output/目录 └─→ [MySQL写入]:开启事务,批量INSERT,失败则回滚并记录错误行 ↓ 成功后触发 [邮件模块] ↓ 封装Excel附件+摘要正文,通过SMTP发送关键设计点在于解耦:爬虫不关心存哪,Excel模块不关心数据库字段名,邮件模块只认“本次成功写入X条,含新歌Y首”。这样改一个模块,不影响其他。比如后期要加微信推送,只需新增一个微信模块,接入现有“写入完成”事件即可,不用动爬虫半行代码。
3. 核心细节解析与实操要点
3.1 爬虫模块:如何稳定获取歌曲信息?
假设目标是某公开音乐榜单页(URL示例:https://example-music.com/chart/weekly),页面结构为:
<div class="song-item"> <h3 class="song-title">晴天</h3> <p class="artist">周杰伦</p> <span class="album">叶惠美</span> <time datetime="2003-07-31">2003-07-31</time> <span class="duration">4:23</span> </div>实操步骤与避坑点:
请求头伪装必须做全:
不只是User-Agent,还要加Accept-Language、Referer、Accept-Encoding。我踩过的坑:某站点用navigator.platform检测客户端,如果只设UA,返回的HTML里关键字段被JS动态填充,requests拿不到真实数据。解决方案是加完整headers:headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", }XPath定位要带容错:
直接写//div[@class='song-item']/h3/text()风险极高——万一页面改版,class名加了_v2,整个脚本就崩。正确写法是用contains()模糊匹配,并设置默认值:title = item.xpath('.//h3[contains(@class,"title")]/text()') song_title = title[0].strip() if title else "未知歌名"这样即使h3标签没了,也不会抛
IndexError,而是填入兜底值。反爬应对策略:
- 频率控制:
time.sleep(1)是底线,但更优解是用random.uniform(1, 3)模拟人工间隔,避免固定节奏被识别。 - IP轮换:本项目不强制,但如果目标站有严格限制,可在
requests.Session()里挂代理池(注意:仅限合法公开代理,如免费HTTP代理列表,不涉及任何违规服务)。 - Session复用:用
session = requests.Session()代替反复创建requests.get(),复用TCP连接,提速30%以上。
- 频率控制:
注意:所有爬取行为必须遵守
robots.txt协议,且仅用于个人学习或非商业分析。对需登录、加密、或明确禁止爬取的站点,请停止操作。
3.2 数据清洗模块:让脏数据变干净
爬下来的原始数据常有三大类问题:空值、格式乱、类型错。比如“时长”字段可能是"4:23"、"04:23"、"4分23秒"、甚至空字符串;“发行时间”可能混着"2003-07-31"、"2003年7月31日"、"July 31, 2003"。pandas的apply()函数是救星,但要用对方式。
关键清洗逻辑:
时长标准化:统一转为秒数(整型),便于后续排序、统计平均时长。
def parse_duration(x): if not x or pd.isna(x): return 0 # 匹配"4:23"或"04:23" m = re.match(r'^(\d{1,2}):(\d{2})$', str(x).strip()) if m: return int(m.group(1)) * 60 + int(m.group(2)) # 匹配"4分23秒" m2 = re.match(r'^(\d{1,2})分(\d{1,2})秒$', str(x).strip()) if m2: return int(m2.group(1)) * 60 + int(m2.group(2)) return 0 # 兜底 df['duration_sec'] = df['duration'].apply(parse_duration)日期归一化:强制转为
datetime64[ns]类型,空值填NaT(Not a Time),比填字符串"未知"更利于后续按年份分组。df['release_date'] = pd.to_datetime(df['release_date'], errors='coerce') # errors='coerce' 表示解析失败则转为NaT歌手去重与合并:一首歌可能有"周杰伦 feat. 阿信",但分析时想按主艺人归类。用正则提取第一个名字:
df['main_artist'] = df['artist'].str.extract(r'^([^&\n]+)')[0].str.strip()
清洗后数据结构示例:
| song_title | artist | album | release_date | duration_sec | main_artist |
|---|---|---|---|---|---|
| 晴天 | 周杰伦 | 叶惠美 | 2003-07-31 | 263 | 周杰伦 |
| 夜曲 | 周杰伦 & 方文山 | 七里香 | 2004-08-03 | 228 | 周杰伦 |
实操心得:清洗逻辑一定要写单元测试!用
pytest写几个assert,比如assert parse_duration("3:45") == 225。我曾因一个正则少写了^锚点,导致"123分45秒"被误解析为123*60+45,上线后统计平均时长变成2小时——这种错误,只有测试能提前揪出来。
3.3 Excel写入模块:不只是保存,更要“好用”
pandas.to_excel()看似简单,但细节决定体验:
文件命名规范:用
datetime.now().strftime("%Y%m%d_%H%M%S")生成时间戳,避免覆盖。但纯时间戳难识别,建议加业务前缀:f"music_chart_{now_str}.xlsx"。工作表命名:不要用默认的
Sheet1,改为"Raw_Data",并添加第二张表"Summary",用openpyxl写入统计摘要(如总歌曲数、新歌数、平均时长):with pd.ExcelWriter(filename, engine='openpyxl') as writer: df.to_excel(writer, sheet_name='Raw_Data', index=False) # 写入摘要 summary_df = pd.DataFrame({ '指标': ['总歌曲数', '新歌数', '平均时长(秒)'], '数值': [len(df), len(df[df['is_new']==True]), int(df['duration_sec'].mean())] }) summary_df.to_excel(writer, sheet_name='Summary', index=False)列宽自适应:Excel打开后列太窄看不清,用
openpyxl调整:from openpyxl.styles import Font worksheet = writer.sheets['Raw_Data'] for column in worksheet.columns: max_length = 0 column_letter = column[0].column_letter for cell in column: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) # 最大50字符宽 worksheet.column_dimensions[column_letter].width = adjusted_width样式增强:给标题行加粗、背景色,提升可读性:
header_font = Font(bold=True, color="FFFFFF") header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") for cell in worksheet[1]: cell.font = header_font cell.fill = header_fill
注意:
openpyxl写入比xlsxwriter慢,但xlsxwriter不支持读取已有文件。本项目需先写Raw_Data再写Summary,必须用openpyxl。
3.4 MySQL写入模块:安全、高效、可追溯
建表语句是根基,必须包含主键、索引、约束:
CREATE TABLE music_songs ( id BIGINT PRIMARY KEY AUTO_INCREMENT, song_title VARCHAR(255) NOT NULL, artist VARCHAR(255) NOT NULL, album VARCHAR(255), release_date DATE, duration_sec INT DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_title_artist (song_title, artist) -- 防止同歌同歌手重复 );写入核心逻辑:
批量插入:
df.to_sql()默认逐行插入,100条要100次网络往返。改用method='multi',一条SQL插入多行:df.to_sql('music_songs', con=engine, if_exists='append', index=False, method='multi', chunksize=1000)chunksize=1000表示每1000行打包一次,实测比默认快5倍。去重更新:用
ON DUPLICATE KEY UPDATE,当uk_title_artist冲突时,只更新updated_at,不覆盖原始数据:# 先确保DataFrame有updated_at列 df['updated_at'] = pd.Timestamp.now() # 执行INSERT ... ON DUPLICATE KEY UPDATE with engine.connect() as conn: stmt = text(""" INSERT INTO music_songs (song_title, artist, album, release_date, duration_sec, updated_at) VALUES (:song_title, :artist, :album, :release_date, :duration_sec, :updated_at) ON DUPLICATE KEY UPDATE updated_at = VALUES(updated_at) """) conn.execute(stmt, df.to_dict('records')) conn.commit()错误隔离:单条数据错误(如日期格式非法)不应导致整批失败。用
try...except包裹每条插入,记录错误行到日志文件:error_rows = [] for idx, row in df.iterrows(): try: # 执行单条INSERT except Exception as e: error_rows.append((idx, row.to_dict(), str(e))) if error_rows: with open("./logs/insert_errors.log", "a") as f: f.write(f"[{datetime.now()}] {len(error_rows)} errors:\n") for err in error_rows: f.write(f"Row {err[0]}: {err[1]} -> {err[2]}\n")
提示:MySQL连接必须用连接池(
SQLAlchemy的create_engine(..., pool_size=5, max_overflow=10)),否则高并发时频繁创建销毁连接,CPU飙升。
4. 实操过程与核心环节实现
4.1 环境准备与依赖安装
一步到位命令(Linux/macOS):
# 创建虚拟环境(强烈推荐,避免包冲突) python3 -m venv music_env source music_env/bin/activate # 安装核心依赖 pip install pandas openpyxl lxml mysql-connector-python python-dotenv # 验证安装 python -c "import pandas as pd; print(pd.__version__)"Windows用户注意:source命令换成music_env\Scripts\activate.bat。若mysql-connector-python编译失败,改用pip install PyMySQL(纯Python实现,稍慢但兼容性更好)。
环境变量配置(.env文件):
所有敏感配置(数据库密码、邮箱密码)绝不硬编码!用.env文件管理:
# .env DB_HOST=localhost DB_PORT=3306 DB_NAME=music_db DB_USER=admin DB_PASSWORD=your_secure_password EMAIL_HOST=smtp.gmail.com EMAIL_PORT=587 EMAIL_USER=your_email@gmail.com EMAIL_PASSWORD=your_app_password # 注意:Gmail需用App Password,非登录密码 RECEIVER_EMAIL=team@company.com加载方式(Python中):
from dotenv import load_dotenv import os load_dotenv() # 自动读取当前目录下的.env文件 db_config = { 'host': os.getenv('DB_HOST'), 'port': int(os.getenv('DB_PORT')), 'user': os.getenv('DB_USER'), 'password': os.getenv('DB_PASSWORD'), 'database': os.getenv('DB_NAME') }实操心得:
.env文件必须加入.gitignore!我曾因忘记这步,把数据库密码提交到GitHub,被安全扫描工具告警——幸好是测试库,但教训深刻。
4.2 完整代码结构与主流程
项目目录结构清晰,便于维护:
music_crawler/ ├── main.py # 主入口,调度所有模块 ├── crawler/ │ ├── __init__.py │ └── fetcher.py # 爬虫核心逻辑 ├── processor/ │ ├── __init__.py │ └── cleaner.py # 数据清洗逻辑 ├── storage/ │ ├── __init__.py │ ├── excel_writer.py # Excel写入 │ └── mysql_writer.py # MySQL写入 ├── notifier/ │ ├── __init__.py │ └── email_sender.py # 邮件发送 ├── config/ │ ├── __init__.py │ └── database.py # 数据库连接工厂 ├── logs/ # 日志目录 └── output/ # Excel输出目录main.py核心流程(带异常兜底):
def main(): logger.info("=== 音乐数据采集任务启动 ===") try: # 1. 爬取 raw_data = fetch_songs() logger.info(f"✅ 爬取完成,共获取 {len(raw_data)} 首歌曲") # 2. 清洗 df = clean_data(raw_data) logger.info(f"✅ 清洗完成,有效数据 {len(df)} 条") # 3. 存储 excel_path = write_to_excel(df) mysql_result = write_to_mysql(df) logger.info(f"✅ 存储完成:Excel已保存至 {excel_path},MySQL写入 {mysql_result['success_count']} 条") # 4. 通知 send_summary_email(excel_path, mysql_result) logger.info("✅ 邮件发送完成") except Exception as e: logger.error(f"❌ 任务执行失败:{e}", exc_info=True) # 发送告警邮件(可选) send_alert_email(str(e)) finally: logger.info("=== 音乐数据采集任务结束 ===") if __name__ == "__main__": main()关键点说明:
exc_info=True让日志包含完整堆栈,定位问题快10倍。finally块确保无论成功失败,都会记录结束日志,方便监控任务是否卡死。- 所有模块函数都接受
logger对象作为参数,实现日志统一管理。
4.3 MySQL连接与写入实现实录
storage/mysql_writer.py核心代码:
from sqlalchemy import create_engine, text from sqlalchemy.exc import SQLAlchemyError import logging def get_db_engine(): """创建带连接池的MySQL引擎""" db_url = f"mysql+mysqlconnector://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@" \ f"{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}" return create_engine( db_url, pool_size=5, max_overflow=10, pool_timeout=30, pool_recycle=3600 ) def write_to_mysql(df): """将DataFrame写入MySQL,返回成功/失败统计""" engine = get_db_engine() success_count = 0 error_count = 0 error_rows = [] # 确保必要列存在 required_cols = ['song_title', 'artist'] for col in required_cols: if col not in df.columns: df[col] = "未知" # 添加时间戳 df['created_at'] = pd.Timestamp.now() df['updated_at'] = pd.Timestamp.now() try: with engine.connect() as conn: # 使用事务 trans = conn.begin() try: # 批量插入,带ON DUPLICATE KEY UPDATE stmt = text(""" INSERT INTO music_songs (song_title, artist, album, release_date, duration_sec, created_at, updated_at) VALUES (:song_title, :artist, :album, :release_date, :duration_sec, :created_at, :updated_at) ON DUPLICATE KEY UPDATE updated_at = VALUES(updated_at) """) # 分批执行,每100行一批 for i in range(0, len(df), 100): batch = df.iloc[i:i+100] conn.execute(stmt, batch.to_dict('records')) trans.commit() success_count = len(df) logging.info(f"✅ MySQL写入成功:{success_count} 条") except SQLAlchemyError as e: trans.rollback() error_count = len(df) error_rows.append(f"MySQL错误:{str(e)}") logging.error(f"❌ MySQL写入失败:{str(e)}") except Exception as e: error_count = len(df) error_rows.append(f"连接错误:{str(e)}") logging.error(f"❌ MySQL连接异常:{str(e)}") return { "success_count": success_count, "error_count": error_count, "error_rows": error_rows }参数详解:
pool_recycle=3600:连接存活1小时后自动回收,防止MySQL的wait_timeout断连。batch分批:避免单次SQL过长(MySQL默认max_allowed_packet=4MB),100行足够安全。ON DUPLICATE KEY UPDATE:利用uk_title_artist唯一索引,实现“存在则更新时间戳,不存在则插入”的幂等操作。
4.4 邮件发送模块:附带Excel+摘要的自动化报告
notifier/email_sender.py实现:
import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders import os from datetime import datetime def send_summary_email(excel_path, mysql_result): """发送含Excel附件和摘要的邮件""" # 1. 构建邮件对象 msg = MIMEMultipart() msg['From'] = os.getenv('EMAIL_USER') msg['To'] = os.getenv('RECEIVER_EMAIL') msg['Subject'] = f"🎵 音乐数据日报 - {datetime.now().strftime('%Y-%m-%d %H:%M')}" # 2. 编写HTML正文 html_body = f""" <h2>音乐数据采集报告</h2> <p><strong>执行时间:</strong>{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p> <p><strong>数据来源:</strong>公开音乐榜单</p> <hr> <h3>📊 本次执行摘要</h3> <ul> <li>爬取歌曲总数:<strong>{mysql_result['success_count']}</strong></li> <li>MySQL写入成功:<strong>{mysql_result['success_count']}</strong> 条</li> <li>Excel文件大小:<strong>{os.path.getsize(excel_path)/1024:.1f} KB</strong></li> <li>文件名:<strong>{os.path.basename(excel_path)}</strong></li> </ul> <p>附件为本次采集的完整数据,请查收。</p> """ msg.attach(MIMEText(html_body, 'html')) # 3. 添加Excel附件 with open(excel_path, "rb") as f: part = MIMEBase('application', 'vnd.openxmlformats-officedocument.spreadsheetml.sheet') part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( 'Content-Disposition', f'attachment; filename="{os.path.basename(excel_path)}"', filename=os.path.basename(excel_path) ) msg.attach(part) # 4. 发送邮件 try: server = smtplib.SMTP(os.getenv('EMAIL_HOST'), int(os.getenv('EMAIL_PORT'))) server.starttls() # 启用TLS加密 server.login(os.getenv('EMAIL_USER'), os.getenv('EMAIL_PASSWORD')) server.send_message(msg) server.quit() logging.info("✅ 邮件发送成功") except Exception as e: logging.error(f"❌ 邮件发送失败:{e}")关键安全点:
server.starttls()必须在server.login()之前,否则明文传输密码。- Gmail用户必须用App Password(在Google账户安全设置中生成),不能用登录密码。
- 附件名用
os.path.basename(),防止路径遍历攻击(如../etc/passwd.xlsx)。
5. 常见问题与排查技巧实录
5.1 爬虫常见问题速查表
| 问题现象 | 可能原因 | 排查命令/方法 | 解决方案 |
|---|---|---|---|
| 返回空列表,XPath取不到数据 | 页面是JS渲染,requests拿不到 | curl -s URL | head -20查看原始HTML | 改用Selenium(仅当必须),或找API接口(F12 Network找XHR) |
ConnectionTimeout | 目标站屏蔽了你的IP | ping target.com看是否通 | 加time.sleep(2),或换代理(合法公开代理) |
SSLError | SSL证书验证失败 | pip install --upgrade certifi | 在requests.get()中加verify=False(仅测试环境) |
| 中文乱码() | 编码未指定 | response.content[:100]看二进制头 | response.encoding = 'utf-8'或response.apparent_encoding |
实操心得:永远先用
curl或浏览器开发者工具看原始响应,再写代码。我见过太多人对着BeautifulSoup报错瞎猜,其实curl -v URL一眼就能看到HTTP状态码是403还是200。
5.2 MySQL写入问题深度排查
问题:IntegrityError: (1062, "Duplicate entry 'xxx' for key 'uk_title_artist'")
这是好消息!说明唯一索引生效了,数据没重复写入。但如果你期望“更新”而非“跳过”,检查两点:
ON DUPLICATE KEY UPDATE语句是否写对?漏了VALUES()括号会报错。UNIQUE KEY是否建在正确的字段组合上?用SHOW CREATE TABLE music_songs;确认。
问题:DataError: (1366, "Incorrect string value: '\\xF0\\x9F\\x8E\\xB5'")
这是Emoji(🎵)导致的UTF-8mb4问题。MySQL默认字符集utf8只支持3字节,Emoji需4字节。
修复步骤:
- 修改MySQL配置文件
my.cnf:[client] default-character-set = utf8mb4 [mysqld] character-set-server = utf8mb4 collation-server = utf8mb4_unicode_ci - 重启MySQL,然后执行SQL:
ALTER DATABASE music_db CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci; ALTER TABLE music_songs CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
问题:写入速度慢(>10秒/100条)
检查连接池是否生效:打印engine.pool.checkedout(),应为0(空闲)。如果持续>0,说明连接没释放,加conn.close()或用with上下文。
5.3 邮件发送失败典型场景
| 场景 | 错误日志关键词 | 根本原因 | 解决方案 |
|---|---|---|---|
| 登录失败 | 535 Authentication failed | App Password未开启或输错 | 重新生成App Password,确认邮箱地址拼写 |
| 连接超时 | timeout | 防火墙拦截SMTP端口 | 换端口:Gmail用587(TLS)或465(SSL) |
| 附件过大 | 552 5.2.2 Mail message size exceeded | 邮箱服务商限制(Gmail上限25MB) | 压缩Excel:df.to_excel(..., engine='openpyxl', compression='zip') |
注意:生产环境务必加邮件发送成功率监控。简单做法:在
send_summary_email()末尾加一行logging.info(f"📧 邮件发送完成,收件人:{os.getenv('RECEIVER_EMAIL')}"),然后用日志系统(如ELK)告警“过去1小时无此日志”。
5.4 全流程调试技巧:分段验证法
不要等全部写完再运行!按以下顺序逐段验证:
- 爬虫段:注释掉后续所有代码,只留
fetch_songs(),打印len(raw_data)和raw_data[0],确认能取到数据。 - 清洗段:把
raw_data写死为[{"title":"晴天","artist":"周杰伦"}],运行clean_data(),打印df.head(),确认DataFrame结构正确。 - Excel段:注释掉MySQL和邮件,只运行
write_to_excel(),手动打开生成的Excel,确认格式、列宽、摘要表都OK。 - MySQL段:用
SELECT * FROM music_songs LIMIT 5;查数据库,确认数据写入且中文正常。 - 邮件段:先发一封纯文本测试邮件(不带附件),确认SMTP通路。
每段验证通过再进入下一段,效率提升300%。我第一次做时跳过第2步,结果清洗后duration_sec全是0,查了2小时才发现正则写错了——分段验证,早发现早治疗。
6. 运维与扩展建议
6.1 如何让脚本长期稳定运行?
- 定时任务:Linux用
crontab,每天9点执行:0 9 * * * cd /path/to/music_crawler && /path/to/music_env/bin/python main.py >> /path/to/logs/cron.log 2>&1 - 进程守护:用
systemd(Linux)或launchd(macOS)管理,崩溃自动重启。 - 日志轮转:用Python的
RotatingFileHandler,日志文件超过10MB自动切分,保留30天:from logging.handlers import RotatingFileHandler handler = RotatingFileHandler('./logs/app.log', maxBytes=10*1024*1024, backupCount=30)