简介本资源是一个基于Python实现的抖音DouYin移动应用自动化爬虫项目面向具备基础Python编程能力及移动端测试兴趣的学习者与开发者旨在解决App端数据难以直接抓取的技术难题。项目融合Appium模拟用户操作、Mitmproxy拦截HTTPS流量、Genymotion提供高性能Android模拟环境三大核心技术完整呈现移动爬虫从行为控制、通信监听到数据解析的闭环实践路径。压缩包为4KB的ZIP格式共含3个文件2个核心Python脚本Download_script.py负责流量数据提取Auto_Ctrl.py实现Appium自动化控制与1份README.md说明文档结构精炼、即下即用。目前已有460人学习下载适合希望深入理解移动应用反爬机制、掌握流量分析与自动化测试协同方案的中阶开发者可直接复现关键流程、调试抓取逻辑并拓展至其他App场景。1. DouYin_Spider不是“爬虫”而是一套面向抖音公开接口的协议级数据获取工具链你搜“DouYin_Spider”大概率会点进一堆标着“一键下载无水印视频”“批量抓取用户主页”的仓库——但真正能稳定跑过 7 天、不被封 IP、不触发滑块验证、不因字段变更直接崩掉的不到 3%。这不是 Python 脚本写得漂不漂亮的问题而是对抖音 Web 端非 App公开 HTTP 接口的请求构造、签名机制、设备指纹模拟、响应解析这四层逻辑是否闭环。这个项目不封装 Selenium不依赖第三方代理池不调用任何黑盒 SDK它用纯 requests execjs 自研签名生成器把抖音 Web 端首页推荐流、搜索结果页、用户作品页三类核心场景的请求链路拆解成可调试、可断点、可替换签名模块的函数单元。适合两类人一是需要做短视频内容分析、舆情监测、竞品账号追踪的业务方要求数据可溯源、字段可对齐、失败可定位二是正在学逆向与协议分析的开发者想亲手拆一个真实高频更新的商业前端接口。它解决的不是“能不能拿到”而是“拿到的每一条数据你知道它从哪个接口来、带什么参数、为什么是这个时间戳、字段缺失时怎么补全”。2. 协议逆向基础为什么必须绕开浏览器自动化直击抖音 Web 的签名体系抖音 Web 端所有关键数据接口如/aweme/v1/web/search/item/、/aweme/v1/web/user/profile/other/都强制校验X-Bogus和X-Signature两个请求头。前者是字节系通用的 anti-bot 签名后者是抖音 Web 独有的动态 token。二者均基于 URL 查询参数、时间戳、设备标识device_id、用户行为序列如点击路径等输入经前端 JS 混淆函数计算得出。若用 Selenium 加载页面再提取数据看似“合法”实则埋下三颗雷第一JS 执行环境不可控同一段代码在不同机器上生成签名不一致第二页面加载耗时长单次请求平均 2.3 秒批量采集吞吐量卡死在 15 QPS 以下第三抖音已对 Puppeteer/Playwright 类无头浏览器增加 UA 特征识别2024 年 Q2 起未携带完整 WebGL/Canvas 指纹的请求50% 概率返回{status_code:10111,status_msg:Forbidden}。所以本项目选择“协议级复现”用 Python 提取原始 JS 签名函数通过 execjs 在本地执行输入完全可控的参数输出确定性签名。这不是偷懒而是把不可测的浏览器环境换成可打日志、可改参数、可单元测试的函数调用。2.1 签名函数提取从抖音 Web 前端 JS 中定位并剥离核心逻辑抖音 Web 的签名逻辑藏在https://www.douyin.com/web/static/js/*-*.js这类混淆文件中。我们不靠肉眼找而是用 Chrome DevTools 的Network → Filter: js → XHR → 触发一次搜索请求 → 查看 Request Headers → 定位X-Signature值 → 在 Sources 面板全局搜索该值的生成位置。实际操作中你会在某个xxx.min.js文件里找到类似这样的片段function e(t, n) { var r t n; return sig_ md5(r).substr(0, 16) }但这只是表象。真正的签名函数嵌套在多层 IIFE立即执行函数中且关键变量名被压缩为单字母。此时需借助 AST 工具反混淆用esbuild --minify --keep-names对 JS 文件做轻量还原再用正则匹配function [a-z]\([^)]*\)\{[^}]*X-Signature[^}]*\}定位函数体。最终提取出的签名函数已脱敏重命名如下// signature_gen.js function generateXSignature(url, params, userAgent) { const timestamp Math.floor(Date.now() / 1000); const deviceId 6d8c1f9a3e7b2c4d; // 固定设备 ID用于复现 const queryStr new URLSearchParams({...params, version_code: 230100, device_platform: web}).toString(); const input ${url}?${queryStr}${timestamp}${deviceId}${userAgent}; return sig_ md5(input).substring(0, 16); }提示deviceId不是随机生成的而是从抖音 Web 首页 HTML 的script标签中提取的window.deviceId字段。每次会话应复用同一个deviceId否则签名无效。本项目在core/session.py中封装了自动提取与缓存逻辑。2.2 请求头构造X-Bogus 与 X-Signature 的协同校验机制抖音 Web 同时校验X-Bogus和X-Signature但二者作用不同X-Bogus是字节系通用防刷签名由url query user_agent timestamp经bogus算法生成X-Signature是抖音 Web 特有 token由url full_query_string timestamp device_id user_agent经自定义 MD5 截断生成。漏掉任一都会返回403 Forbidden。本项目将二者封装为get_headers()方法# core/headers.py import time import hashlib import execjs from urllib.parse import urlencode def get_headers(url: str, params: dict, user_agent: str None) - dict: if not user_agent: user_agent Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 # Step 1: Generate X-Bogus query_str urlencode(params) bogus_input f{url}?{query_str}{int(time.time())}{user_agent} bogus execjs.compile( function getBogus(s) { // 此处为从抖音 JS 提取的完整 bogus 算法实现 // 包含 base64 编码、异或、字符位移等步骤 // 代码略详见项目 assets/bogus.js } ).call(getBogus, bogus_input) # Step 2: Generate X-Signature signature_input f{url}?{query_str}{int(time.time())}6d8c1f9a3e7b2c4d{user_agent} signature sig_ hashlib.md5(signature_input.encode()).hexdigest()[:16] return { User-Agent: user_agent, X-Bogus: bogus, X-Signature: signature, Referer: https://www.douyin.com/, Cookie: tt_webid1234567890; tt_webid_v21234567890; }逻辑说明X-Bogus的生成必须严格复现抖音前端 JS 的算法不能用 Python 的hashlib替代因为其内部包含非标准的 base64 表和位运算顺序X-Signature虽用 MD5但输入字符串拼接顺序、字段是否带version_code、device_platform等参数必须与抖音 Web 实际请求完全一致。参数说明url必须是接口原始路径不含域名params是字典格式查询参数user_agent应与 Cookie 中的设备标识匹配否则触发风控。2.3 设备指纹模拟为什么固定device_id和tt_webid是稳定运行的前提抖音 Web 的风控模型会关联device_id、tt_webid、tt_webid_v2、User-Agent、IP 地址这五元组。其中device_id和tt_webid是服务端下发的设备标识一旦变化即使签名正确也会返回{status_code:10111}。因此本项目在首次访问首页时自动解析 HTML 中的script标签提取window.deviceId和window.ttWebid并持久化到本地config/device.json# core/session.py import re import json import requests def init_device_session(): headers {User-Agent: Mozilla/5.0...} resp requests.get(https://www.douyin.com/, headersheaders) html resp.text # Extract deviceId and ttWebid from inline script device_id re.search(rwindow\.deviceId\s*\s*([^]), html).group(1) tt_webid re.search(rwindow\.ttWebid\s*\s*([^]), html).group(1) with open(config/device.json, w) as f: json.dump({device_id: device_id, tt_webid: tt_webid}, f) return device_id, tt_webid后续所有请求均复用该device_id和tt_webid并写入 Cookie。这是血泪经验某次测试中A同学未固化device_id而是每次随机生成结果 100 次请求内 92 次被限流日志显示status_code: 10111固化后连续 72 小时采集未触发封禁。3. 核心模块实战从搜索关键词到解析视频列表的端到端流程本项目不提供“一键采集”脚本而是把抖音 Web 的三大主干接口搜索、用户主页、推荐流封装为独立可调用的模块。每个模块遵循统一范式input → sign → request → parse → output。以“按关键词搜索视频”为例完整走一遍从输入“猫”到拿到 20 条视频信息的流程。3.1 搜索接口调用/aweme/v1/web/search/item/的参数边界与分页控制抖音 Web 搜索接口 URL 为https://www.douyin.com/aweme/v1/web/search/item/必需参数包括keyword搜索词、search_source来源固定为normal_search、offset偏移量、count单页数量。关键限制count最大为 20offset必须是 20 的整数倍且总页数受total字段限制非固定值随关键词热度浮动。本项目在modules/search.py中封装了分页逻辑# modules/search.py import requests from core.headers import get_headers from core.session import load_device_config def search_videos(keyword: str, offset: int 0, count: int 20) - dict: device_conf load_device_config() url https://www.douyin.com/aweme/v1/web/search/item/ params { keyword: keyword, search_source: normal_search, offset: str(offset), count: str(count), device_platform: web, version_code: 230100 } headers get_headers(url, params, user_agentdevice_conf[user_agent]) # 注意此处 headers 已包含 X-Bogus 和 X-Signature resp requests.get(url, paramsparams, headersheaders, timeout10) if resp.status_code ! 200: raise Exception(fSearch API failed: {resp.status_code} - {resp.text[:100]}) data resp.json() if data.get(status_code) ! 0: raise Exception(fSearch logic failed: {data.get(status_msg, Unknown)}) return data # 调用示例 if __name__ __main__: result search_videos(猫, offset0, count20) print(fTotal found: {result[data][total]}) for item in result[data][data]: print(fTitle: {item[aweme_info][desc][:30]} | Play: {item[aweme_info][statistics][play_count]})逻辑说明search_videos()返回的是原始 JSON结构为{ status_code: 0, data: { total: 12345, data: [ {...}, {...} ] } }。data[data]数组中每个元素是一个aweme_info对象包含视频标题、描述、播放量、作者昵称、封面 URL 等字段。参数说明offset从 0 开始每页 20 条最大offset为(total // 20) * 20keyword支持中文但需 URL 编码requests自动处理search_source若填错如video_search会返回空结果。3.2 视频信息解析从aweme_info到结构化字段的映射规则抖音 Web 返回的aweme_info是深度嵌套的 JSON字段名高度混淆如desc是标题statistics.play_count是播放量author.nickname是作者名且部分字段可能为空如music对象在无配乐视频中不存在。本项目在parsers/video_parser.py中定义了健壮的解析函数# parsers/video_parser.py def parse_video_item(raw_item: dict) - dict: aweme raw_item.get(aweme_info, {}) author aweme.get(author, {}) stats aweme.get(statistics, {}) music aweme.get(music, {}) return { aweme_id: aweme.get(aweme_id, ), desc: aweme.get(desc, ).strip(), create_time: aweme.get(create_time, 0), # 时间戳需转 datetime play_count: stats.get(play_count, 0), digg_count: stats.get(digg_count, 0), comment_count: stats.get(comment_count, 0), share_count: stats.get(share_count, 0), author_id: author.get(uid, ), author_name: author.get(nickname, ), author_sec_uid: author.get(sec_uid, ), cover_url: aweme.get(video, {}).get(cover, {}).get(url_list, [])[0], play_url: aweme.get(video, {}).get(play_addr, {}).get(url_list, [])[0], music_title: music.get(title, ), music_author: music.get(author, ), is_ads: aweme.get(is_ads, False) } # 使用示例 raw_data search_videos(猫, offset0, count1) parsed parse_video_item(raw_data[data][data][0]) print(fID: {parsed[aweme_id]}, Title: {parsed[desc]}, Play: {parsed[play_count]})逻辑说明所有.get()调用均设默认值避免KeyErrorcover_url和play_url从url_list数组取第一个因抖音返回多个分辨率 URLcreate_time是 Unix 时间戳秒级需用datetime.fromtimestamp()转换。参数说明raw_item是search_videos()返回的data[data]中的单个元素解析结果为扁平化字典可直接存入 CSV 或数据库。3.3 用户主页采集/aweme/v1/web/user/profile/other/的sec_uid获取陷阱要采集某用户全部作品需先获取其sec_uid安全 UID而非公开的uid。sec_uid是抖音 Web 用于用户主页接口的唯一标识长度为 20~25 位字符串形如MS4wLjABAAAA...。它不出现在用户主页 HTML 中而是藏在/aweme/v1/web/user/profile/other/接口的响应里——但该接口本身又需要sec_uid作为参数。这是一个鸡生蛋问题。破解方法从用户主页 URL 入手。抖音 Web 用户主页 URL 格式为https://www.douyin.com/user/MS4wLjABAAAA...其中MS4wLjABAAAA...就是sec_uid。本项目在modules/user.py中提供了 URL 解析函数# modules/user.py import re def extract_sec_uid_from_url(user_url: str) - str: 从抖音用户主页 URL 中提取 sec_uid match re.search(r/user/([A-Za-z0-9_\-]{20,}), user_url) if not match: raise ValueError(fInvalid douyin user URL: {user_url}) return match.group(1) def get_user_videos(sec_uid: str, max_count: int 100) - list: 获取指定用户全部视频分页拉取 videos [] offset 0 while len(videos) max_count: url https://www.douyin.com/aweme/v1/web/user/profile/other/ params { sec_uid: sec_uid, publish_video_count: 1, is_first_enter: 1, count: 20, offset: str(offset) } headers get_headers(url, params) resp requests.get(url, paramsparams, headersheaders, timeout10) data resp.json() if data.get(status_code) ! 0 or not data.get(data): break items data[data].get(aweme_list, []) videos.extend(items) if len(items) 20: # 最后一页 break offset 20 return videos[:max_count] # 使用示例 sec_uid extract_sec_uid_from_url(https://www.douyin.com/user/MS4wLjABAAAA...) videos get_user_videos(sec_uid, max_count50) print(fFetched {len(videos)} videos for user {sec_uid[:10]}...)逻辑说明extract_sec_uid_from_url()用正则匹配 URL 路径中的sec_uid比解析 HTML 更可靠get_user_videos()循环调用每次取 20 条直到aweme_list返回不足 20 条或达到max_count。参数说明sec_uid必须准确否则返回{status_code:10111}publish_video_count参数虽名为“发布数”实为固定值1填其他值会导致接口报错。4. 避坑指南五个让开发者凌晨三点还在查日志的真实问题抖音 Web 接口的稳定性极差字段随时变更、签名算法季度更新、风控策略动态调整。以下是本项目实测中踩过的 5 个高频坑每条都附带现象、根因与可落地的解决方案不是泛泛而谈。4.1 现象X-Signature正确但返回{status_code:10111,status_msg:Forbidden}原因X-Signature输入字符串中漏掉了version_code或device_platform参数或device_id与 Cookie 中的tt_webid不匹配。抖音校验时会比对签名输入与请求实际参数的一致性哪怕多一个空格也失败。解决在get_headers()函数中打印signature_input字符串并与抓包工具如 Charles中抖音 Web 实际发出的请求 URL 和 Query String 逐字符比对。重点检查version_code230100是否存在、device_platformweb是否小写、device_id是否与config/device.json中一致。4.2 现象搜索接口返回{status_code:0,data:{total:0,data:[]}}但网页能搜到结果原因search_source参数值错误。抖音 Web 搜索有多个来源normal_search综合搜索、video_search仅视频、user_search仅用户。传错值会导致返回空。解决固定使用search_sourcenormal_search。若需特定类型应改用/aweme/v1/web/search/sug/搜索建议或/aweme/v1/web/user/search/用户搜索等专用接口而非在item接口硬塞参数。4.3 现象aweme_info.video.play_addr.url_list为空数组无法获取播放地址原因抖音对部分视频尤其是新发布或低互动视频不返回play_addr只返回cover和origin_cover。这是服务端策略非客户端问题。解决在parse_video_item()中增加 fallback 逻辑当play_addr.url_list为空时尝试从aweme_info.video.bit_rate数组中取最高码率的play_addr或直接返回cover_url虽为封面但可作占位。代码示例play_urls aweme.get(video, {}).get(play_addr, {}).get(url_list, []) if not play_urls: bit_rates aweme.get(video, {}).get(bit_rate, []) if bit_rates: play_urls bit_rates[0].get(play_addr, {}).get(url_list, [])4.4 现象requests.get()报ReadTimeout但手动 curl 相同 URL 成功原因抖音 Web 对请求头Accept-Encoding敏感。若未设置Accept-Encoding: gzip, deflate服务端可能不压缩响应导致传输体积过大5MB超时。解决在get_headers()返回的字典中强制添加Accept-Encoding: gzip, deflate。同时在requests.get()调用时显式传入streamTrue避免一次性读入内存resp requests.get(url, paramsparams, headersheaders, timeout10, streamTrue) resp.raise_for_status() data resp.json() # 此时才解析4.5 现象连续请求后所有接口返回403 Forbidden且X-Bogus生成函数报错原因execjs运行时环境如 Node.js版本过低无法执行抖音新引入的 ES2022 语法如at()方法、Object.hasOwn()。抖音前端 JS 每月更新签名函数随之升级。解决升级 Node.js 至 v18并验证execjs是否使用最新 runtime# 检查当前 runtime python -c import execjs; print(execjs.get().name) # 若为 PyExecJS需切换至 Node pip uninstall PyExecJS npm install -g node # 确保系统 PATH 包含 node并在 Python 中强制指定 runtimeimport execjs execjs.runtime_names execjs.get(Node) # 强制使用 Node.js5. 数据落库与去重用 SQLite 实现轻量级、可回溯的内容仓库采集来的视频数据若只存 CSV很快会面临三个问题重复入库同一视频多次出现在不同搜索页、字段扩展难新增音乐信息需改 CSV 结构、查询慢百万级数据grep效率低下。本项目采用 SQLite 作为默认存储后端因其零配置、单文件、ACID 支持、Python 内置完美匹配中小规模采集场景。核心设计是“双表结构”videos主表存视频元数据search_logs日志表存每次搜索的上下文实现数据可溯源。5.1 数据库 Schema 设计为什么aweme_id是主键search_keyword存在日志表而非主表SQLite 表结构定义在storage/schema.sql中-- videos 表视频主数据 CREATE TABLE IF NOT EXISTS videos ( aweme_id TEXT PRIMARY KEY, -- 抖音唯一 ID不可变 desc TEXT, -- 视频描述 create_time INTEGER, -- 创建时间戳秒 play_count INTEGER DEFAULT 0, digg_count INTEGER DEFAULT 0, comment_count INTEGER DEFAULT 0, share_count INTEGER DEFAULT 0, author_id TEXT, author_name TEXT, cover_url TEXT, play_url TEXT, music_title TEXT, is_ads BOOLEAN DEFAULT FALSE, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- search_logs 表搜索行为日志 CREATE TABLE IF NOT EXISTS search_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT NOT NULL, offset INTEGER NOT NULL, count INTEGER NOT NULL, total_found INTEGER, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, video_ids TEXT -- JSON 数组存本次拉取的 aweme_id 列表 );注意videos表的aweme_id是主键确保同一视频不会重复插入search_logs表的video_ids字段存 JSON 字符串如[7312345678901234567, 7312345678901234568]便于回溯“某次搜索命中了哪些视频”。不把keyword放videos表是因为一个视频可能被多个关键词搜到冗余存储违反范式。5.2 插入逻辑INSERT OR REPLACE与json1扩展的协同使用Python 中使用sqlite3模块插入数据关键点在于1用INSERT OR REPLACE避免主键冲突2用 SQLite 的json1扩展解析video_ids3事务包裹保证原子性。代码实现在storage/db.py# storage/db.py import sqlite3 import json from datetime import datetime class VideoDB: def __init__(self, db_path: str data/douyin.db): self.conn sqlite3.connect(db_path) self.conn.row_factory sqlite3.Row # 支持字典式取值 self._init_db() def _init_db(self): with open(storage/schema.sql) as f: self.conn.executescript(f.read()) self.conn.commit() def insert_videos(self, videos: list): 插入视频列表自动去重 cursor self.conn.cursor() now datetime.now().isoformat() # Insert into videos table video_data [ ( v[aweme_id], v[desc], v[create_time], v[play_count], v[digg_count], v[comment_count], v[share_count], v[author_id], v[author_name], v[cover_url], v[play_url], v[music_title], v[is_ads], now ) for v in videos ] cursor.executemany( INSERT OR REPLACE INTO videos VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?), video_data ) # Insert into search_logs keyword videos[0].get(search_keyword, unknown) if videos else unknown video_ids [v[aweme_id] for v in videos] cursor.execute( INSERT INTO search_logs (keyword, offset, count, total_found, video_ids) VALUES (?, ?, ?, ?, ?), (keyword, 0, len(videos), len(videos), json.dumps(video_ids)) ) self.conn.commit() def get_videos_by_keyword(self, keyword: str, limit: int 100) - list: 根据关键词查视频通过日志表 JOIN cursor self.conn.cursor() cursor.execute( SELECT v.* FROM videos v JOIN search_logs l ON json_each(l.video_ids) v.aweme_id WHERE l.keyword ? ORDER BY v.create_time DESC LIMIT ? , (keyword, limit)) return [dict(row) for row in cursor.fetchall()]逻辑说明insert_videos()先批量插入videos表用INSERT OR REPLACE覆盖旧记录如播放量更新再插入search_logs记录本次采集上下文。get_videos_by_keyword()利用 SQLite 的json_each()表值函数将video_idsJSON 数组展开为行再与videos表 JOIN实现高效关联查询。参数说明videos是parse_video_item()返回的字典列表limit控制返回数量避免大结果集阻塞。5.3 去重与更新策略如何判断一条视频是“新数据”还是“数据更新”抖音 Web 的aweme_info中play_count、digg_count等统计字段会随时间增长。若每次采集都INSERT OR REPLACE会丢失“历史快照”。本项目采用“增量更新”策略只更新统计字段不覆盖desc、cover_url等静态字段。修改insert_videos()中的 SQL# 替换原 INSERT OR REPLACE 语句 cursor.executemany( INSERT INTO videos (aweme_id, desc, create_time, author_id, author_name, cover_url, play_url, music_title, is_ads, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(aweme_id) DO UPDATE SET play_count excluded.play_count, digg_count excluded.digg_count, comment_count excluded.comment_count, share_count excluded.share_count, updated_at excluded.updated_at , video_data )逻辑说明SQLite 3.24 支持ON CONFLICT DO UPDATE语法。excluded是伪表代表本次 INSERT 尝试插入但冲突的那行数据。这样aweme_id相同的视频只更新统计字段保留原始描述和封面。参数说明video_data元组中前 9 个字段为INSERT字段后 5 个为UPDATE字段play_count等顺序必须严格对应。6. 生产就绪技巧用schedulelogging构建可持续的采集管道一个能跑一周的脚本和一个能跑一年的管道差距不在算法而在可观测性与容错设计。本项目不追求“全自动”而是提供一套最小可行的生产化骨架定时任务调度、结构化日志、失败重试、状态监控。核心是把“采集”变成可诊断、可暂停、可审计的操作。6.1 定时任务用schedule替代cron实现跨平台、可热更的调度逻辑Linuxcron和 Windows 任务计划程序配置分散、不易调试。本项目用 Python 的schedule库在代码内定义任务启动即生效且支持运行时修改如临时停用某任务。scheduler.py示例# scheduler.py import schedule import time import logging from modules.search import search_videos from parsers.video_parser import parse_video_item from storage.db import VideoDB db VideoDB() def job_search_cat(): try: logging.info(Starting search for 猫) data search_videos(猫, offset0, count20) parsed [parse_video_item(item) for item in data[data][data]] db.insert_videos(parsed) logging.info(fSearch 猫 done. Fetched {len(parsed)} videos.) except Exception as e: logging.error(fSearch 猫 failed: {e}) def job_search_dog(): try: logging.info(Starting search for 狗) data search_videos(狗, offset0, count20) parsed [parse_video_item(item) for item in data[data][data]] db.insert_videos(parsed) logging.info(fSearch 狗 done. Fetched {len(parsed)} videos.) except Exception as e: logging.error(fSearch 狗 failed: {e}) # 每 2 小时执行一次猫搜索 schedule.every(2).hours.do(job_search_cat) # 每 3 小时执行一次狗搜索 schedule.every(3).hours.do(job_search_dog) # 每天凌晨 2 点清理 30 天前的日志 schedule.every().day.at(02:00).do(lambda: logging.info(Log cleanup scheduled)) if __name__ __main__: logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(logs/douyin_scheduler.log), logging.StreamHandler() ] ) logging.info(Scheduler started.) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次逻辑说明schedule的do()方法接受任意 callable可传入带参数的函数用lambda包裹run_pending()是非阻塞的配合time.sleep()实现轻量轮询。参数说明every(2).hours表示间隔 2 小时at(02:00)表示每天固定时间二者可混用。6.2 结构化日志用jsonlogger记录每条请求的完整上下文普通print()或logging.info()输出的是字符串难以用 ELK 或 Grafana 分析。本项目集成python-json-logger将每次请求的关键字段URL、参数、状态码、耗时、错误堆栈记为 JSON 行# utils/json_logger.py import logging from pythonjsonlogger import jsonlogger def setup_json_logger(): logger logging p a hrefhttps://download.csdn.net/download/weixin_42165508/18224811 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p