簡(jiǎn)介這份資源是一個(gè)用Python爬取百度遷徙數(shù)據(jù)的腳本壓縮包面向具備Python基礎(chǔ)、希望入門(mén)網(wǎng)絡(luò)爬蟲(chóng)與數(shù)據(jù)解析的開(kāi)發(fā)者解決按日期獲取全國(guó)熱門(mén)城市人口遷徙數(shù)據(jù)的問(wèn)題。包內(nèi)共1個(gè)文件為單個(gè)py腳本壓縮包約1KB輕量易讀可直接運(yùn)行并修改日期參數(shù)。腳本涉及requests發(fā)送HTTP請(qǐng)求、json解析返回?cái)?shù)據(jù)、異常捕獲處理以及文件保存等環(huán)節(jié)幫助讀者理解從接口請(qǐng)求到數(shù)據(jù)落地的完整流程。已有424人學(xué)習(xí)說(shuō)明該案例在爬蟲(chóng)入門(mén)場(chǎng)景中具有一定參考價(jià)值。通過(guò)閱讀與調(diào)試這份代碼讀者可以掌握按日期抓取遷徙數(shù)據(jù)的方法并在此基礎(chǔ)上擴(kuò)展數(shù)據(jù)可視化或批量采集適合作為爬蟲(chóng)練手與數(shù)據(jù)獲取的起點(diǎn)。1. 百度遷徙數(shù)據(jù)到底能拿到什么先搞清楚數(shù)據(jù)邊界再動(dòng)手很多人第一次搜「python爬取百度遷徙數(shù)據(jù)」腦子里想的是把全國(guó)人口流動(dòng)的原始記錄整表拉下來(lái)。實(shí)際打開(kāi)百度遷徙頁(yè)面會(huì)發(fā)現(xiàn)它對(duì)外展示的是指數(shù)化的遷徙規(guī)模不是絕對(duì)人次。頁(yè)面按城市對(duì)城市給出遷入遷出比例按日期切換背后是一套歸一化后的相對(duì)值。這意味著你能做的是趨勢(shì)對(duì)比、城市熱度排序、節(jié)前節(jié)后流向變化而不是精確統(tǒng)計(jì)「今天有多少人從A市到了B市」。這個(gè)區(qū)別決定了整個(gè)爬取方案的設(shè)計(jì)。如果目標(biāo)是做春運(yùn)人口流動(dòng)可視化、區(qū)域經(jīng)濟(jì)分析、或者給選址模型加一個(gè)「人口活躍度」特征指數(shù)數(shù)據(jù)完全夠用。但如果要做精確客流預(yù)測(cè)這套數(shù)據(jù)只能當(dāng)輔助信號(hào)。我一般會(huì)先跟需求方確認(rèn)這一點(diǎn)避免后面返工。適合讀這篇的人有三類(lèi)會(huì)一點(diǎn) Python 基礎(chǔ)、想拿真實(shí)數(shù)據(jù)練手爬蟲(chóng)的做數(shù)據(jù)分析需要人口流動(dòng)特征的以及想搭一個(gè)定時(shí)采集管道、長(zhǎng)期跟蹤城市遷徙趨勢(shì)的。下面從接口結(jié)構(gòu)講到落地代碼再到長(zhǎng)期跑會(huì)遇到的問(wèn)題按能復(fù)現(xiàn)的路徑走。2. 接口結(jié)構(gòu)與請(qǐng)求參數(shù)把遷徙數(shù)據(jù)的入口摸清楚2.1 頁(yè)面數(shù)據(jù)是怎么加載出來(lái)的百度遷徙的頁(yè)面本身不直接嵌數(shù)據(jù)前端通過(guò)異步請(qǐng)求拿 JSON 再渲染。用瀏覽器開(kāi)發(fā)者工具的 Network 面板切換到 XHR/Fetch 過(guò)濾刷新頁(yè)面就能看到返回遷徙數(shù)據(jù)的請(qǐng)求。常見(jiàn)做法是觀察請(qǐng)求 URL 里的幾個(gè)關(guān)鍵參數(shù)城市代碼、日期、遷移方向遷入/遷出、時(shí)間粒度。這里有個(gè)容易翻車(chē)的點(diǎn)頁(yè)面上的城市名和接口里的城市代碼不是一回事。接口用的是內(nèi)部編碼你得先拿到一份城市名到代碼的映射。這個(gè)映射通常也能從頁(yè)面加載的某個(gè)配置請(qǐng)求里拿到或者從城市選擇列表的接口里提取。我一般會(huì)先把這份映射抓下來(lái)存成本地 JSON后面所有請(qǐng)求都基于它避免每次硬編碼。請(qǐng)求里通常還帶一個(gè)時(shí)間戳或簽名參數(shù)這類(lèi)參數(shù)往往和當(dāng)前時(shí)間掛鉤。如果直接復(fù)制瀏覽器里的 URL 去請(qǐng)求過(guò)一段時(shí)間就會(huì)失效。所以方案不能是「抄一個(gè) URL 反復(fù)用」而要把參數(shù)拼裝邏輯還原出來(lái)。2.2 用 requests 拼出第一個(gè)可用請(qǐng)求先裝依賴(lài)再寫(xiě)最小請(qǐng)求。下面這段是探路用的目的是確認(rèn)參數(shù)拼對(duì)了、能拿到 JSON。import requests import json import time # 城市代碼映射實(shí)際應(yīng)從配置接口獲取這里示意結(jié)構(gòu) CITY_CODE { 北京: 110000, 上海: 310000, 廣州: 440100, } def fetch_migration(city_code, date_str, directionin): city_code: 城市內(nèi)部編碼 date_str: 形如 20240101 direction: in 表示遷入out 表示遷出 url https://huiyan.baidu.com/migration/cityrank.jsonp params { dt: city, id: city_code, type: move, # move 表示遷徙 date: date_str, direction: direction, } headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Referer: https://huiyan.baidu.com/migration/, } resp requests.get(url, paramsparams, headersheaders, timeout10) resp.raise_for_status() text resp.text # jsonp 需要?jiǎng)兊艋卣{(diào)函數(shù)外殼 start text.find(() end text.rfind()) data json.loads(text[start 1:end]) return data if __name__ __main__: result fetch_migration(CITY_CODE[北京], 20240101, in) print(json.dumps(result, ensure_asciiFalse)[:500]) time.sleep(1)邏輯說(shuō)明這段代碼做了三件事——拼參數(shù)、發(fā)請(qǐng)求、剝 jsonp 外殼。params里的dt、type、direction是接口區(qū)分?jǐn)?shù)據(jù)維度的關(guān)鍵改direction就切換遷入遷出改date就切換日期。headers里帶Referer是因?yàn)椴糠纸涌跁?huì)校驗(yàn)來(lái)源不帶可能返回空數(shù)據(jù)。參數(shù)說(shuō)明timeout10防止請(qǐng)求卡死time.sleep(1)是給后續(xù)批量請(qǐng)求留的節(jié)流位單次請(qǐng)求看不出作用批量時(shí)是保命的。jsonp的剝殼邏輯用find和rfind而不是固定切片是因?yàn)榛卣{(diào)函數(shù)名可能變化固定位置容易翻車(chē)。提示接口地址和參數(shù)名可能隨頁(yè)面改版變化上面是結(jié)構(gòu)示意。實(shí)際動(dòng)手時(shí)以你在 Network 面板里看到的真實(shí)請(qǐng)求為準(zhǔn)把參數(shù)名對(duì)齊。2.3 城市代碼映射怎么批量拿單個(gè)城市能跑通后下一步是把城市列表拿全。常見(jiàn)做法是找到城市選擇列表對(duì)應(yīng)的接口它一般返回一個(gè)帶層級(jí)結(jié)構(gòu)的 JSON包含省、市、區(qū)。解析時(shí)注意兩點(diǎn)一是直轄市和省的結(jié)構(gòu)可能不同二是有些城市有別名。我一般會(huì)把原始返回存一份再寫(xiě)一個(gè)清洗函數(shù)生成{城市名: 代碼}的扁平字典存成city_code.json。def build_city_map(raw_list): mapping {} for province in raw_list: for city in province.get(city, []): name city.get(name) code city.get(code) if name and code: mapping[name] code return mapping # 假設(shè) raw 是從城市列表接口拿到的原始結(jié)構(gòu) # with open(city_code.json, w, encodingutf-8) as f: # json.dump(build_city_map(raw), f, ensure_asciiFalse, indent2)這段的價(jià)值在于把「每次手動(dòng)查代碼」變成「一次生成、長(zhǎng)期復(fù)用」。參數(shù)上沒(méi)什么可調(diào)的重點(diǎn)是清洗時(shí)對(duì)空值和層級(jí)差異做防御否則后面按城市名取代碼時(shí)會(huì) KeyError。3. 批量采集與存儲(chǔ)從單次請(qǐng)求到可復(fù)用數(shù)據(jù)集3.1 按日期和城市雙維度循環(huán)單城市單日期跑通后真正的數(shù)據(jù)集需要「多城市 × 多日期」。這里要設(shè)計(jì)循環(huán)順序和節(jié)流策略。我一般外層循環(huán)日期、內(nèi)層循環(huán)城市因?yàn)橥蝗掌诘恼?qǐng)求參數(shù)結(jié)構(gòu)一致便于復(fù)用 session。import requests import json import time import random session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Referer: https://huiyan.baidu.com/migration/, }) def fetch_one(city_code, date_str, direction): url https://huiyan.baidu.com/migration/cityrank.jsonp params { dt: city, id: city_code, type: move, date: date_str, direction: direction, } resp session.get(url, paramsparams, timeout10) text resp.text data json.loads(text[text.find(() 1:text.rfind())]) return data def crawl_range(city_map, dates, directionin): records [] for date_str in dates: for name, code in city_map.items(): try: data fetch_one(code, date_str, direction) records.append({ date: date_str, city: name, code: code, direction: direction, payload: data, }) except Exception as e: print(ffail {name} {date_str}: {e}) # 隨機(jī)節(jié)流降低被限流的概率 time.sleep(random.uniform(0.8, 1.8)) return records邏輯說(shuō)明用Session復(fù)用連接比每次新建請(qǐng)求快。異常捕獲放在內(nèi)層單個(gè)城市失敗不影響整體。random.uniform做隨機(jī)節(jié)流比固定 sleep 更接近人工節(jié)奏。參數(shù)說(shuō)明節(jié)流區(qū)間0.8~1.8秒是我常用的起點(diǎn)城市數(shù)量多時(shí)可以適當(dāng)放大。direction作為參數(shù)傳入方便同一套邏輯跑遷入和遷出兩遍。3.2 存成什么格式JSON 還是 CSV原始返回是嵌套 JSON直接存 JSON 最省事但后續(xù)分析不方便。我的做法是兩層存儲(chǔ)原始層存 JSON 保底分析層拍平成 CSV。存儲(chǔ)層格式用途注意點(diǎn)原始層JSON留檔、重解析按日期分文件避免單文件過(guò)大分析層CSV直接進(jìn) pandas字段拍平一行一條記錄映射層JSON城市代碼復(fù)用單獨(dú)維護(hù)不混在數(shù)據(jù)里拍平的時(shí)候把嵌套里的城市對(duì)、比例值展開(kāi)成列。下面是一個(gè)拍平函數(shù)示意。import pandas as pd def flatten(records): rows [] for rec in records: payload rec[payload] # 具體字段名以實(shí)際返回為準(zhǔn)這里示意 for item in payload.get(data, {}).get(list, []): rows.append({ date: rec[date], city: rec[city], direction: rec[direction], target_city: item.get(city_name), value: item.get(value), }) return pd.DataFrame(rows) # df flatten(records) # df.to_csv(migration_flat.csv, indexFalse, encodingutf-8-sig)encodingutf-8-sig是為了 Excel 打開(kāi)不亂碼這個(gè)細(xì)節(jié)很多人第一次導(dǎo)出 CSV 時(shí)會(huì)踩。字段名city_name、value是示意實(shí)際以接口返回為準(zhǔn)拍平前先打印一條完整記錄看清楚結(jié)構(gòu)。3.3 斷點(diǎn)續(xù)采別讓一次中斷白跑批量跑幾百個(gè)城市乘幾十天中途網(wǎng)絡(luò)抖動(dòng)或限流很常見(jiàn)。沒(méi)有斷點(diǎn)機(jī)制的話一次失敗就得從頭來(lái)。我的習(xí)慣是每采完一個(gè)日期就把該日期的結(jié)果落盤(pán)并在一個(gè)進(jìn)度文件里記錄已完成日期。import os def save_progress(done_dates, pathprogress.json): with open(path, w, encodingutf-8) as f: json.dump({done: done_dates}, f) def load_progress(pathprogress.json): if os.path.exists(path): with open(path, encodingutf-8) as f: return json.load(f).get(done, []) return []啟動(dòng)時(shí)先讀進(jìn)度跳過(guò)已完成日期。這個(gè)機(jī)制看起來(lái)簡(jiǎn)單但在長(zhǎng)時(shí)間采集里是后悔藥級(jí)別的存在。參數(shù)上沒(méi)什么玄學(xué)關(guān)鍵是落盤(pán)時(shí)機(jī)要夠頻繁我一般按日期粒度落不按城市粒度平衡 IO 和恢復(fù)成本。4. 避坑與排查遷徙數(shù)據(jù)采集里最容易翻車(chē)的幾件事4.1 請(qǐng)求返回空數(shù)據(jù)或結(jié)構(gòu)突變現(xiàn)象某天開(kāi)始原本正常的請(qǐng)求返回空 list或者 JSON 結(jié)構(gòu)里字段名變了。原因頁(yè)面改版或接口調(diào)整參數(shù)名、返回字段都可能變。這類(lèi)接口沒(méi)有對(duì)外承諾穩(wěn)定性。解決不要假設(shè)結(jié)構(gòu)不變。在拍平函數(shù)里對(duì)字段做防御性取值用.get()加默認(rèn)值同時(shí)保留原始 JSON結(jié)構(gòu)變了可以重新解析歷史數(shù)據(jù)不用重采。發(fā)現(xiàn)空數(shù)據(jù)時(shí)先手動(dòng)在瀏覽器里復(fù)現(xiàn)一次請(qǐng)求對(duì)比參數(shù)差異。4.2 高頻請(qǐng)求被限流現(xiàn)象跑了幾十個(gè)城市后后續(xù)請(qǐng)求開(kāi)始超時(shí)或返回錯(cuò)誤碼。原因請(qǐng)求頻率過(guò)高觸發(fā)服務(wù)端限流。解決加大隨機(jī)節(jié)流區(qū)間必要時(shí)在每采完一個(gè)日期后插入更長(zhǎng)停頓。我一般會(huì)在連續(xù)失敗達(dá)到閾值時(shí)自動(dòng)暫停幾分鐘再繼續(xù)。別用固定 sleep固定節(jié)奏反而容易被識(shí)別。4.3 城市代碼對(duì)不上導(dǎo)致數(shù)據(jù)錯(cuò)位現(xiàn)象采集回來(lái)的數(shù)據(jù)里城市名和預(yù)期不符或者某些城市一直失敗。原因城市映射里存在別名、層級(jí)差異或者代碼過(guò)期。解決采集前先做一次映射校驗(yàn)?zāi)脦讉€(gè)已知城市試請(qǐng)求確認(rèn)返回的城市名和映射一致。映射文件單獨(dú)維護(hù)發(fā)現(xiàn)對(duì)不上就更新不要在每個(gè)腳本里硬編碼。4.4 日期格式和時(shí)區(qū)問(wèn)題現(xiàn)象某些日期的數(shù)據(jù)查不到或者查到的日期和預(yù)期差一天。原因接口對(duì)日期格式敏感且數(shù)據(jù)可能有更新延遲。解決統(tǒng)一用YYYYMMDD字符串采集前先確認(rèn)目標(biāo)日期數(shù)據(jù)已更新。做時(shí)間序列分析時(shí)把采集日期和數(shù)據(jù)日期分開(kāi)存避免混淆。4.5 數(shù)據(jù)指數(shù)化帶來(lái)的誤讀現(xiàn)象拿指數(shù)值直接當(dāng)人次做加總或預(yù)測(cè)結(jié)論離譜。原因指數(shù)是歸一化后的相對(duì)值不同日期、不同城市的基準(zhǔn)可能不同。解決只用它做同城市跨時(shí)間的趨勢(shì)對(duì)比或同時(shí)間跨城市的排序?qū)Ρ取Ｒ隹缇S度分析先確認(rèn)歸一化基準(zhǔn)是否一致。這一點(diǎn)在需求溝通階段就要說(shuō)清楚否則后面分析全白做。5. 讓數(shù)據(jù)真正可用校驗(yàn)、增量更新與一個(gè)長(zhǎng)期習(xí)慣采集只是第一步數(shù)據(jù)能不能用取決于你有沒(méi)有做校驗(yàn)和增量維護(hù)。我一般會(huì)在拍平后跑幾個(gè)檢查記錄數(shù)是否和預(yù)期城市數(shù)乘日期數(shù)吻合、有沒(méi)有大段缺失日期、同一城市同一日期的遷入遷出是否都齊。缺失的地方標(biāo)記出來(lái)而不是直接丟進(jìn)分析。增量更新是長(zhǎng)期跑的關(guān)鍵。我的做法是維護(hù)一個(gè)「已采集日期」清單每次運(yùn)行只采新增日期歷史數(shù)據(jù)不動(dòng)。這樣每天或每周跑一次成本很低。下面是一個(gè)增量入口的示意。def incremental_dates(all_dates, done_dates): done set(done_dates) return [d for d in all_dates if d not in done] # dates [20240101, 20240102, 20240103] # done load_progress() # todo incremental_dates(dates, done) # records crawl_range(city_map, todo) # save_progress(done todo)參數(shù)上all_dates可以按需生成比如最近 30 天。done_dates從進(jìn)度文件讀。這個(gè)模式跑順之后你手里就有了一份持續(xù)更新的城市遷徙數(shù)據(jù)集比一次性采集有價(jià)值得多。還有一個(gè)我踩過(guò)的坑早期我沒(méi)做數(shù)據(jù)版本管理后來(lái)接口字段變了想回溯舊數(shù)據(jù)發(fā)現(xiàn)原始 JSON 被覆蓋了。從那以后我養(yǎng)成的習(xí)慣是原始層按日期分文件、只增不改分析層可以隨時(shí)從原始層重建。這個(gè)習(xí)慣讓我在后來(lái)的多次接口變動(dòng)里都沒(méi)慌過(guò)。如果你打算把這個(gè)方向做成長(zhǎng)期項(xiàng)目建議先把城市映射和進(jìn)度機(jī)制搭好再考慮加定時(shí)任務(wù)。數(shù)據(jù)本身不難采難的是穩(wěn)定地采、可復(fù)現(xiàn)地存、清楚地知道它的邊界在哪。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取