Python для работы с файлами: от текстовых до бинарных форматов

# Чтение текстового файла
with open('data.txt', 'r', encoding='utf-8') as file:
content = file.read()

# Запись в текстовый файл
with open('output.txt', 'w', encoding='utf-8') as file:
file.write('Данные успешно записаны')

# Чтение бинарного файла
with open('image.jpg', 'rb') as file:
binary_data = file.read()

# Запись бинарных данных
with open('output.bin', 'wb') as file:
file.write(b'\x00\x01\x02')

with open('data.bin', 'rb+') as file:
# Перемещение на 10 байт от начала
file.seek(10)
# Чтение 5 байт с текущей позиции
data = file.read(5)
# Запись данных в текущую позицию
file.write(b'\xFF\xFF')

# Эффективная построчная обработка
with open('huge_log.txt', 'r') as file:
for line in file:
if 'ERROR' in line:
print(f"Найдена ошибка: {line.strip()}")

import csv

# Чтение CSV файла
with open('data.csv', 'r', newline='', encoding='utf-8') as file:
reader = csv.reader(file)
header = next(reader) # Пропускаем заголовок
for row in reader:
print(f"Имя: {row[0]}, Возраст: {row[1]}")

# Запись в CSV файл
with open('output.csv', 'w', newline='', encoding='utf-8') as file:
writer = csv.writer(file)
writer.writerow(['Имя', 'Возраст', 'Город'])
writer.writerow(['Алексей', 25, 'Москва'])
writer.writerow(['Мария', 30, 'Санкт-Петербург'])

# Использование DictReader для более наглядной работы
with open('data.csv', 'r', encoding='utf-8', newline='') as file:
reader = csv.DictReader(file)
for row in reader:
print(f"Имя: {row['Имя']}, Город: {row['Город']}")

import json

# Преобразование объекта Python в JSON
data = {
"name": "Иван",
"age": 30,
"city": "Москва",
"languages": ["Python", "JavaScript", "Go"],
"is_developer": True
}
json_string = json.dumps(data, ensure_ascii=False, indent=4)
print(json_string)

# Запись в файл
with open('data.json', 'w', encoding='utf-8') as file:
json.dump(data, file, ensure_ascii=False, indent=4)

# Чтение из файла
with open('data.json', 'r', encoding='utf-8') as file:
loaded_data = json.load(file)
print(f"Имя: {loaded_data['name']}, Возраст: {loaded_data['age']}")

import xml.etree.ElementTree as ET

# Создание XML документа
root = ET.Element("employees")
employee = ET.SubElement(root, "employee", attrib={"id": "1"})
ET.SubElement(employee, "name").text = "Иван Иванов"
ET.SubElement(employee, "position").text = "Разработчик"

# Запись в файл
tree = ET.ElementTree(root)
tree.write("employees.xml", encoding="utf-8", xml_declaration=True)

# Чтение из файла
tree = ET.parse("employees.xml")
root = tree.getroot()
for employee in root.findall("employee"):
name = employee.find("name").text
position = employee.find("position").text
print(f"{name} работает как {position}")

from lxml import etree

# XPath запросы
tree = etree.parse("employees.xml")
developers = tree.xpath("//employee[position='Разработчик']/name/text()")
print(f"Разработчики: {', '.join(developers)}")

import pandas as pd

# Чтение Excel-файла
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')

# Базовый анализ
print(f"Количество записей: {len(df)}")
print(f"Средний возраст: {df['Возраст'].mean()}")

# Фильтрация данных
filtered_df = df[df['Зарплата'] > 100000]
print(f"Сотрудники с высокой зарплатой:\n{filtered_df[['Имя', 'Должность']]}")

# Запись в Excel
filtered_df.to_excel('high_salary.xlsx', index=False)

from openpyxl import Workbook, load_workbook
from openpyxl.styles import Font, PatternFill, Alignment

# Создание нового файла с форматированием
wb = Workbook()
ws = wb.active
ws.title = "Отчет"

