目录
七、历史文章指路
Python 提供了必要的函数和方法进行默认情况下的文件基本操作,可以使用 file 对象做大部分的文件操作。
使用 Python 内置的 open() 函数打开一个文件,创建一个 file 对象:
file f = open([file_name], [access_mode], [encoding], [buffering])
file_name:访问文件名称,包含文件路径。
access_mode:打开文件的模式 - 只读、写入、追加等。常用取值如下,默认文件访问模式为只读 (r)。
模式 | 描述 |
---|---|
r | 以只读方式打开一个文件,文件的指针会放在文件的开头。 |
w | 打开一个文件只用于写入,文件的指针会放在文件的开头并从头开始编辑,原有的内容会被删除。 |
a | 打开一个文件用于追加写,如果文件已存在,文件的指针会放在文件的结尾;如果文件不存在则新建。 |
b | 以二进制模式打开一个文件。 |
+ | 打开一个文件进行更新 (可读可写)。 |
r+ | 打开一个文件用于读写,文件的指针会放在文件的开头。 |
w+ | 打开一个文件用于读写,如果文件已存在则打开文件,文件的指针会放在文件的开头并从头开始编辑,原有的内容会被删除。如果文件不存在则新建。 |
a+ | 打开一个文件用于追加读写,如果文件已存在则打开文件,文件的指针会放在文件的结尾;如果文件不存在则新建。 |
encoding:文件编码,默认 GBK。
buffering:缓冲区。开启后使用行缓冲,也就是说遇到换行符就缓冲,或者一行的缓冲内存被占满时,就会写入到磁盘。如果 buffering=0,表示关闭缓冲区;buffering<0 表示缓冲大小为系统默认;buffering=1 表示为行缓冲模式,遇到换行符才 flush;buffering>1则指定缓冲区大小,单位字节。
1-asd
2-jkl
file = open('test.txt', 'r+', encoding='utf-8')
print(file.tell()) # 返回当前的指针位置
print(file.read()) # 当前打开模式中指针指向起始位置,因此返回整个文本
# 1-asd
# 2-jkl
# 改变指针指向
file.seek(2)
print(file.tell()) # 1
print(file.read())
# asd
# 2-jkl
file.close()
print(file.read())
# 1-asd
# 2-jkl
print(file.read(2)) # 指定读取内容的长度
# 1-
print(file.readline()) # 获取指针所在行的内容
# 1-asd
file.seek(7)
print(file.readline(2)) # 获取指针所在行指定长度的内容
# 2-
print(file.readlines()) # 获取所有行
# ['1-asd\n', '2-jkl']
file.close()
file.write('3-qwe')
# 覆盖写模式下, 默认在文件内容开头覆盖写入, 否则在指针所在位置覆盖写入指定的长度, 例如以上为覆盖5个长度的文件内容
# 追加写模式下, 默认在文件内容末尾追加写入, 否则在指针所在位置追加写入指定的长度
file.seek(6)
file.write('4-xyz')
file.writelines(['1-asd\n', '2-jkl\n', '3-qwe\n', '4-xyz']) # 将字符串列表写入文件
file.close()
当使用 open() 打开一个文件时,如果不关闭文件,文件对象会保持使用状态,占用计算机资源。
file.write('3-qwe')
# 覆盖写模式下, 默认在文件内容开头覆盖写入, 否则在指针所在位置覆盖写入指定的长度, 例如以上为覆盖5个长度的文件内容
# 追加写模式下, 默认在文件内容末尾追加写入, 否则在指针所在位置追加写入指定的长度
file.seek(6)
file.write('4-xyz')
file.writelines(['1-asd\n', '2-jkl\n', '3-qwe\n', '4-xyz']) # 将字符串列表写入文件
file.close()
对于类似于文件对象的 IO 对象,一般来说都需要在不使用的时候关闭、注销以释放资源。
解决文件关闭的问题,除了使用 close() 函数,还可以使用上下文管理的 with 语句块
with open('test', 'r+', encoding='utf-8') as file:
pass
# 进入with语句块的对象, 在离开时一定会关闭
# 即使在with语句块中出现了异常中断, 该对象依然会进行关闭
从数据库中提取记录,写入文件中
写入文本:
import pymysql
from dbutils.pooled_db import PooledDB
pool = PooledDB(creator=pymysql, host='127.0.0.1', user='root', passwd='root', charset='utf8')
conn = pool.connection()
cursor = conn.cursor(pymysql.cursors.DictCursor)
apply_no_tuple = ('S00001', 'S00002')
cursor.execute(f'select * from test.demo where apply_no in {apply_no_tuple};')
result = cursor.fetchall()
# [{'id': 1, 'apply_no': 'S00001', 'order_no': 'XNA0000000000', 'fb_no': 'xna', 'id_no': '51735219750222XXXX', 'loan_amt': 6000.0, 'period': 1, 'product_name': 'zydb', 'create_datetime': datetime.datetime(2022, 1, 5, 16, 6, 42), 'update_datetime': datetime.datetime(2022, 1, 5, 16, 6, 42)},
# {'id': 2, 'apply_no': 'S00002', 'order_no': 'XNA0000000001', 'fb_no': 'xnb', 'id_no': '156246199603273XXX', 'loan_amt': 4500.0, 'period': 2, 'product_name': 'zydb', 'create_datetime': datetime.datetime(2022, 1, 5, 16, 7, 15), 'update_datetime': datetime.datetime(2022, 1, 5, 16, 7, 15)}]
with open(file='test.txt', mode='w+', encoding='utf-8') as file:
for apply_info in result:
apply_no = apply_info['apply_no']
order_no = apply_info['order_no']
fb_no = apply_info['fb_no']
id_no = apply_info['id_no']
loan_amt = apply_info['loan_amt']
period = apply_info['period']
product_name = apply_info['product_name']
create_datetime = apply_info['create_datetime'].__format__('%Y-%m-%d')
file.write(
f'{apply_no}|+|{order_no}|+|{fb_no}|+|{id_no}|+|{loan_amt}|+|{period}|+|{product_name}|+|{create_datetime}\n')
反向处理:
with open(file='test.txt', mode='r+', encoding='utf-8') as file:
for line in file:
if line[-1] == '\n':
line = line[: -1]
params = line.split('+')
for i in range(len(params)):
params[i] = params[i].replace('|', '')
apply_no = params[0]
order_no = params[1]
fb_no = params[2]
id_no = params[3]
loan_amt = float(params[4])
period = int(params[5])
product_name = params[6]
create_datetime = datetime.datetime.strptime(params[7], '%Y-%m-%d')
cursor.execute('select max(id) as id from test.demo;')
max_id = cursor.fetchone()['id']
if max_id is None:
max_id = 1
else:
max_id += 1
try:
cursor.execute(f'insert into test.demo values ('
f'{max_id}, "{apply_no}", "{order_no}", "{fb_no}", "{id_no}", {loan_amt}, {period}, "{product_name}", "{create_datetime}", now());')
except:
conn.rollback()
else:
conn.commit()