2025年P(guān)ythonPython核心庫專項訓(xùn)練試卷：模塊應(yīng)用實戰(zhàn)版考試時間：______分鐘總分：______分姓名：______第一題請編寫Python代碼，定義一個函數(shù)`process_files_in_directory`，該函數(shù)接收一個目錄路徑`dir_path`作為參數(shù)。函數(shù)需要遍歷指定目錄及其所有子目錄下所有的`.txt`文件。對于每個`.txt`文件，讀取其內(nèi)容，并將內(nèi)容中的所有大寫字母轉(zhuǎn)換為小寫字母，然后覆蓋原文件。如果指定目錄下沒有`.txt`文件，或者讀取/寫入文件過程中發(fā)生任何異常，函數(shù)應(yīng)捕獲這些異常并打印錯誤信息（包含異常類型和描述），但不應(yīng)中斷整個目錄的處理過程。第二題使用`collections`模塊和`itertools`模塊，編寫Python代碼完成以下任務(wù)：從一個列表`data`（例如`data=['apple','banana','apple','orange','banana','banana']`）中，統(tǒng)計每個元素的出現(xiàn)次數(shù)，并按出現(xiàn)次數(shù)從多到少排序輸出。對于出現(xiàn)次數(shù)相同的元素，按它們在原列表`data`中首次出現(xiàn)的順序排列。請使用盡可能簡潔和高效的方式實現(xiàn)。第三題假設(shè)你需要處理一個包含日期時間信息的CSV文件`events.csv`，其內(nèi)容格式如下（假設(shè)字段有`event_id`,`event_name`,`timestamp`）：```event_id,event_name,timestamp1,Login,"2025-05-2014:30:00"2,Logout,"2025-05-2014:35:05"3,Login,"2025-05-2014:40:00"4,Logout,"2025-05-2014:45:00"...```請編寫Python代碼，使用`pandas`庫讀取該CSV文件。然后，計算每個用戶的平均登錄時長（即相鄰的登錄和登出事件之間的時間差，單位為分鐘）。結(jié)果應(yīng)包含用戶ID和對應(yīng)的平均登錄時長（分鐘），按平均時長從長到短排序輸出。你需要處理可能的數(shù)據(jù)缺失或時間格式錯誤的情況。第四題請編寫Python代碼，實現(xiàn)一個簡單的HTTP客戶端功能。使用`requests`庫向網(wǎng)址`/posts`發(fā)送一個GET請求，獲取JSON格式的響應(yīng)數(shù)據(jù)。然后，篩選出所有`userId`為5的帖子，提取這些帖子的`title`和`body`字段，并將它們組成一個新的字典列表，其中每個字典包含一個帖子的`title`和`body`。最后，將這個列表轉(zhuǎn)換為JSON字符串，并打印輸出。第五題編寫Python代碼，使用`multiprocessing`模塊實現(xiàn)并行計算。定義一個函數(shù)`compute_square`接收一個數(shù)字列表`numbers`，返回一個包含每個數(shù)字平方的新列表。然后，創(chuàng)建一個`Pool`對象，使用4個進(jìn)程并行計算列表`[1,2,3,...,20]`中每個數(shù)字的平方。收集所有進(jìn)程的返回結(jié)果，并將這些平方值存儲在一個新的列表`squared_numbers`中。最后，打印輸出`squared_numbers`列表。請確保使用`Pool`的`map`或`apply_async`方法，并妥善處理進(jìn)程間通信。試卷答案第一題答案```pythonimportosimportglobdefprocess_files_in_directory(dir_path):try:fortxt_fileinglob.glob(os.path.join(dir_path,'','*.txt'),recursive=True):try:withopen(txt_file,'r',encoding='utf-8')asf:content=f.read()content_lower=content.lower()withopen(txt_file,'w',encoding='utf-8')asf:f.write(content_lower)exceptExceptionase:print(f"Errorprocessingfile{txt_file}:{e}")exceptExceptionase:print(f"Erroraccessingdirectory{dir_path}:{e}")```第一題解析思路1.導(dǎo)入模塊：導(dǎo)入`os`用于路徑操作和`glob`用于文件模式匹配。2.定義函數(shù)：定義`process_files_in_directory`函數(shù)，接收目錄路徑`dir_path`。3.異常處理（外層）：使用`try...except`捕獲訪問目錄本身可能引發(fā)的異常（如路徑不存在），并打印錯誤信息。4.文件查找：使用`glob.glob`查找指定目錄及其子目錄下所有`.txt`文件。`os.path.join`用于構(gòu)建跨平臺的路徑，``表示任意深度的子目錄，`*.txt`表示文件名以`.txt`結(jié)尾。5.迭代文件：對找到的每個`.txt`文件路徑進(jìn)行迭代。6.異常處理（內(nèi)層）：對每個文件的讀取和寫入操作使用`try...except`捕獲可能發(fā)生的異常（如文件不存在、讀取權(quán)限問題、寫入權(quán)限問題等），并打印包含文件路徑和異常信息的錯誤。7.讀取文件內(nèi)容：使用`withopen(...)`以讀模式打開文件，`encoding='utf-8'`指定編碼，`f.read()`讀取全部內(nèi)容。8.轉(zhuǎn)換為小寫：使用字符串的`lower()`方法將內(nèi)容轉(zhuǎn)換為全小寫。9.寫入文件內(nèi)容：使用`withopen(...)`以寫模式（覆蓋）打開同一文件，`f.write(content_lower)`將轉(zhuǎn)換后的小寫內(nèi)容寫入文件。10.流程控制：外層異常處理確保一個文件的失敗不影響其他文件的處理；內(nèi)層異常處理確保文件操作失敗時能報錯并繼續(xù)處理下一個文件。第二題答案```pythonfromcollectionsimportCounterfromitertoolsimportgroupbydefcount_elements_sorted(data):#統(tǒng)計元素出現(xiàn)次數(shù)counts=Counter(data)#按出現(xiàn)次數(shù)降序、原列表首次出現(xiàn)順序排序#items()將字典轉(zhuǎn)為(元素,次數(shù))列表#key=lambdax:(-x[1],data.index(x[0]))設(shè)置排序鍵#x[1]是次數(shù)，-x[1]實現(xiàn)降序；data.index(x[0])獲取元素首次出現(xiàn)位置，用于穩(wěn)定排序sorted_counts=sorted(counts.items(),key=lambdax:(-x[1],data.index(x[0])))#輸出結(jié)果forelement,countinsorted_counts:print(f"{element}:{count}")#示例調(diào)用#data=['apple','banana','apple','orange','banana','banana']#count_elements_sorted(data)```第二題解析思路1.導(dǎo)入模塊：導(dǎo)入`collections.Counter`用于高效計數(shù)，導(dǎo)入`itertools.groupby`（雖然此題核心用不到`groupby`，但屬于該模塊，題目可能暗示其潛力）。2.定義函數(shù)：定義`count_elements_sorted`函數(shù)，接收列表`data`。3.計數(shù)：使用`Counter(data)`對列表`data`中的元素進(jìn)行計數(shù)，得到一個字典，鍵為元素，值為出現(xiàn)次數(shù)。4.排序：對`Counter`對象的`.items()`進(jìn)行排序。*排序鍵：使用`lambdax:(-x[1],data.index(x[0]))`作為`sorted`的`key`參數(shù)。*`-x[1]`：`x[1]`是元素的計數(shù)（次數(shù)），取負(fù)值實現(xiàn)降序排列（次數(shù)多的排前面）。*`data.index(x[0])`：`x[0]`是元素本身，`data.index(x[0])`獲取該元素在原列表`data`中首次出現(xiàn)的位置。由于`index`方法會返回第一個匹配的位置，這確保了在次數(shù)相同的情況下，按照元素在原列表中出現(xiàn)的先后順序排列。*排序方式：`sorted()`默認(rèn)為升序，通過`key`參數(shù)指定的復(fù)雜值，實現(xiàn)了先按次數(shù)降序，次數(shù)相同再按首次出現(xiàn)位置升序的穩(wěn)定排序。5.輸出：使用`for`循環(huán)遍歷排序后的元素和計數(shù)對，格式化輸出。第三題答案```pythonimportpandasaspdfromdatetimeimportdatetimedefcalculate_average_login_duration(file_path):try:#讀取CSV文件df=pd.read_csv(file_path)#確保timestamp列存在且為字符串類型，準(zhǔn)備轉(zhuǎn)換if'timestamp'notindf.columns:print("Error:'timestamp'columnmissing.")returndf['timestamp']=df['timestamp'].astype(str)#嘗試解析時間戳，處理可能的錯誤defparse_timestamp(ts):forfmtin("%Y-%m-%d%H:%M:%S","%Y-%m-%dT%H:%M:%S"):try:returndatetime.strptime(ts,fmt)exceptValueError:continueprint(f"Warning:Cannotparsetimestamp:{ts}")returnpd.NaT#返回NaT(NotaTime)df['timestamp_parsed']=df['timestamp'].apply(parse_timestamp)#篩選登錄和登出事件event_types=df['event_name'].unique()if'Login'notinevent_typesor'Logout'notinevent_types:print("Error:Requiredeventnames'Login'or'Logout'notfound.")return#獲取登錄事件數(shù)據(jù)login_df=df[df['event_name']=='Login'].copy()login_df['event_id']=login_df['event_id'].astype(int)#確保ID為整數(shù)#獲取登出事件數(shù)據(jù)logout_df=df[df['event_name']=='Logout'].copy()logout_df['event_id']=logout_df['event_id'].astype(int)#確保ID為整數(shù)#按用戶ID和時間排序login_df=login_df.sort_values(by=['event_id','timestamp_parsed'])logout_df=logout_df.sort_values(by=['event_id','timestamp_parsed'])#計算平均時長result=[]foruser_id,groupinlogin_df.groupby('event_id'):corresponding_logout=logout_df[logout_df['event_id']==user_id]ifnotcorresponding_logout.empty:duration=(corresponding_logout.iloc[0]['timestamp_parsed']-group.iloc[0]['timestamp_parsed']).total_seconds()/60.0#分鐘result.append({'user_id':user_id,'avg_duration':duration})#轉(zhuǎn)換為DataFrame并排序result_df=pd.DataFrame(result)result_df=result_df.sort_values(by='avg_duration',ascending=False)#輸出結(jié)果print(result_df[['user_id','avg_duration']])exceptExceptionase:print(f"Anerroroccurred:{e}")#示例調(diào)用#calculate_average_login_duration('events.csv')```第三題解析思路1.導(dǎo)入模塊：導(dǎo)入`pandas`用于數(shù)據(jù)處理，導(dǎo)入`datetime`用于時間運算。2.定義函數(shù)：定義`calculate_average_login_duration`函數(shù)，接收文件路徑`file_path`。3.讀取CSV：使用`pd.read_csv(file_path)`讀取CSV文件到DataFrame`df`。4.準(zhǔn)備時間列：*檢查`timestamp`列是否存在。*將`timestamp`列轉(zhuǎn)換為字符串類型，以便后續(xù)處理。*定義`parse_timestamp`函數(shù)嘗試解析時間字符串。使用`try...except`循環(huán)嘗試不同的日期時間格式（`%Y-%m-%d%H:%M:%S`和`%Y-%m-%dT%H:%M:%S`，根據(jù)實際文件調(diào)整）。如果都無法解析，打印警告并返回`pd.NaT`。*使用`df['timestamp'].apply(parse_timestamp)`將原始時間字符串列轉(zhuǎn)換為`datetime`對象的列`timestamp_parsed`。5.數(shù)據(jù)篩選與準(zhǔn)備：*獲取所有唯一的事件類型。*檢查是否包含必需的`'Login'`和`'Logout'`事件，如果缺少則報錯。*分別篩選出`'Login'`和`'Logout'`事件的數(shù)據(jù)，復(fù)制到新的DataFrame(`login_df`,`logout_df`)。*確保`event_id`列在兩個DataFrame中都是整數(shù)類型，以便后續(xù)匹配。*對`login_df`和`logout_df`按`event_id`和`timestamp_parsed`進(jìn)行排序，確保同一用戶的登錄和登出按時間順序排列。6.計算時長與聚合：*使用`login_df.groupby('event_id')`按`event_id`分組。*在`for`循環(huán)中，獲取每個用戶組的第一個登錄時間(`group.iloc[0]['timestamp_parsed']`)和該用戶對應(yīng)的第一個登出時間(`corresponding_logout.iloc[0]['timestamp_parsed']`)。*計算時間差：`logout_time-login_time`。使用`.total_seconds()`獲取總秒數(shù)，除以60轉(zhuǎn)換為分鐘。*將每個用戶的`user_id`和計算出的平均時長存儲在`result`列表中（每個用戶只計算一次平均時長，基于其第一個登錄和第一個登出）。7.結(jié)果處理與輸出：*將`result`列表轉(zhuǎn)換為DataFrame`result_df`。*按`avg_duration`列降序排序。*使用`print(result_df[['user_id','avg_duration']])`輸出用戶ID和對應(yīng)的平均登錄時長。第四題答案```pythonimportrequestsimportjsondeffetch_and_process_posts():url="/posts"try:#發(fā)送GET請求response=requests.get(url)#檢查響應(yīng)狀態(tài)碼response.raise_for_status()#如果不是200，會拋出HTTPError異常#獲取JSON數(shù)據(jù)posts_data=response.json()#篩選userId為5的帖子user_5_posts=[postforpostinposts_dataifpost.get('userId')==5]#提取title和bodyresult=[{'title':post['title'],'body':post['body']}forpostinuser_5_posts]#轉(zhuǎn)換為JSON字符串result_json=json.dumps(result,indent=4)#indent=4使輸出更易讀#打印輸出print(result_json)exceptrequests.exceptions.RequestExceptionase:print(f"Requesterror:{e}")exceptjson.JSONDecodeErrorase:print(f"JSONdecodeerror:{e}")exceptExceptionase:print(f"Anunexpectederroroccurred:{e}")#示例調(diào)用#fetch_and_process_posts()```第四題解析思路1.導(dǎo)入模塊：導(dǎo)入`requests`用于發(fā)起HTTP請求，導(dǎo)入`json`用于處理JSON數(shù)據(jù)。2.定義函數(shù)：定義`fetch_and_process_posts`函數(shù)。3.設(shè)置URL：定義目標(biāo)URL`url`。4.異常處理：使用`try...except`捕獲整個請求和處理過程中可能發(fā)生的異常。5.發(fā)送GET請求：使用`requests.get(url)`向指定URL發(fā)送GET請求，將響應(yīng)對象存儲在`response`變量中。6.檢查狀態(tài)碼：調(diào)用`response.raise_for_status()`。如果響應(yīng)狀態(tài)碼不是200（表示成功），此方法會拋出`HTTPError`異常，被外層`except`捕獲。7.解析JSON：使用`response.json()`方法將響應(yīng)體中的JSON字符串解析為Python對象（通常是列表或字典）。如果響應(yīng)體不是有效的JSON，會拋出`json.JSONDecodeError`。8.篩選數(shù)據(jù)：使用列表推導(dǎo)式`[postforpostinposts_dataifpost.get('userId')==5]`遍歷解析后的`posts_data`（假設(shè)是列表），篩選出`userId`等于5的帖子字典。9.提取字段：使用另一個列表推導(dǎo)式`[{'title':post['title'],'body':post['body']}forpostinuser_5_posts]`遍歷篩選出的帖子列表，為每個帖子創(chuàng)建一個只包含`title`和`body`字段的新字典。10.轉(zhuǎn)換為JSON字符串：使用`json.dumps(result,indent=4)`將提取后的列表`result`轉(zhuǎn)換為格式化的JSON字符串。`indent=4`參數(shù)使輸出的JSON更易讀。11.打印輸出：使用`print(result_json)`打印最終的JSON字符串。12.異常處理（細(xì)化）：在`except`塊中，分別捕獲`requests.exceptions.RequestException`（處理請求相關(guān)的錯誤）、`json.JSONDecodeError`（處理JSON解析錯誤），以及一個通用的`Exception`來捕獲其他意外錯誤。第五題答案```pythonimportmultiprocessingdefcompute_square(number):returnnumber*numberdefparallel_square(numbers,num_processes):ifnum_processes<=0:raiseValueError("num_processesmustbegreaterthan0")ifnotnumbers:return[]#創(chuàng)建Pool對象withmultiprocessing.Pool(num_processes)aspool:#使用map方法分配任務(wù)squared_numbers=pool.map(compute_square,numbers)##或者使用apply_async，需要手動收集結(jié)果#results=[]#fornumberinnumbers:#result=pool.apply_async(compute_square,args=(number,))#results.append(result)#squared_numbers=[res.get()forresinresults]returnsquared_numbers#示例調(diào)用#if__name__=='__main__':#num_processes=4#numbers=list(range(1,21))#result=parallel_square(numbers,num_processes)#print(result)```第五題解析思路1.導(dǎo)入模塊：導(dǎo)入`multiprocessing`模塊用于實現(xiàn)并發(fā)計算。2.定義計算函數(shù)：定義`compute_square