Python实现ASPX页面访问次数统计及最新访问日期获取

针对处理30GB大规模日志文件的需求，既要避免内存溢出，又要新增「最新访问日期」统计，推荐采用逐文件分块读取+增量聚合的方案，结合pandas实现高效处理：

核心思路 #

1. 分块读取：由于总数据量巨大，不能一次性加载到内存，用pandas分块读取单日志文件，控制每块数据大小（如10000行），降低内存占用。
2. 增量聚合：维护一个全局统计字典，每次处理完一块数据后，将当前块的统计结果（访问次数、最大访问日期）与全局结果合并——已存在的页面累加次数、更新最新日期；新页面直接新增记录。
3. 字段筛选：先过滤状态码为200的记录，再提取.aspx结尾的页面路径，减少无效数据处理。

具体实现代码 #

1. 导入依赖库 #

import pandas as pd
import re
from datetime import datetime
import glob

2. 日期解析函数 #

根据日志中的日期格式，将字符串转换为可比较的datetime对象（请根据实际日志格式调整）：

def parse_log_date(date_str):
    # 处理格式如 "[10/Oct/2000:13:55:36 +0000]" 的日期字符串
    date_str = date_str.strip('[]')
    return datetime.strptime(date_str, '%d/%b/%Y:%H:%M:%S %z')

3. 单日志文件处理函数 #

负责分块读取、筛选数据、局部统计，并合并到全局结果：

def process_log_file(file_path, global_stats):
    # 正则匹配日志字段（请根据你的实际日志格式调整）
    # 示例日志格式：192.168.1.1 - - [10/Oct/2000:13:55:36 +0000] "GET /index.aspx HTTP/1.1" 200 1234
    log_pattern = re.compile(
        r'(?P<ip>\S+) \S+ \S+ \[(?P<date>.+?)\] "(?P<method>\S+) (?P<path>\S+) \S+" (?P<status>\d+) (?P<size>\S+)'
    )
    
    # 分块读取日志，chunksize可根据内存情况调整
    chunk_iter = pd.read_csv(
        file_path,
        sep='\s+',
        header=None,
        chunksize=10000,
        engine='python',
        names=['ip', '_1', '_2', 'date', '_3', 'method', 'path', '_4', 'status', 'size']
    )
    
    for chunk in chunk_iter:
        # 筛选状态码为200的记录
        chunk = chunk[chunk['status'] == '200']
        # 筛选.aspx结尾的页面
        chunk = chunk[chunk['path'].str.endswith('.aspx')]
        if chunk.empty:
            continue
        
        # 解析日期为datetime对象
        chunk['datetime'] = chunk['date'].apply(parse_log_date)
        
        # 按页面路径分组，统计访问次数和最大访问日期
        chunk_stats = chunk.groupby('path').agg(
            count=('path', 'count'),
            latest_date=('datetime', 'max')
        ).reset_index()
        
        # 合并到全局统计字典
        for _, row in chunk_stats.iterrows():
            path = row['path']
            if path in global_stats:
                global_stats[path]['count'] += row['count']
                if row['latest_date'] > global_stats[path]['latest_date']:
                    global_stats[path]['latest_date'] = row['latest_date']
            else:
                global_stats[path] = {
                    'count': row['count'],
                    'latest_date': row['latest_date']
                }

4. 主函数：遍历所有日志文件并输出结果 #

def main(log_dir):
    # 初始化全局统计字典
    global_stats = {}
    
    # 获取指定目录下的所有日志文件（此处假设后缀为.log，可根据实际修改）
    log_files = glob.glob(f'{log_dir}/*.log')
    
    for file in log_files:
        print(f'Processing {file}...')
        process_log_file(file, global_stats)
    
    # 将统计结果转换为DataFrame，方便排序和输出
    result_df = pd.DataFrame.from_dict(global_stats, orient='index').reset_index()
    result_df.columns = ['page_path', 'visit_count', 'latest_visit_date']
    
    # 按访问次数降序排序（可选）
    result_df = result_df.sort_values('visit_count', ascending=False)
    
    # 保存结果到CSV文件
    result_df.to_csv('page_access_stats.csv', index=False)
    print('处理完成，结果已保存至 page_access_stats.csv')

if __name__ == '__main__':
    # 替换为你的日志文件所在目录
    main('/path/to/your/logs')

关键注意事项 #

• 日志格式适配：如果你的日志格式与示例不同，请修改正则表达式和read_csv的列名、分隔符，确保能正确提取status、path、date字段。
• 日期格式调整：若日志日期格式不是%d/%b/%Y:%H:%M:%S %z，请修改parse_log_date中的格式字符串。
• 内存优化：若内存占用过高，可减小chunksize的值；如果需要更高性能，可改用multiprocessing多进程处理（注意用Manager.dict共享全局统计字典）。
• 性能提升：对于超大规模文件，也可考虑使用Dask库替代pandas，它专为大数据集设计，无需手动分块。

内容的提问来源于stack exchange，提问作者user2672216