从零到接单 06:文件读写与异常处理——程序开始和外部世界交互

pythontutorialbeginner自动化爬虫接单文件异常处理

系列目录:本文是「从零到接单:Python 自动化与爬虫实战」系列的第 6 篇。学了这么多语法,程序一直在"自言自语"——输入靠 input(),输出靠 print()。真正的项目需要读写文件、处理错误。这篇是数据持久化和健壮编程的关键一课。


前面写的程序,数据都存在变量里,关了程序就没了。真正的项目:

  • 爬虫抓下来的数据要存到文件里
  • 自动化脚本要读取 Excel/CSV/文本文件
  • 程序可能会出错,不能一崩了之

一、文件读写基础

打开文件的模式

| 模式 | 含义 | 文件不存在时 | |------|------|-------------| | "r" | 只读(默认) | ❌ 报错 | | "w" | 只写(覆盖) | ✅ 创建新文件 | | "a" | 追加(不覆盖) | ✅ 创建新文件 | | "r+" | 读写 | ❌ 报错 | | "b" | 二进制模式 | 配合以上模式使用 |

读文件

# 一次性读取全部内容
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(content)

# 逐行读取(推荐,大文件也不怕)
with open("data.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())  # strip() 去掉末尾的换行符

# 读取所有行到列表
with open("data.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()
    print(f"共 {len(lines)} 行")

⚠️ 一定要指定 encoding="utf-8",否则含中文的文件在 Windows 上可能乱码。

写文件

# 覆盖写入
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")
    f.write("你好,世界!\n")

# 追加写入
with open("log.txt", "a", encoding="utf-8") as f:
    f.write(f"[2026-07-11] 程序启动\n")

# 写入多行
lines = ["苹果\n", "香蕉\n", "橘子\n"]
with open("fruits.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)

二、with 语句:自动关闭文件

# ❌ 旧式写法(容易忘记关文件)
f = open("data.txt", "r", encoding="utf-8")
content = f.read()
f.close()  # 别忘了!

# ✅ 推荐写法:with 语句
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()
# 离开 with 块,文件自动关闭。就算中间出错了也会关。

三、路径处理(os.path / pathlib)

import os

# 获取当前脚本所在目录
current_dir = os.path.dirname(__file__)  # 或 os.getcwd()

# 拼接路径(跨平台)
data_path = os.path.join("data", "2026", "report.csv")
print(data_path)  # data/2026/report.csv(Mac/Linux)
                  # data\2026\report.csv(Windows)

# 检查文件/目录是否存在
print(os.path.exists("data.txt"))      # True/False
print(os.path.isfile("data.txt"))      # 是文件吗?
print(os.path.isdir("data"))           # 是目录吗?

# 创建目录
os.makedirs("output/2026/07", exist_ok=True)  # exist_ok=True:目录存在也不报错

# 列出目录内容
files = os.listdir(".")
for f in files:
    print(f)

# pathlib(Python 3.4+,更现代)
from pathlib import Path

data_file = Path("data") / "report.csv"       # 用 / 拼接路径
print(data_file.exists())                      # 是否存在
print(data_file.suffix)                        # .csv
print(data_file.stem)                          # report
Path("output/2026").mkdir(parents=True, exist_ok=True)  # 创建目录

四、CSV 文件读写

CSV(逗号分隔值)是数据存储最常用的格式之一。

import csv

# 写入 CSV
data = [
    {"姓名": "张三", "年龄": "25", "城市": "北京"},
    {"姓名": "李四", "年龄": "30", "城市": "上海"},
    {"姓名": "王五", "年龄": "28", "城市": "广州"},
]

with open("users.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["姓名", "年龄", "城市"])
    writer.writeheader()  # 写表头
    writer.writerows(data)

# 读取 CSV
with open("users.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['姓名']},{row['年龄']}岁,{row['城市']}")

五、JSON 文件读写

JSON 是互联网数据交换的标准格式,爬虫必打交道。

import json

# Python 对象 → JSON 字符串
data = {
    "name": "小明",
    "age": 25,
    "skills": ["Python", "爬虫", "自动化"],
    "contact": {
        "email": "xiaoming@example.com",
        "phone": None
    }
}

# 写入 JSON 文件
with open("user.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)
    # ensure_ascii=False: 保留中文
    # indent=2: 格式化缩进

# 从 JSON 文件读取
with open("user.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)
    print(loaded["name"])  # 小明
    print(loaded["skills"])  # ['Python', '爬虫', '自动化']

# JSON 字符串 ↔ Python 对象
json_str = '{"name": "小明", "age": 25}'
parsed = json.loads(json_str)      # 字符串 → 对象
back_to_str = json.dumps(parsed)   # 对象 → 字符串

六、异常处理:程序不出错的方法

为什么要处理异常?

# ❌ 不处理:程序直接崩溃
num1 = int(input("输入第一个数:"))
num2 = int(input("输入第二个数:"))
result = num1 / num2
print(f"结果:{result}")

# 如果用户输入了字母,或者 num2 是 0,程序直接报错退出!

try-except 基础

try:
    num1 = int(input("输入第一个数:"))
    num2 = int(input("输入第二个数:"))
    result = num1 / num2
    print(f"结果:{result}")
except ValueError:
    print("错误:请输入有效的数字!")
except ZeroDivisionError:
    print("错误:除数不能为 0!")
except Exception as e:
    print(f"发生了未知错误:{e}")

完整的异常处理结构

try:
    file = open("data.txt", "r", encoding="utf-8")
    content = file.read()
    numbers = [int(x) for x in content.split()]
    avg = sum(numbers) / len(numbers)
except FileNotFoundError:
    print("文件不存在,请检查路径")
except ValueError:
    print("文件内容不是纯数字")
except ZeroDivisionError:
    print("文件为空")
else:
    # 没有异常时执行
    print(f"平均值:{avg}")
finally:
    # 无论是否发生异常都执行
    file.close()
    print("文件已关闭")

常见异常类型

| 异常 | 原因 | |------|------| | ValueError | 类型对但值不对(如 int("abc")) | | TypeError | 类型不对的运算(如 "a" + 1) | | ZeroDivisionError | 除以 0 | | FileNotFoundError | 文件不存在 | | KeyError | 字典中不存在的 key | | IndexError | 列表索引越界 | | AttributeError | 对象没有这个属性/方法 | | ImportError | 导入模块失败 | | ConnectionError | 网络连接失败 |

抛出自定义异常

def get_age():
    age = int(input("输入年龄:"))
    if age < 0 or age > 150:
        raise ValueError(f"年龄 {age} 不合理!")
    return age

try:
    age = get_age()
    print(f"你的年龄是:{age}")
except ValueError as e:
    print(f"输入错误:{e}")

七、实战:简易日志工具

# logger.py —— 一个简易的日志记录器

import os
from datetime import datetime

class SimpleLogger:
    """简易日志工具"""
    
    def __init__(self, log_dir="logs"):
        self.log_dir = log_dir
        os.makedirs(log_dir, exist_ok=True)
        self.log_file = os.path.join(log_dir, f"log_{datetime.now().strftime('%Y%m%d')}.txt")
    
    def _write(self, level, message):
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        line = f"[{timestamp}] [{level}] {message}\n"
        try:
            with open(self.log_file, "a", encoding="utf-8") as f:
                f.write(line)
        except Exception as e:
            print(f"写入日志失败:{e}")
        print(line, end="")
    
    def info(self, message):
        self._write("INFO", message)
    
    def warning(self, message):
        self._write("WARNING", message)
    
    def error(self, message):
        self._write("ERROR", message)

# 使用示例
if __name__ == "__main__":
    logger = SimpleLogger()
    logger.info("爬虫启动")
    logger.info("正在抓取第 1 页...")
    logger.warning("响应速度较慢")
    logger.error("第 3 页抓取失败,已跳过")
    logger.info("爬虫结束")

总结

| 操作 | 代码 | 要点 | |------|------|------| | 读文件 | with open(f, "r") as f: | 用 with,指定 encoding | | 写文件 | with open(f, "w") as f: | w 覆盖,a 追加 | | CSV | csv.DictReader/DictWriter | 用字典方式操作 | | JSON | json.load/dump | ensure_ascii=False 保留中文 | | try-except | try...except Error... | 具体异常在前,兜底在后 | | finally | finally: | 清理代码一定执行 |

下一篇,我们将进入正则表达式——文本处理的终极武器,也是爬虫从网页中提取信息的基础。


练习:写一个程序,读取一个 CSV 文件(假设有三列:日期、类别、金额),计算每个类别的总金额,并将结果写入新的 JSON 文件。文件不存在时给出友好的错误提示。

Comments

Sign in to leave a comment.