Python高效实现读写文件和处理CSV/JSON详解

# 读取文件（默认模式，文件不存在会报 FileNotFoundError）
file = open("access.log", "r")

# 写入文件（会覆盖原有内容，文件不存在则创建）
file = open("output.txt", "w")

# 追加内容（不覆盖，在原有内容末尾继续写）
file = open("output.txt", "a")

# 推荐写法：with 语句保证文件一定被关闭
with open("access.log", "r", encoding="utf-8") as f:
    content = f.read()
# 离开 with 代码块后，f 自动关闭，无论是否有异常

# 不推荐写法：需要手动 close()，一旦出错容易忘记
f = open("access.log", "r", encoding="utf-8")
content = f.read()
f.close()  # 如果 read() 出错，这行就执行不到

with open("access.log", "r", encoding="utf-8") as f:

    # 方式1：read() — 一次性读取全部内容，返回字符串
    content = f.read()
    # 适合：小文件（< 几十 MB），需要整体处理

    # 方式2：readline() — 每次读取一行，返回字符串（含 \n）
    line = f.readline()
    # 适合：需要手动控制读取进度的场景

    # 方式3：readlines() — 一次性读取所有行，返回列表
    lines = f.readlines()
    # 适合：小文件，需要按行处理并保留所有行

    # 方式4：for line in f — 逐行迭代（推荐）
    for line in f:
        process(line)
    # 适合：大文件（按需读取，内存友好）

# write()：写入一个字符串，返回写入的字符数
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")        # \n 换行符需要手动添加
    f.write("第二行\n")
    count = f.write("第三行\n")
    print(count)               # 输出写入的字符数

