【Python学习】Python csv 数据处理实例——bilibili 视频播放量跟踪

1. 理论知识准备

在 Python 中，处理 CSV（Comma-Separated Values）文件非常常见，因为 CSV 是一种广泛使用的数据交换格式。Python 标准库中的csv模块提供了读取和写入 CSV 文件的功能。以下是一些基本的使用方法。

读取 CSV 文件

首先，确保你的 CSV 文件是有效的，并且你知道其中数据的结构（例如，是否有标题行）。

import csv

# 打开CSV文件
with open('example.csv', newline='', encoding='utf-8') as csvfile:
    # 创建CSV阅读器
    csvreader = csv.reader(csvfile, delimiter=',')
    
    # 读取并打印每一行
    for row in csvreader:
        print(', '.join(row))

如果你的 CSV 文件包含标题行，你可能想使用csv.DictReader来将每一行作为字典读取，这样可以通过标题访问每个字段。

import csv

# 使用DictReader读取CSV文件
with open('example.csv', newline='', encoding='utf-8') as csvfile:
    csvreader = csv.DictReader(csvfile)
    
    # 读取并打印每一行作为字典
    for row in csvreader:
        print(row)  # 或者 print(row['列名']) 来访问特定列

写入 CSV 文件

写入 CSV 文件也很直接。你可以使用csv.writer或csv.DictWriter（如果你希望以字典的形式写入行）。

import csv

# 要写入的数据
data = [
    ['姓名', '年龄', '城市'],
    ['Alice', '24', 'New York'],
    ['Bob', '27', 'Los Angeles'],
    ['Charlie', '22', 'Boston']
]

# 打开文件并写入
with open('output.csv', mode='w', newline='', encoding='utf-8') as file:
    writer = csv.writer(file)
    
    # 写入数据
    for row in data:
        writer.writerow(row)

使用csv.DictWriter写入字典数据到 CSV 文件：

import csv

# 要写入的数据（字典形式）
data = [
    {'姓名': 'Alice', '年龄': '24', '城市': 'New York'},
    {'姓名': 'Bob', '年龄': '27', '城市': 'Los Angeles'},
    {'姓名': 'Charlie', '年龄': '22', '城市': 'Boston'}
]

# 获取字典中的键名作为标题
fieldnames = data.keys()

# 打开文件并写入
with open('output_dict.csv', mode='w', newline='', encoding='utf-8') as file:
    writer = csv.DictWriter(file, fieldnames=fieldnames)
    
    # 写入标题
    writer.writeheader()
    
    # 写入数据
    for row in data:
        writer.writerow(row)

注意事项

1. ‌文件编码‌：处理 CSV 文件时，确保指定正确的文件编码（如utf-8），特别是当文件中包含非 ASCII 字符时。
2. ‌数据清洗‌：在读取或写入 CSV 文件之前，可能需要对数据进行清洗，例如去除多余的空格、处理缺失值等。
3. ‌异常处理‌：使用try-except块来捕获并处理可能出现的异常，如文件不存在或读写权限问题。

通过掌握这些基本技能，你可以有效地在 Python 中处理 CSV 数据。

2. 实践案例

需求

跟踪 bilibili 视频播放量。

1. 要获取视频的播放量等数据
2. 要持久化跟踪，保存成文件是最好的方式。
3. 便于后续数据分析处理，轻量化的csv是个不错的选择。

实践开始

创建一个函数来获取视频数据

# 获取数据
def get_data(bvid):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
        'Accept-Language': 'zh-CN,zh;q=0.9',
        'Accept-Encoding': 'gzip, deflate, br',
        'Origin': 'https://www.bilibili.com',
        'Connection': 'keep-alive'
    }
    resp = requests.get(f'https://api.bilibili.com/x/web-interface/view?bvid={bvid}', headers=headers).json()
    # print(resp['data']['stat'])
    return resp['data']

准备一个含有视频bvids的文本，bilibili_all.txt

# 读取文件，导入所有bvids
def read_file():
    with open(r'bilibili_all.txt', 'r') as f:
        content = f.read()

    bvids = content.split(',')
    print(bvids)
    data = []
    for bvid in bvids:
        data.append(get_data(bvid))
        time.sleep(1)

    return data

基本的数据准备完成了，还需要一个时间标记

# 时间戳转时间
def timestampToTime(timestamp):
    timeArray = time.localtime(timestamp)
    timeStr = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
    return timeStr

生成一份数据模板，后续可持续添加数据

# 将当前时间点的播放量写入到文件
def writecsv(data, filename):
    try:

        with open(filename, mode='w', newline='', encoding='utf-8') as file:
            
            writer = csv.writer(file)
            viewdata = timestampToTime(int(time.time()))
            head = ['bvid', 'title', viewdata]
            writer.writerow(head)
            
            for vdata in data:
                writer.writerow([vdata['bvid'], vdata['title'], vdata['stat']['view']])

    except Exception as e:
        print(f"Error writing log file: {e}")


    # print((timestampToTime(int(time.time())) + "\n"))
    #
    # print("bvid\ttitle\tview")
    # for vdata in data:
    #     print("{}\t{}\t{}\n".format(vdata["bvid"], vdata["title"], vdata['stat']["view"]))

模板：

获取新数据，并追加写入

# 新数据处理

def handle_data(data):
    new_column_data = [timestampToTime(int(time.time()))] + [item['stat']['view'] for item in data]
    return new_column_data

# 写入追加新列数据

def writecsv_append(data, filename):
    try:
        # 读取文件内容
        with open(filename, mode='r', newline='', encoding='utf-8') as infile:
            reader = csv.reader(infile)
            rows = list(reader)

        # 处理数据
        new_column_data = handle_data(data)

        # 检查数据长度是否匹配
        if len(rows) != len(new_column_data):
            raise ValueError("数据长度与文件行数不匹配")

        # 写入新内容
        with open(filename, mode='w', newline='', encoding='utf-8') as outfile:
            writer = csv.writer(outfile)
            for row, newvalue in zip(rows, new_column_data):
                row.append(newvalue)
                writer.writerow(row)

        print("写入结束")

    except FileNotFoundError:
        print(f"文件未找到: {filename}")
    except ValueError as ve:
        print(f"值错误: {ve}")
    except Exception as e:
        print(f"写入日志文件时发生错误: {e}")

测试成果

if __name__ == '__main__':
    data = read_file()
    # writecsv(data, "log_all.csv")
    writecsv_append(data, "log_all.csv")

后续展望

基本实现了跟踪bilibili 视频播放量的需求，但还有一些需要改进的地方。比如视频bvids文本发生变化时可能会出现错误。时间格式修改成日期，便于按日追踪。