查與分析源碼解析 新手不再盲目調(diào)錯)
5步搞定調(diào)查與分析源碼解析 新手不再盲目調(diào)錯
復(fù)制來的代碼跑不通不知道怎么調(diào)，是不是也讓你抓狂？別慌，今天咱們就拆解調(diào)查與分析里的核心邏輯。
很多新手在搞數(shù)據(jù)處理或邏輯驗證時，習(xí)慣直接復(fù)制網(wǎng)上現(xiàn)成的腳本。結(jié)果一運行，報錯滿天飛，或者結(jié)果完全不對。這時候，光看報錯信息是解決不了問題的，必須深入到源碼層面去理解。源碼解析不是讓你死記硬行，而是幫你建立對代碼執(zhí)行流程的直覺。比如一個看似簡單的數(shù)據(jù)清洗函數(shù)，背后可能藏著空值處理、類型轉(zhuǎn)換的陷阱。如果你不去看它內(nèi)部怎么寫的，只盯著輸入輸出，那永遠是在碰運氣。
在CSDN等社區(qū)里，經(jīng)?？吹筋愃频那笾骸盀槭裁催@段代碼在我電腦上能跑，換了數(shù)據(jù)就崩了？”答案往往藏在那些不起眼的邊界條件里。今天我們就以一個典型的“調(diào)查數(shù)據(jù)清洗與分析”實戰(zhàn)項目為例，從零搭建，帶你把這套源碼扒開揉碎講清楚。
項目目標(biāo)
咱們這個項目的核心目標(biāo)很明確：處理一份模擬的用戶調(diào)查原始數(shù)據(jù)，清洗臟數(shù)據(jù)，并輸出結(jié)構(gòu)化的分析結(jié)果。
原始數(shù)據(jù)通常很“臟”，比如：字段缺失：有的用戶沒填年齡。
格式混亂：年齡有的是字符串“25”，有的是數(shù)字25.0，還有的是“二十五”。
邏輯錯誤：年齡填了200，或者-5。我們的目標(biāo)代碼需要做到：自動識別并剔除無效記錄。
將非標(biāo)準(zhǔn)格式統(tǒng)一為標(biāo)準(zhǔn)類型。
輸出清洗后的干凈數(shù)據(jù)以及簡單的統(tǒng)計摘要。這不是為了炫技，而是為了讓你理解：一個合格的“調(diào)查與分析”模塊，到底需要處理哪些細節(jié)。只有把這些細節(jié)搞清楚，下次你復(fù)制別人的代碼時，才知道該改哪里。
目錄結(jié)構(gòu)
為了保持工程化思維，我們不能把所有代碼塞在一個文件里。即使是小項目，也要有清晰的結(jié)構(gòu)。下面是我們推薦的最小化目錄結(jié)構(gòu)：
investigation_analysis/
├── data/
│ └── raw_survey.csv # 原始臟數(shù)據(jù)
├── src/
│ ├── __init__.py
│ ├── cleaner.py # 數(shù)據(jù)清洗邏輯
│ ├── analyzer.py # 分析邏輯
│ └── main.py # 入口文件
├── tests/
│ ├── __init__.py
│ └── test_cleaner.py # 單元測試
├── requirements.txt # 依賴庫
└── README.md # 項目說明為什么要這樣分？cleaner.py：專門負責(zé)“洗”數(shù)據(jù)。輸入是原始列表，輸出是干凈列表。
analyzer.py：專門負責(zé)“算”數(shù)據(jù)。輸入是干凈列表，輸出是統(tǒng)計結(jié)果。
main.py：負責(zé)串聯(lián)流程，讀取文件，調(diào)用清洗，調(diào)用分析，最后打印結(jié)果。這種分層設(shè)計的好處是，當(dāng)你發(fā)現(xiàn)某個環(huán)節(jié)出問題時，可以單獨調(diào)試該模塊，而不需要盯著整個長腳本發(fā)呆。這就是工程化的第一步：關(guān)注點分離。
核心代碼實現(xiàn)
接下來是重頭戲。我們不看那種幾十行的復(fù)雜庫，而是用最基礎(chǔ)的Python邏輯，把源碼解析做到極致。
1. 數(shù)據(jù)清洗模塊 cleaner.py
這是最容易出Bug的地方。很多人復(fù)制代碼時，只復(fù)制了if age 18，卻忽略了age可能根本不是一個數(shù)字。
import re
import logging# 配置日志，方便調(diào)試時查看中間狀態(tài)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)def clean_age(value):清洗年齡字段:param value: 原始年齡值，可能是str, int, float, None:return: 清洗后的整數(shù)年齡，如果無效則返回Noneif value is None:logger.debug(年齡字段為空，跳過)return None# 處理字符串類型的數(shù)字if isinstance(value, str):# 去除空格value = value.strip()if not value:return None# 嘗試直接轉(zhuǎn)換為整數(shù)try:# 這里用int()而不是float()，因為年齡通常是整數(shù)# 如果失敗，說明包含非數(shù)字字符age_int = int(value)except ValueError:# 如果是25.5這種，int()會報錯# 我們嘗試轉(zhuǎn)float再取整，或者根據(jù)業(yè)務(wù)決定丟棄try:age_float = float(value)age_int = int(age_float)logger.debug(f將浮點數(shù)字符串 '{value}' 轉(zhuǎn)為整數(shù) {age_int})except ValueError:# 如果是中文數(shù)字如二十五，需要更復(fù)雜的映射# 這里為了簡化，直接標(biāo)記為無效logger.warning(f無法識別的年齡格式: '{value}')return None# 校驗?zāi)挲g合理性if not (0 age_int 150):logger.warning(f年齡不合理: {age_int})return Nonereturn age_int# 處理數(shù)值類型if isinstance(value, (int, float)):age_int = int(value)if not (0 age_int 150):logger.warning(f數(shù)值年齡不合理: {age_int})return Nonereturn age_int# 其他類型一律無效logger.warning(f不支持的年齡類型: {type(value)})return Nonedef clean_survey_data(raw_records):清洗整批調(diào)查數(shù)據(jù):param raw_records: 原始記錄列表，每個元素是一個字典:return: 清洗后的記錄列表cleaned_records = []for i, record in enumerate(raw_records):new_record = {}# 1. 清洗ID，確保唯一且非空user_id = record.get('id')if not user_id:logger.warning(f第{i}條記錄缺少ID，丟棄)continuenew_record['id'] = str(user_id)# 2. 清洗年齡age = clean_age(record.get('age'))if age is None:logger.warning(f第{i}條記錄(ID:{user_id})年齡無效，丟棄)continuenew_record['age'] = age# 3. 清洗滿意度評分 (1-5)rating = record.get('rating')if rating is None:# 這里可以選擇默認值或者丟棄，根據(jù)業(yè)務(wù)需求# 假設(shè)調(diào)查必須評分，所以丟棄logger.warning(f第{i}條記錄(ID:{user_id})缺少評分，丟棄)continuetry:rating_int = int(rating)if not (1 = rating_int = 5):logger.warning(f第{i}條記錄評分超出范圍: {rating_int})continueexcept (ValueError, TypeError):logger.warning(f第{i}條記錄評分格式錯誤: {rating})continuenew_record['rating'] = rating_int# 4. 保留其他合法字段if 'comment' in record and isinstance(record['comment'], str):new_record['comment'] = record['comment'].strip()cleaned_records.append(new_record)logger.info(f清洗完成: 原始 {len(raw_records)} 條, 有效 {len(cleaned_records)} 條)return cleaned_records源碼解析關(guān)鍵點：類型判斷前置：在轉(zhuǎn)換之前，先判斷isinstance。這是防止TypeError的關(guān)鍵。
異常捕獲細化：try-except塊里，區(qū)分了int()失敗和float()失敗的情況。很多新手只寫一個except Exception，導(dǎo)致問題被掩蓋。
日志記錄：logger不是裝飾，它是你調(diào)試時的眼睛。當(dāng)數(shù)據(jù)量大了，你不可能逐行打印，日志能幫你快速定位哪條數(shù)據(jù)被丟棄了。2. 分析模塊 analyzer.py
數(shù)據(jù)干凈了，接下來算指標(biāo)。這里我們計算平均年齡和平均評分。
from collections import defaultdictdef analyze_survey_data(cleaned_records):對清洗后的數(shù)據(jù)進行基本分析:param cleaned_records: 清洗后的記錄列表:return: 包含統(tǒng)計結(jié)果的字典if not cleaned_records:return {total_count: 0,avg_age: 0,avg_rating: 0,error: No valid data}total_age = 0total_rating = 0count = len(cleaned_records)for record in cleaned_records:total_age += record['age']total_rating += record['rating']# 防止除以零，雖然前面判斷了count0，但好習(xí)慣要保留avg_age = round(total_age / count, 2) if count 0 else 0avg_rating = round(total_rating / count, 2) if count 0 else 0return {total_count: count,avg_age: avg_age,avg_rating: avg_rating}注意：這里的邏輯很簡單，但重點在于輸入契約。analyze_survey_data只接受cleaned_records。如果上游數(shù)據(jù)沒洗干凈，這里就會報錯。這就是為什么我們要把清洗和分析分開。
運行與測試
代碼寫好了，不能只靠肉眼檢查。我們需要寫一個簡單的單元測試，來驗證我們的“源碼解析”是否準(zhǔn)確。
1. 準(zhǔn)備測試數(shù)據(jù)
在tests/test_cleaner.py中：
import unittest
from src.cleaner import clean_survey_dataclass TestCleaner(unittest.TestCase):def setUp(self):self.raw_data = [{id: 1, age: 25, rating: 4, comment: Good},{id: 2, age: 30.5, rating: 5, comment: Excellent},{id: 3, age: abc, rating: 3, comment: Bad}, # 年齡無效{id: 4, age: 200, rating: 2, comment: Old}, # 年齡不合理{id: 5, age: None, rating: 1, comment: No age}, # 年齡缺失{id: 6, age: 40, rating: high, comment: Rating bad}, # 評分無效{id: 7, age: 45, rating: 5, comment: Great}]def test_clean_survey_data(self):cleaned = clean_survey_data(self.raw_data)# 預(yù)期只有 id 1, 2, 7 是有效的self.assertEqual(len(cleaned), 3)# 驗證第一條self.assertEqual(cleaned[0]['id'], '1')self.assertEqual(cleaned[0]['age'], 25)self.assertEqual(cleaned[0]['rating'], 4)# 驗證第二條，浮點數(shù)轉(zhuǎn)整數(shù)self.assertEqual(cleaned[1]['id'], '2')self.assertEqual(cleaned[1]['age'], 30)# 驗證第三條self.assertEqual(cleaned[2]['id'], '7')self.assertEqual(cleaned[2]['age'], 45)print(所有測試通過!)if __name__ == '__main__':unittest.main()2. 運行主程序
src/main.py負責(zé)串聯(lián)：
import csv
import json
from src.cleaner import clean_survey_data
from src.analyzer import analyze_survey_datadef load_csv(filename):加載CSV文件為字典列表records = []try:with open(filename, 'r', encoding='utf-8') as f:reader = csv.DictReader(f)for row in reader:# 簡單轉(zhuǎn)換：將數(shù)字字符串轉(zhuǎn)為float或int，以便cleaner處理# 實際項目中可能需要更復(fù)雜的類型推斷records.append(row)except FileNotFoundError:print(f錯誤: 找不到文件 {filename})return []return recordsdef main():raw_file = 'data/raw_survey.csv'output_file = 'data/cleaned_result.json'print(f正在加載數(shù)據(jù): {raw_file})raw_records = load_csv(raw_file)if not raw_records:print(沒有加載到數(shù)據(jù)，退出)returnprint(開始清洗數(shù)據(jù)...)cleaned_records = clean_survey_data(raw_records)print(開始分析數(shù)據(jù)...)result = analyze_survey_data(cleaned_records)print(f分析結(jié)果: {json.dumps(result, indent=2, ensure_ascii=False)})# 可選：保存清洗后的數(shù)據(jù)# with open(output_file, 'w', encoding='utf-8') as f:# json.dump(cleaned_records, f, indent=2, ensure_ascii=False)# print(f清洗后數(shù)據(jù)已保存至 {output_file})if __name__ == '__main__':main()調(diào)試技巧：
如果運行報錯，不要只看最后一行。往上翻，看logger輸出。比如看到Warning: 第3條記錄年齡無效，你就知道是測試數(shù)據(jù)里的abc導(dǎo)致的，這時候你就該去檢查clean_age里的ValueError捕獲邏輯是否覆蓋了這種情況。
優(yōu)化擴展
基礎(chǔ)版跑通了，但離生產(chǎn)級還有距離。以下是幾個進階方向：配置化管理：
把年齡范圍（0-150）、評分范圍（1-5）提取到config.py或.yaml文件中。這樣如果業(yè)務(wù)規(guī)則變了，不用改代碼，只改配置。數(shù)據(jù)持久化：
目前數(shù)據(jù)在內(nèi)存中。實際項目中，原始數(shù)據(jù)可能在MySQL或PostgreSQL里。你需要用SQLAlchemy或pandas.read_sql來讀取，而不是csv。并發(fā)處理：
如果數(shù)據(jù)量達到百萬級，單線程清洗會很慢。可以使用multiprocessing或concurrent.futures來并行處理數(shù)據(jù)塊。但要注意，日志記錄可能會交錯，需要使用線程安全的日志處理器。錯誤重試機制：
如果是從API拉取數(shù)據(jù)，可能會失敗。需要加上retry裝飾器（如tenacity庫），自動重試3次，間隔遞增?？梢暬?分析結(jié)果不只是數(shù)字?？梢杂胢atplotlib或seaborn畫出年齡分布直方圖、評分餅圖。這一步能幫你快速發(fā)現(xiàn)數(shù)據(jù)中的異常模式，比如“為什么40歲以上的人評分特別低？”小結(jié)
通過這個小小的“調(diào)查與分析”項目，我們其實講透了一個核心思路：源碼解析的本質(zhì)，是對數(shù)據(jù)流向和控制流的掌控。
很多新手覺得代碼難懂，是因為他們只把代碼當(dāng)成“黑盒”。你丟進去數(shù)據(jù)，它吐出來結(jié)果，中間發(fā)生了什么，不管。但一旦出錯，黑盒就失效了。
我們剛才做的每一步：分目錄，是為了隔離關(guān)注點。
寫日志，是為了追蹤數(shù)據(jù)流。
寫測試，是為了驗證控制流。
處理異常，是為了覆蓋邊界條件。這些不是花架子，而是你從“代碼搬運工”變成“工程師”的分水嶺。下次再遇到復(fù)制來的代碼跑不通，別再盲目改參數(shù)。打開logger，加點斷點，順著數(shù)據(jù)流走一遍，問題往往就浮出水面了。
在CSDN或者StackOverflow上，高手的回答通常不是“試試重啟”，而是“檢查一下這里的空值判斷”。這就是差距所在。
最后，拋出一個問題引發(fā)討論：
在你實際工作中，有沒有遇到過那種“邏輯看起來沒錯，但跑出來結(jié)果就是不對”的代碼？你是怎么排查的？是加日志、看堆棧，還是干脆重寫？
還有什么不懂的？評論區(qū)留言挨個回。特別是關(guān)于類型轉(zhuǎn)換陷阱或者日志配置的具體寫法，歡迎提問。 5步搞定調(diào)查與分析源碼解析 新手不再盲目調(diào)錯
5步搞定調(diào)查與分析源碼解析 新手不再盲目調(diào)錯
復(fù)制來的代碼跑不通不知道怎么調(diào)，是不是也讓你抓狂？別慌，今天咱們就拆解調(diào)查與分析里的核心邏輯。
很多新手在搞數(shù)據(jù)處理或邏輯驗證時，習(xí)慣直接復(fù)制網(wǎng)上現(xiàn)成的腳本。結(jié)果一運行，報錯滿天飛，或者結(jié)果完全不對。這時候，光看報錯信息是解決不了問題的，必須深入到源碼層面去理解。源碼解析不是讓你死記硬行，而是幫你建立對代碼執(zhí)行流程的直覺。比如一個看似簡單的數(shù)據(jù)清洗函數(shù)，背后可能藏著空值處理、類型轉(zhuǎn)換的陷阱。如果你不去看它內(nèi)部怎么寫的，只盯著輸入輸出，那永遠是在碰運氣。
在CSDN等社區(qū)里，經(jīng)?？吹筋愃频那笾骸盀槭裁催@段代碼在我電腦上能跑，換了數(shù)據(jù)就崩了？”答案往往藏在那些不起眼的邊界條件里。今天我們就以一個典型的“調(diào)查數(shù)據(jù)清洗與分析”實戰(zhàn)項目為例，從零搭建，帶你把這套源碼扒開揉碎講清楚。
項目目標(biāo)
咱們這個項目的核心目標(biāo)很明確：處理一份模擬的用戶調(diào)查原始數(shù)據(jù)，清洗臟數(shù)據(jù)，并輸出結(jié)構(gòu)化的分析結(jié)果。
原始數(shù)據(jù)通常很“臟”，比如：字段缺失：有的用戶沒填年齡。
格式混亂：年齡有的是字符串“25”，有的是數(shù)字25.0，還有的是“二十五”。
邏輯錯誤：年齡填了200，或者-5。我們的目標(biāo)代碼需要做到：自動識別并剔除無效記錄。
將非標(biāo)準(zhǔn)格式統(tǒng)一為標(biāo)準(zhǔn)類型。
輸出清洗后的干凈數(shù)據(jù)以及簡單的統(tǒng)計摘要。這不是為了炫技，而是為了讓你理解：一個合格的“調(diào)查與分析”模塊，到底需要處理哪些細節(jié)。只有把這些細節(jié)搞清楚，下次你復(fù)制別人的代碼時，才知道該改哪里。
目錄結(jié)構(gòu)
為了保持工程化思維，我們不能把所有代碼塞在一個文件里。即使是小項目，也要有清晰的結(jié)構(gòu)。下面是我們推薦的最小化目錄結(jié)構(gòu)：
investigation_analysis/
├── data/
│ └── raw_survey.csv # 原始臟數(shù)據(jù)
├── src/
│ ├── __init__.py
│ ├── cleaner.py # 數(shù)據(jù)清洗邏輯
│ ├── analyzer.py # 分析邏輯
│ └── main.py # 入口文件
├── tests/
│ ├── __init__.py
│ └── test_cleaner.py # 單元測試
├── requirements.txt # 依賴庫
└── README.md # 項目說明為什么要這樣分？cleaner.py：專門負責(zé)“洗”數(shù)據(jù)。輸入是原始列表，輸出是干凈列表。
analyzer.py：專門負責(zé)“算”數(shù)據(jù)。輸入是干凈列表，輸出是統(tǒng)計結(jié)果。
main.py：負責(zé)串聯(lián)流程，讀取文件，調(diào)用清洗，調(diào)用分析，最后打印結(jié)果。這種分層設(shè)計的好處是，當(dāng)你發(fā)現(xiàn)某個環(huán)節(jié)出問題時，可以單獨調(diào)試該模塊，而不需要盯著整個長腳本發(fā)呆。這就是工程化的第一步：關(guān)注點分離。
核心代碼實現(xiàn)
接下來是重頭戲。我們不看那種幾十行的復(fù)雜庫，而是用最基礎(chǔ)的Python邏輯，把源碼解析做到極致。
1. 數(shù)據(jù)清洗模塊 cleaner.py
這是最容易出Bug的地方。很多人復(fù)制代碼時，只復(fù)制了if age 18，卻忽略了age可能根本不是一個數(shù)字。
import re
import logging# 配置日志，方便調(diào)試時查看中間狀態(tài)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)def clean_age(value):清洗年齡字段:param value: 原始年齡值，可能是str, int, float, None:return: 清洗后的整數(shù)年齡，如果無效則返回Noneif value is None:logger.debug(年齡字段為空，跳過)return None# 處理字符串類型的數(shù)字if isinstance(value, str):# 去除空格value = value.strip()if not value:return None# 嘗試直接轉(zhuǎn)換為整數(shù)try:# 這里用int()而不是float()，因為年齡通常是整數(shù)# 如果失敗，說明包含非數(shù)字字符age_int = int(value)except ValueError:# 如果是25.5這種，int()會報錯# 我們嘗試轉(zhuǎn)float再取整，或者根據(jù)業(yè)務(wù)決定丟棄try:age_float = float(value)age_int = int(age_float)logger.debug(f將浮點數(shù)字符串 '{value}' 轉(zhuǎn)為整數(shù) {age_int})except ValueError:# 如果是中文數(shù)字如二十五，需要更復(fù)雜的映射# 這里為了簡化，直接標(biāo)記為無效logger.warning(f無法識別的年齡格式: '{value}')return None# 校驗?zāi)挲g合理性if not (0 age_int 150):logger.warning(f年齡不合理: {age_int})return Nonereturn age_int# 處理數(shù)值類型if isinstance(value, (int, float)):age_int = int(value)if not (0 age_int 150):logger.warning(f數(shù)值年齡不合理: {age_int})return Nonereturn age_int# 其他類型一律無效logger.warning(f不支持的年齡類型: {type(value)})return Nonedef clean_survey_data(raw_records):清洗整批調(diào)查數(shù)據(jù):param raw_records: 原始記錄列表，每個元素是一個字典:return: 清洗后的記錄列表cleaned_records = []for i, record in enumerate(raw_records):new_record = {}# 1. 清洗ID，確保唯一且非空user_id = record.get('id')if not user_id:logger.warning(f第{i}條記錄缺少ID，丟棄)continuenew_record['id'] = str(user_id)# 2. 清洗年齡age = clean_age(record.get('age'))if age is None:logger.warning(f第{i}條記錄(ID:{user_id})年齡無效，丟棄)continuenew_record['age'] = age# 3. 清洗滿意度評分 (1-5)rating = record.get('rating')if rating is None:# 這里可以選擇默認值或者丟棄，根據(jù)業(yè)務(wù)需求# 假設(shè)調(diào)查必須評分，所以丟棄logger.warning(f第{i}條記錄(ID:{user_id})缺少評分，丟棄)continuetry:rating_int = int(rating)if not (1 = rating_int = 5):logger.warning(f第{i}條記錄評分超出范圍: {rating_int})continueexcept (ValueError, TypeError):logger.warning(f第{i}條記錄評分格式錯誤: {rating})continuenew_record['rating'] = rating_int# 4. 保留其他合法字段if 'comment' in record and isinstance(record['comment'], str):new_record['comment'] = record['comment'].strip()cleaned_records.append(new_record)logger.info(f清洗完成: 原始 {len(raw_records)} 條, 有效 {len(cleaned_records)} 條)return cleaned_records源碼解析關(guān)鍵點：類型判斷前置：在轉(zhuǎn)換之前，先判斷isinstance。這是防止TypeError的關(guān)鍵。
異常捕獲細化：try-except塊里，區(qū)分了int()失敗和float()失敗的情況。很多新手只寫一個except Exception，導(dǎo)致問題被掩蓋。
日志記錄：logger不是裝飾，它是你調(diào)試時的眼睛。當(dāng)數(shù)據(jù)量大了，你不可能逐行打印，日志能幫你快速定位哪條數(shù)據(jù)被丟棄了。2. 分析模塊 analyzer.py
數(shù)據(jù)干凈了，接下來算指標(biāo)。這里我們計算平均年齡和平均評分。
from collections import defaultdictdef analyze_survey_data(cleaned_records):對清洗后的數(shù)據(jù)進行基本分析:param cleaned_records: 清洗后的記錄列表:return: 包含統(tǒng)計結(jié)果的字典if not cleaned_records:return {total_count: 0,avg_age: 0,avg_rating: 0,error: No valid data}total_age = 0total_rating = 0count = len(cleaned_records)for record in cleaned_records:total_age += record['age']total_rating += record['rating']# 防止除以零，雖然前面判斷了count0，但好習(xí)慣要保留avg_age = round(total_age / count, 2) if count 0 else 0avg_rating = round(total_rating / count, 2) if count 0 else 0return {total_count: count,avg_age: avg_age,avg_rating: avg_rating}注意：這里的邏輯很簡單，但重點在于輸入契約。analyze_survey_data只接受cleaned_records。如果上游數(shù)據(jù)沒洗干凈，這里就會報錯。這就是為什么我們要把清洗和分析分開。
運行與測試
代碼寫好了，不能只靠肉眼檢查。我們需要寫一個簡單的單元測試，來驗證我們的“源碼解析”是否準(zhǔn)確。
1. 準(zhǔn)備測試數(shù)據(jù)
在tests/test_cleaner.py中：
import unittest
from src.cleaner import clean_survey_dataclass TestCleaner(unittest.TestCase):def setUp(self):self.raw_data = [{id: 1, age: 25, rating: 4, comment: Good},{id: 2, age: 30.5, rating: 5, comment: Excellent},{id: 3, age: abc, rating: 3, comment: Bad}, # 年齡無效{id: 4, age: 200, rating: 2, comment: Old}, # 年齡不合理{id: 5, age: None, rating: 1, comment: No age}, # 年齡缺失{id: 6, age: 40, rating: high, comment: Rating bad}, # 評分無效{id: 7, age: 45, rating: 5, comment: Great}]def test_clean_survey_data(self):cleaned = clean_survey_data(self.raw_data)# 預(yù)期只有 id 1, 2, 7 是有效的self.assertEqual(len(cleaned), 3)# 驗證第一條self.assertEqual(cleaned[0]['id'], '1')self.assertEqual(cleaned[0]['age'], 25)self.assertEqual(cleaned[0]['rating'], 4)# 驗證第二條，浮點數(shù)轉(zhuǎn)整數(shù)self.assertEqual(cleaned[1]['id'], '2')self.assertEqual(cleaned[1]['age'], 30)# 驗證第三條self.assertEqual(cleaned[2]['id'], '7')self.assertEqual(cleaned[2]['age'], 45)print(所有測試通過!)if __name__ == '__main__':unittest.main()2. 運行主程序
src/main.py負責(zé)串聯(lián)：
import csv
import json
from src.cleaner import clean_survey_data
from src.analyzer import analyze_survey_datadef load_csv(filename):加載CSV文件為字典列表records = []try:with open(filename, 'r', encoding='utf-8') as f:reader = csv.DictReader(f)for row in reader:# 簡單轉(zhuǎn)換：將數(shù)字字符串轉(zhuǎn)為float或int，以便cleaner處理# 實際項目中可能需要更復(fù)雜的類型推斷records.append(row)except FileNotFoundError:print(f錯誤: 找不到文件 {filename})return []return recordsdef main():raw_file = 'data/raw_survey.csv'output_file = 'data/cleaned_result.json'print(f正在加載數(shù)據(jù): {raw_file})raw_records = load_csv(raw_file)if not raw_records:print(沒有加載到數(shù)據(jù)，退出)returnprint(開始清洗數(shù)據(jù)...)cleaned_records = clean_survey_data(raw_records)print(開始分析數(shù)據(jù)...)result = analyze_survey_data(cleaned_records)print(f分析結(jié)果: {json.dumps(result, indent=2, ensure_ascii=False)})# 可選：保存清洗后的數(shù)據(jù)# with open(output_file, 'w', encoding='utf-8') as f:# json.dump(cleaned_records, f, indent=2, ensure_ascii=False)# print(f清洗后數(shù)據(jù)已保存至 {output_file})if __name__ == '__main__':main()調(diào)試技巧：
如果運行報錯，不要只看最后一行。往上翻，看logger輸出。比如看到Warning: 第3條記錄年齡無效，你就知道是測試數(shù)據(jù)里的abc導(dǎo)致的，這時候你就該去檢查clean_age里的ValueError捕獲邏輯是否覆蓋了這種情況。
優(yōu)化擴展
基礎(chǔ)版跑通了，但離生產(chǎn)級還有距離。以下是幾個進階方向：配置化管理：
把年齡范圍（0-150）、評分范圍（1-5）提取到config.py或.yaml文件中。這樣如果業(yè)務(wù)規(guī)則變了，不用改代碼，只改配置。數(shù)據(jù)持久化：
目前數(shù)據(jù)在內(nèi)存中。實際項目中，原始數(shù)據(jù)可能在MySQL或PostgreSQL里。你需要用SQLAlchemy或pandas.read_sql來讀取，而不是csv。并發(fā)處理：
如果數(shù)據(jù)量達到百萬級，單線程清洗會很慢。可以使用multiprocessing或concurrent.futures來并行處理數(shù)據(jù)塊。但要注意，日志記錄可能會交錯，需要使用線程安全的日志處理器。錯誤重試機制：
如果是從API拉取數(shù)據(jù)，可能會失敗。需要加上retry裝飾器（如tenacity庫），自動重試3次，間隔遞增?？梢暬?分析結(jié)果不只是數(shù)字?？梢杂胢atplotlib或seaborn畫出年齡分布直方圖、評分餅圖。這一步能幫你快速發(fā)現(xiàn)數(shù)據(jù)中的異常模式，比如“為什么40歲以上的人評分特別低？”小結(jié)
通過這個小小的“調(diào)查與分析”項目，我們其實講透了一個核心思路：源碼解析的本質(zhì)，是對數(shù)據(jù)流向和控制流的掌控。
很多新手覺得代碼難懂，是因為他們只把代碼當(dāng)成“黑盒”。你丟進去數(shù)據(jù)，它吐出來結(jié)果，中間發(fā)生了什么，不管。但一旦出錯，黑盒就失效了。
我們剛才做的每一步：分目錄，是為了隔離關(guān)注點。
寫日志，是為了追蹤數(shù)據(jù)流。
寫測試，是為了驗證控制流。
處理異常，是為了覆蓋邊界條件。這些不是花架子，而是你從“代碼搬運工”變成“工程師”的分水嶺。下次再遇到復(fù)制來的代碼跑不通，別再盲目改參數(shù)。打開logger，加點斷點，順著數(shù)據(jù)流走一遍，問題往往就浮出水面了。
在CSDN或者StackOverflow上，高手的回答通常不是“試試重啟”，而是“檢查一下這里的空值判斷”。這就是差距所在。
最后，拋出一個問題引發(fā)討論：
在你實際工作中，有沒有遇到過那種“邏輯看起來沒錯，但跑出來結(jié)果就是不對”的代碼？你是怎么排查的？是加日志、看堆棧，還是干脆重寫？
還有什么不懂的？評論區(qū)留言挨個回。特別是關(guān)于類型轉(zhuǎn)換陷阱或者日志配置的具體寫法，歡迎提問。