# writelines()：写入字符串列表，不自动添加换行符
lines = ["第一行\n", "第二行\n", "第三行\n"]
with open("output.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)

def append_log(message):
    """追加一条日志记录"""
    import datetime
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open("app.log", "a", encoding="utf-8") as f:
        f.write(f"[{timestamp}] {message}\n")

append_log("用户登录成功")
append_log("查询请求处理完成")
# 每次调用都在 app.log 末尾追加一行

import csv

# 读取 CSV 文件，每行返回一个列表
with open("students.csv", "r", encoding="utf-8") as f:
    reader = csv.reader(f)
    header = next(reader)    # 第一行通常是表头
    print("字段名:", header)  # ['姓名', '分数', '班级']

    for row in reader:
        print(row)            # ['张三', '92', '高一甲班']

import csv

with open("students.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)

    for row in reader:
        # row 是一个字典：{'姓名': '张三', '分数': '92', '班级': '高一甲班'}
        name = row["姓名"]
        score = int(row["分数"])   # CSV 里数字是字符串，需要手动转换
        print(f"{name}: {score} 分")

import csv

students = [
    {"姓名": "张三", "分数": 92, "班级": "高一甲班"},
    {"姓名": "李四", "分数": 87, "班级": "高一乙班"},
    {"姓名": "王五", "分数": 95, "班级": "高一甲班"},
]

with open("output.csv", "w", encoding="utf-8-sig", newline="") as f:
    fieldnames = ["姓名", "分数", "班级"]
    writer = csv.DictWriter(f, fieldnames=fieldnames)

    writer.writeheader()   # 写入表头行
    writer.writerows(students)

# 写入时用 utf-8
with open("output.csv", "w", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["姓名", "分数"])
# 用 Excel 打开：第一列显示 "锘縿名" 或乱码

# 写入时用 utf-8-sig（BOM 标记版本）
with open("output.csv", "w", encoding="utf-8-sig") as f:
    writer = csv.writer(f)
    writer.writerow(["姓名", "分数"])
# 用 Excel 打开：正常显示中文

import json

# 从文件读取 JSON，自动转换为 Python 对象
with open("config.json", "r", encoding="utf-8") as f:
    config = json.load(f)

# config 现在是 Python 字典或列表
print(config["database"]["host"])  # 像操作普通字典一样

import json

report = {
    "date": "2024-01-15",
    "total_requests": 12453,
    "error_count": 23,
    "error_urls": ["/api/user/login", "/api/order/create"],
    "success_rate": 99.82
}

with open("report.json", "w", encoding="utf-8") as f:
    json.dump(
        report,
        f,
        ensure_ascii=False,   # 不把中文转义成 \uXXXX，保持可读性
        indent=2              # 每级缩进 2 个空格，文件更易读
    )

{
  "date": "2024-01-15",
  "total_requests": 12453,
  "error_count": 23,
  "error_urls": [
    "/api/user/login",
    "/api/order/create"
  ],
  "success_rate": 99.82
}

import datetime, json

data = {
 "tags": {"python", "backend"},     # set 类型
 "created": datetime.datetime.now() # datetime 类型
}

# 直接 json.dump 会报 TypeError
# 需要手动转换
data["tags"] = list(data["tags"])
data["created"] = data["created"].isoformat()

print(json.dumps(data))
# {"tags": ["python", "backend"], "created": "2024-01-15T10:30:00"}

from pathlib import Path

# 创建路径对象
data_dir = Path("data")
log_file = data_dir / "access.log"   # / 运算符拼接路径（自动处理斜杠）
print(log_file)                        # data/access.log（或 data\access.log）

# 常用属性
p = Path("/home/user/projects/app.log")
print(p.name)       # app.log（文件名含扩展名）
print(p.stem)       # app（文件名不含扩展名）
print(p.suffix)     # .log（扩展名）
print(p.parent)     # /home/user/projects（父目录）

# 检查路径状态
print(p.exists())           # 是否存在
print(p.is_file())          # 是否是文件
print(p.is_dir())           # 是否是目录

# 创建目录（mkdir 和 parents=True 允许多级创建）
output_dir = Path("output/reports/2024")
output_dir.mkdir(parents=True, exist_ok=True)  # exist_ok=True 已存在不报错

import os
from pathlib import Path

# ———— 路径拼接 ————
# os.path 写法（字符串，跨平台需要注意斜杠）
config_path = os.path.join("config", "settings", "dev.json")

# pathlib 写法（/ 操作符，自动处理跨平台）
config_path = Path("config") / "settings" / "dev.json"

# ———— 读取文件内容 ————
# os.path 写法
with open(os.path.join("data", "users.json"), "r", encoding="utf-8") as f:
    data = json.load(f)

# pathlib 写法（直接在 Path 对象上调用 read_text）
data = json.loads(Path("data", "users.json").read_text(encoding="utf-8"))

# ———— 列出目录下所有 .log 文件 ————
# os.path 写法（需要 os.listdir + 过滤）
logs = [f for f in os.listdir("logs") if f.endswith(".log")]

# pathlib 写法（glob 模式，更直观）
logs = list(Path("logs").glob("*.log"))
# 递归查找所有子目录的 .log 文件
all_logs = list(Path("logs").rglob("*.log"))

from pathlib import Path

def analyze_app_log(log_path):
    """
    统计应用日志中各级别的出现次数。

    参数:
        log_path: 日志文件路径（字符串或 Path 对象）

    返回:
        dict: {'ERROR': n, 'WARNING': n, 'INFO': n, 'UNKNOWN': n}
    """
    log_path = Path(log_path)

    if not log_path.exists():
        print(f"日志文件不存在: {log_path}")
        return {}

    counts = {"ERROR": 0, "WARNING": 0, "INFO": 0, "UNKNOWN": 0}

    with open(log_path, "r", encoding="utf-8") as f:
        for line in f:          # 逐行读取，内存友好
            line = line.strip() # 去掉行首行尾空白和换行符
            if not line:        # 跳过空行
                continue

            # 判断日志级别（日志格式通常包含 [ERROR] 或 ERROR:）
            matched = False
            for level in ["ERROR", "WARNING", "INFO"]:
                if level in line:
                    counts[level] += 1
                    matched = True
                    break

            if not matched:
                counts["UNKNOWN"] += 1

    return counts


# 使用示例
result = analyze_app_log("app.log")
for level, count in result.items():
    print(f"{level:10s}: {count} 条")

import re
import json
from pathlib import Path
from collections import defaultdict


def parse_nginx_log_line(line):
    """
    解析一行 Nginx 标准访问日志。
    返回包含 ip / url / status / size 的字典，解析失败返回 None。
    """
    # 匹配标准 Nginx combined 日志格式
    pattern = r'(\S+) \S+ \S+ \[.*?\] "(\w+) (\S+) \S+" (\d+) (\d+)'
    match = re.match(pattern, line)
    if not match:
        return None

    return {
        "ip": match.group(1),
        "method": match.group(2),
        "url": match.group(3),
        "status": int(match.group(4)),
        "size": int(match.group(5)),
    }


def analyze_nginx_log(log_path, output_path=None):
    """
    分析 Nginx 访问日志，生成统计报告。

    参数:
        log_path:   输入日志文件路径
        output_path: 输出 JSON 报告路径，为 None 则不写文件

    返回:
        dict: 统计报告
    """
    log_path = Path(log_path)

    if not log_path.exists():
        raise FileNotFoundError(f"日志文件不存在: {log_path}")

    # 统计数据
    total_lines = 0
    parsed_lines = 0
    status_counts = defaultdict(int)   # 各状态码出现次数
    error_urls = defaultdict(int)      # 5xx 错误的 URL 分布
    top_ips = defaultdict(int)         # 各 IP 的请求次数
    total_bytes = 0

    with open(log_path, "r", encoding="utf-8") as f:
        for line in f:
            total_lines += 1
            line = line.strip()
            if not line:
                continue

            parsed = parse_nginx_log_line(line)
            if parsed is None:
                continue  # 解析失败，跳过该行

            parsed_lines += 1
            status = parsed["status"]
            status_counts[status] += 1
            top_ips[parsed["ip"]] += 1
            total_bytes += parsed["size"]

            # 收集 5xx 错误 URL
            if 500 <= status < 600:
                error_urls[parsed["url"]] += 1

    # 构建报告
    error_rate = 0.0
    if parsed_lines > 0:
        error_5xx = sum(v for k, v in status_counts.items() if 500 <= k < 600)
        error_rate = round(error_5xx / parsed_lines * 100, 2)

    # 按请求数排序取 Top 10
    top_10_ips = sorted(top_ips.items(), key=lambda x: x[1], reverse=True)[:10]
    top_10_errors = sorted(error_urls.items(), key=lambda x: x[1], reverse=True)[:10]

    report = {
        "summary": {
            "total_lines": total_lines,
            "parsed_lines": parsed_lines,
            "total_bytes": total_bytes,
            "error_rate_5xx": f"{error_rate}%",
        },
        "status_distribution": dict(sorted(status_counts.items())),
        "top_10_client_ips": [{"ip": ip, "requests": n} for ip, n in top_10_ips],
        "top_10_error_urls": [{"url": url, "errors": n} for url, n in top_10_errors],
    }

    # 写入 JSON 报告
    if output_path:
        output_path = Path(output_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        print(f"报告已写入: {output_path}")

    return report


# 运行分析
if __name__ == "__main__":
    report = analyze_nginx_log(
        log_path="access.log",
        output_path="reports/nginx_report.json"
    )

    print(f"总请求数: {report['summary']['parsed_lines']}")
    print(f"5xx 错误率: {report['summary']['error_rate_5xx']}")
    print("\nTop 5 错误 URL:")
    for item in report["top_10_error_urls"][:5]:
        print(f"  {item['errors']:5d} 次  {item['url']}")