# Заголовки с форматированием
headers = ["Продукт", "Продажи Q1", "Продажи Q2", "Изменение"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
cell.alignment = Alignment(horizontal='center')

# Данные
data = [
["Товар A", 1200, 1450, "=D2-C2"],
["Товар B", 950, 1100, "=D3-C3"],
["Товар C", 1500, 1380, "=D4-C4"]
]

for row_idx, row_data in enumerate(data, 2):
for col_idx, cell_value in enumerate(row_data, 1):
ws.cell(row=row_idx, column=col_idx, value=cell_value)

# Настройка ширины столбцов
for col in ws.columns:
max_length = 0
column = col[0].column_letter
for cell in col:
if cell.value:
max_length = max(max_length, len(str(cell.value)))
ws.column_dimensions[column].width = max_length + 2

wb.save("sales_report.xlsx")

import PyPDF2

# Извлечение текста
with open('document.pdf', 'rb') as file:
reader = PyPDF2.PdfFileReader(file)
text = ""
for page_num in range(reader.numPages):
text += reader.getPage(page_num).extractText()

print(f"Количество страниц: {reader.numPages}")
print(f"Первые 100 символов: {text[:100]}...")

# Объединение PDF-документов
merger = PyPDF2.PdfFileMerger()
files = ['doc1.pdf', 'doc2.pdf', 'doc3.pdf']
for file in files:
merger.append(file)
merger.write('merged.pdf')
merger.close()

from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
from reportlab.lib import colors
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.platypus import SimpleDocTemplate, Paragraph, Table, TableStyle

# Создание простого PDF
c = canvas.Canvas("simple.pdf", pagesize=letter)
c.setFont("Helvetica", 16)
c.drawString(100, 750, "Отчёт о продажах")
c.setFont("Helvetica", 12)
c.drawString(100, 700, "Подготовлен: 15.05.2023")
c.save()

# Создание сложного отчёта
doc = SimpleDocTemplate("report.pdf", pagesize=letter)
styles = getSampleStyleSheet()
elements = []

# Заголовок
title = Paragraph("Квартальный отчёт", styles['Title'])
elements.append(title)

# Таблица данных
data = [
['Регион', 'Q1', 'Q2', 'Q3', 'Q4', 'Итого'],
['Москва', 120000, 135000, 140000, 155000, 550000],
['Петербург', 95000, 110000, 115000, 130000, 450000],
['Казань', 55000, 60000, 65000, 70000, 250000],
['Новосибирск', 45000, 50000, 55000, 60000, 210000]
]

table = Table(data)
table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.grey),
('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
('ALIGN', (0, 0), (-1, -1), 'CENTER'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('BACKGROUND', (0, 1), (-1, -1), colors.beige),
('GRID', (0, 0), (-1, -1), 1, colors.black)
]))
elements.append(table)

doc.build(elements)

# НЕ делайте так с большими файлами!
with open('huge_log.txt', 'r') as file:
content = file.read() # Загружает весь файл в память!
lines = content.split('\n')
for line in lines:
process_line(line)

# Вместо этого используйте итерацию
with open('huge_log.txt', 'r') as file:
for line in file: # Читает по одной строке за раз
process_line(line.strip())

import pandas as pd

# Обработка CSV по частям
chunk_size = 10000 # Количество строк в одном чанке
for chunk in pd.read_csv('massive_data.csv', chunksize=chunk_size):
# Обработка только части данных за раз
processed_data = process_chunk(chunk)
# Сохранение результатов
processed_data.to_csv('results.csv', mode='a', header=False, index=False)

import ijson

# Потоковое чтение JSON
with open('large.json', 'rb') as file:
# Извлечение только нужных элементов
for item in ijson.items(file, 'items.item'):
if item['category'] == 'electronics':
process_electronic_item(item)

# Эффективное копирование большого файла
def copy_large_file(src, dst, chunk_size=1024*1024): # 1MB chunks
with open(src, 'rb') as fsrc, open(dst, 'wb') as fdst:
while True:
buffer = fsrc.read(chunk_size)
if not buffer:
break
fdst.write(buffer)

import mmap
import re

# Поиск по паттерну в большом файле
with open('massive_log.txt', 'r+b') as file:
# Создание отображения файла в память
mm = mmap.mmap(file.fileno(), 0)

# Компиляция регулярного выражения для эффективного поиска
pattern = re.compile(b'ERROR.*Database connection failed')

# Поиск всех совпадений
for match in pattern.finditer(mm):
start = max(0, match.start() – 100) # Контекст до ошибки
end = min(mm.size(), match.end() + 100) # Контекст после ошибки
context = mm[start:end]
print(f"Найдена ошибка: {context.decode('utf-8')}")

mm.close()

import gzip

# Чтение сжатого файла построчно
with gzip.open('logs.gz', 'rt', encoding='utf-8') as file:
for line in file:
process_log_line(line.strip())

# Параллельная обработка с использованием ThreadPoolExecutor
from concurrent.futures import ThreadPoolExecutor
import os

def process_file(filename):
results = []
with open(filename, 'r') as file:
# Обработка файла...
return results

files = [f for f in os.listdir('data_dir') if f.endswith('.csv')]
with ThreadPoolExecutor(max_workers=os.cpu_count()) as executor:
all_results = list(executor.map(process_file, files))

import pandas as pd
import json
import os

def excel_to_json(excel_dir, json_dir):
"""Конвертирует все Excel файлы в директории в JSON формат"""
if not os.path.exists(json_dir):
os.makedirs(json_dir)

for filename in os.listdir(excel_dir):
if filename.endswith(('.xlsx', '.xls')):
filepath = os.path.join(excel_dir, filename)
print(f"Обработка файла: {filepath}")

# Чтение Excel
df = pd.read_excel(filepath)

# Преобразование в список словарей
records = df.to_dict(orient='records')

# Создание имени для JSON файла
json_filename = os.path.splitext(filename)[0] + '.json'
json_path = os.path.join(json_dir, json_filename)

# Запись в JSON
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(records, f, ensure_ascii=False, indent=2)

print(f"Создан файл: {json_path}")

# Использование
excel_to_json('data/excel', 'data/json')

import re
from collections import Counter
import matplotlib.pyplot as plt
from datetime import datetime
import pandas as pd

def analyze_web_server_logs(log_file, output_dir):
"""Анализирует логи веб-сервера и генерирует отчеты"""
# Регулярное выражение для Apache логов
pattern = r'(\S+) \S+ \S+ \[(.*?)\] "(\S+) (\S+) \S+" (\d+) (\S+)'

ip_addresses = []
endpoints = []
status_codes = []
bytes_sent = []
dates = []

with open(log_file, 'r') as file:
for line in file:
match = re.search(pattern, line)
if match:
ip, date_str, method, endpoint, status, byte = match.groups()

# Преобразование даты
date = datetime.strptime(date_str.split()[0], '%d/%b/%Y')

ip_addresses.append(ip)
endpoints.append(endpoint)
status_codes.append(int(status))
bytes_sent.append(int(byte) if byte != '-' else 0)
dates.append(date)

# Создание DataFrame для анализа
df = pd.DataFrame({
'ip': ip_addresses,
'endpoint': endpoints,
'status': status_codes,
'bytes': bytes_sent,
'date': dates
})

# Анализ наиболее посещаемых страниц
top_endpoints = Counter(endpoints).most_common(10)
top_endpoints_df = pd.DataFrame(top_endpoints, columns=['Endpoint', 'Visits'])
top_endpoints_df.to_csv(f"{output_dir}/top_endpoints.csv", index=False)

# Анализ ошибок
errors_df = df[df['status'] >= 400]
error_counts = errors_df.groupby(['status', 'endpoint']).size().reset_index(name='count')
error_counts.sort_values('count', ascending=False).to_csv(f"{output_dir}/errors.csv", index=False)

# Визуализация трафика по дням
daily_traffic = df.groupby(df['date']).size()
plt.figure(figsize=(12, 6))
daily_traffic.plot(kind='line')
plt.title('Ежедневный трафик')
plt.xlabel('Дата')
plt.ylabel('Количество запросов')
plt.savefig(f"{output_dir}/daily_traffic.png")

print(f"Отчеты сохранены в директорию: {output_dir}")

# Использование
analyze_web_server_logs('/var/log/apache2/access.log', 'reports')

from reportlab.lib.pagesizes import A4
from reportlab.lib import colors
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer
from reportlab.lib.styles import getSampleStyleSheet
import pandas as pd
import matplotlib.pyplot as plt
import os

def generate_sales_report(sales_data, output_pdf):
"""Создает PDF-отчет по продажам на основе данных CSV"""
# Загрузка данных о продажах
df = pd.read_csv(sales_data)

# Подготовка данных для отчета
total_sales = df['Amount'].sum()
monthly_sales = df.groupby('Month')['Amount'].sum().reset_index()
product_sales = df.groupby('Product')['Amount'].sum().reset_index()
product_sales = product_sales.sort_values('Amount', ascending=False)

# Создание графика продаж по месяцам
plt.figure(figsize=(8, 4))
plt.bar(monthly_sales['Month'], monthly_sales['Amount'])
plt.title('Продажи по месяцам')
plt.xlabel('Месяц')
plt.ylabel('Объем продаж')
chart_path = 'temp_sales_chart.png'
plt.savefig(chart_path)

# Создание PDF
doc = SimpleDocTemplate(output_pdf, pagesize=A4)
styles = getSampleStyleSheet()
elements = []

# Заголовок
title = Paragraph(f"Отчет по продажам за {df['Year'].iloc[0]} год", styles['Title'])
elements.append(title)
elements.append(Spacer(1, 20))

# Общая информация
elements.append(Paragraph(f"Общий объем продаж: {total_sales:,.2f} руб.", styles['Normal']))
elements.append(Spacer(1, 10))

# Таблица с топ-продуктами
elements.append(Paragraph("Продажи по продуктам", styles['Heading2']))
top_products = product_sales.head(5)
data = [['Продукт', 'Объем продаж (руб.)']]
for _, row in top_products.iterrows():
data.append([row['Product'], f"{row['Amount']:,.2f}"])

table = Table(data, colWidths=[300, 150])
table.setStyle(TableStyle([
('BACKGROUND', (0, 0), (-1, 0), colors.grey),
('TEXTCOLOR', (0, 0), (-1, 0), colors.white),
('ALIGN', (0, 0), (-1, -1), 'CENTER'),
('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
('BOTTOMPADDING', (0, 0), (-1, 0), 12),
('GRID', (0, 0), (-1, -1), 1, colors.black)
]))
elements.append(table)
elements.append(Spacer(1, 20))

# Построение документа
doc.build(elements)

# Удаление временного файла графика
os.remove(chart_path)

print(f"Отчет сохранен в {output_pdf}")

# Использование
generate_sales_report('sales_data_2023.csv', 'annual_sales_report_2023.pdf')