Python + FTP: как автоматизировать передачу файлов на сервер

import ftplib

# Создание FTP-объекта
ftp = ftplib.FTP()

from ftplib import FTP_TLS

ftps = FTP_TLS()
ftps.connect('ftp.example.com', 21)
ftps.login('username', 'password')
ftps.prot_p() # Включение защищенного режима для передачи данных

import ftplib

# Создание объекта FTP
ftp = ftplib.FTP()

# Подключение к серверу
ftp.connect('ftp.example.com', 21) # Хост и порт (21 – стандартный)

# Аутентификация
ftp.login('username', 'password') # Логин и пароль

# Проверка подключения
print(ftp.getwelcome()) # Выводит приветствие сервера

# Закрытие соединения после работы
ftp.quit()

ftp = ftplib.FTP('ftp.example.com', 'username', 'password')

# Вывод содержимого текущей директории
ftp.dir() 

# Получение списка файлов в переменную
file_list = []
ftp.dir(file_list.append)
print(file_list)

# Получение списка имен файлов
filenames = ftp.nlst()
print(filenames)

# Узнать текущую директорию
current_dir = ftp.pwd()
print(f"Текущая директория: {current_dir}")

# Перейти в другую директорию
ftp.cwd('/path/to/directory')

# Подняться на уровень выше
ftp.cwd('..')

# Создать новую директорию
ftp.mkd('new_directory')

# Удалить директорию
ftp.rmd('directory_to_remove')

# Получение детальной информации о файлах
details = []
ftp.dir(details.append)

for item in details:
# Парсинг строки с информацией о файле
print(item) # Например: "drwxr-xr-x 2 user group 4096 Jan 1 12:34 directory_name"

# Включение пассивного режима (рекомендуется в большинстве случаев)
ftp.set_pasv(True)

# Включение подробного вывода для отладки (0 – выкл., 1 – команды, 2 – подробно)
ftp.set_debuglevel(2)

import ftplib

def download_file(server, username, password, remote_path, local_path):
"""
Скачивает файл с FTP-сервера

Args:
server (str): Адрес FTP-сервера
username (str): Имя пользователя
password (str): Пароль
remote_path (str): Путь к файлу на сервере
local_path (str): Локальный путь для сохранения
"""
try:
# Подключение к серверу
with ftplib.FTP(server) as ftp:
ftp.login(username, password)

# Открытие локального файла для записи
with open(local_path, 'wb') as local_file:
# Скачивание файла
ftp.retrbinary(f'RETR {remote_path}', local_file.write)

print(f"Файл успешно скачан: {local_path}")
return True

except Exception as e:
print(f"Ошибка при скачивании файла: {str(e)}")
return False

# Пример использования
download_file('ftp.example.com', 'user', 'password', 
'/path/to/remote/file.csv', 'local_file.csv')

def upload_file(server, username, password, local_path, remote_path):
"""
Загружает файл на FTP-сервер

Args:
server (str): Адрес FTP-сервера
username (str): Имя пользователя
password (str): Пароль
local_path (str): Путь к локальному файлу
remote_path (str): Путь для сохранения на сервере
"""
try:
# Подключение к серверу
with ftplib.FTP(server) as ftp:
ftp.login(username, password)

# Открытие локального файла для чтения
with open(local_path, 'rb') as local_file:
# Загрузка файла
ftp.storbinary(f'STOR {remote_path}', local_file)

print(f"Файл успешно загружен: {remote_path}")
return True

except Exception as e:
print(f"Ошибка при загрузке файла: {str(e)}")
return False

# Пример использования
upload_file('ftp.example.com', 'user', 'password', 
'data.xlsx', '/upload/data.xlsx')

def download_with_progress(server, username, password, remote_path, local_path):
"""Скачивает файл с отображением прогресса"""

# Получение размера файла
def get_file_size():
ftp = ftplib.FTP(server)
ftp.login(username, password)
ftp.sendcmd("TYPE I") # Переключение в бинарный режим
size = ftp.size(remote_path)
ftp.quit()
return size

try:
total_size = get_file_size()
downloaded = 0

# Функция для отслеживания прогресса
def callback(data):
nonlocal downloaded
downloaded += len(data)
percent = (downloaded / total_size) * 100
print(f"\rПрогресс: {percent:.2f}% ({downloaded}/{total_size})", end="")
return file.write(data)

with ftplib.FTP(server) as ftp:
ftp.login(username, password)
with open(local_path, 'wb') as file:
ftp.retrbinary(f'RETR {remote_path}', callback)

print("\nЗагрузка завершена")
return True

except Exception as e:
print(f"\nОшибка: {str(e)}")
return False

import os

def download_directory(ftp, remote_dir, local_dir):
"""Рекурсивно скачивает директорию с FTP-сервера"""

# Создаем локальную директорию, если она не существует
if not os.path.exists(local_dir):
os.makedirs(local_dir)

# Запоминаем текущую директорию на сервере
initial_dir = ftp.pwd()

try:
# Переходим в указанную директорию
ftp.cwd(remote_dir)

# Получаем список файлов
files = []
ftp.dir(files.append)

# Обрабатываем каждый элемент
for item in files:
# Парсинг вывода команды dir
tokens = item.split()
file_name = tokens[-1]

# Пропускаем специальные директории
if file_name in ('.', '..'):
continue

local_path = os.path.join(local_dir, file_name)

# Проверяем, директория это или файл
if item.startswith('d'): # Директория
download_directory(ftp, file_name, local_path)
else: # Файл
with open(local_path, 'wb') as f:
ftp.retrbinary(f'RETR {file_name}', f.write)
print(f"Скачан файл: {local_path}")

finally:
# Возвращаемся в исходную директорию
ftp.cwd(initial_dir)

import os
import ftplib
import datetime

def sync_directory(local_dir, ftp_server, username, password, remote_dir):
"""
Синхронизирует локальную директорию с FTP-сервером,
загружая только новые или измененные файлы
"""
# Подключаемся к серверу
ftp = ftplib.FTP(ftp_server)
ftp.login(username, password)

# Переходим в удаленную директорию
try:
ftp.cwd(remote_dir)
except ftplib.error_perm:
# Директория не существует, создаем
ftp.mkd(remote_dir)
ftp.cwd(remote_dir)

# Получаем список файлов на сервере и их даты изменения
remote_files = {}
try:
# Для некоторых серверов нужно использовать специальный формат даты
ftp.sendcmd('MDTM')
support_mdtm = True
except ftplib.error_perm:
support_mdtm = False

if support_mdtm:
for filename in ftp.nlst():
try:
# Получение времени модификации файла
timestamp = ftp.sendcmd(f'MDTM {filename}')
remote_files[filename] = timestamp
except ftplib.error_perm:
# Вероятно, это директория
pass

# Перебираем локальные файлы
for root, _, files in os.walk(local_dir):
for filename in files:
local_path = os.path.join(root, filename)

# Определяем относительный путь
rel_path = os.path.relpath(local_path, local_dir)

# Проверяем, есть ли файл на сервере
upload_file = False

if rel_path not in remote_files:
upload_file = True
elif support_mdtm:
# Сравниваем даты, если поддерживается
local_time = os.path.getmtime(local_path)
local_time = datetime.datetime.fromtimestamp(local_time)
# Конвертация в формат FTP
# ...код конвертации...
upload_file = True # Для примера всегда загружаем
else:
# Если нельзя проверить дату, загружаем всегда
upload_file = True

if upload_file:
print(f"Загрузка файла: {rel_path}")
with open(local_path, 'rb') as file:
ftp.storbinary(f'STOR {rel_path}', file)

ftp.quit()
print("Синхронизация завершена")

import schedule
import time

def daily_sync():
"""Функция для ежедневной синхронизации"""
print(f"Запуск синхронизации: {datetime.datetime.now()}")
sync_directory(
local_dir="/path/to/local/files",
ftp_server="ftp.example.com",
username="user",
password="pass",
remote_dir="/backup"
)

# Планирование ежедневного запуска в 2:00
schedule.every().day.at("02:00").do(daily_sync)

# Запуск планировщика
while True:
schedule.run_pending()
time.sleep(60) # Проверка каждую минуту

def process_and_upload(local_dir, ftp_server, username, password):
"""Обрабатывает файлы разных типов и загружает их на FTP-сервер"""

# Подключение к серверу
ftp = ftplib.FTP(ftp_server)
ftp.login(username, password)

# Перебор файлов в директории
for filename in os.listdir(local_dir):
file_path = os.path.join(local_dir, filename)

# Пропускаем директории
if os.path.isdir(file_path):
continue

# Определяем тип файла по расширению
_, ext = os.path.splitext(filename)
ext = ext.lower()

# Обработка в зависимости от типа
processed_file = None

if ext == '.csv':
# Обработка CSV (например, конвертация в JSON)
processed_file = process_csv(file_path)
upload_path = f"/processed/json/{os.path.basename(processed_file)}"

elif ext in ('.jpg', '.png', '.gif'):
# Обработка изображений (например, изменение размера)
processed_file = process_image(file_path)
upload_path = f"/processed/images/{os.path.basename(processed_file)}"

elif ext == '.log':
# Обработка лог-файлов (например, сжатие)
processed_file = compress_log(file_path)
upload_path = f"/processed/logs/{os.path.basename(processed_file)}"

else:
# Для остальных файлов просто загружаем без обработки
processed_file = file_path
upload_path = f"/raw/{filename}"

# Загрузка файла
if processed_file:
with open(processed_file, 'rb') as file:
ftp.storbinary(f'STOR {upload_path}', file)
print(f"Файл {filename} обработан и загружен как {upload_path}")

ftp.quit()

import smtplib
from email.mime.text import MIMEText

def send_notification(subject, message, recipient):
"""Отправляет email-уведомление"""
msg = MIMEText(message)
msg['Subject'] = subject
msg['From'] = 'ftp-monitor@example.com'
msg['To'] = recipient

try:
smtp = smtplib.SMTP('smtp.example.com')
smtp.send_message(msg)
smtp.quit()
print("Уведомление отправлено")
except Exception as e:
print(f"Ошибка отправки уведомления: {str(e)}")

def automated_ftp_job():
"""Полная автоматизированная задача с уведомлениями"""
start_time = datetime.datetime.now()
success_count = 0
error_count = 0
log_messages = []

try:
# Выполнение основной работы
# ... код синхронизации ...

# Подготовка отчета
end_time = datetime.datetime.now()
duration = (end_time – start_time).total_seconds()

message = f"""
FTP-синхронизация завершена.
Время начала: {start_time}
Время окончания: {end_time}
Продолжительность: {duration} секунд
Успешно: {success_count} файлов
Ошибки: {error_count} файлов

Детали:
{''.join(log_messages)}
"""

send_notification("FTP Sync Report", message, "admin@example.com")

except Exception as e:
send_notification(
"FTP Sync Failed", 
f"Ошибка синхронизации: {str(e)}", 
"admin@example.com"
)

import ftplib
import socket
import time
import logging

# Настройка логгера
logging.basicConfig(
filename='ftp_operations.log',
level=logging.INFO,
format='%(asctime)s – %(levelname)s – %(message)s'
)

def robust_ftp_operation(max_retries=3, retry_delay=5):
"""Декоратор для повторных попыток FTP-операций при ошибках"""
def decorator(func):
def wrapper(*args, **kwargs):
retries = 0
while retries < max_retries:
try:
return func(*args, **kwargs)
except ftplib.error_temp as e:
# Временная ошибка – можно повторить
logging.warning(f"Временная ошибка FTP: {str(e)}. Попытка {retries+1}/{max_retries}")
retries += 1
if retries < max_retries:
time.sleep(retry_delay)
except ftplib.error_perm as e:
# Постоянная ошибка – бессмысленно повторять
logging.error(f"Постоянная ошибка FTP: {str(e)}")
raise
except socket.timeout:
# Таймаут соединения
logging.warning(f"Таймаут FTP-соединения. Попытка {retries+1}/{max_retries}")
retries += 1
if retries < max_retries:
time.sleep(retry_delay)
except socket.error as e:
# Сетевая ошибка
logging.warning(f"Сетевая ошибка: {str(e)}. Попытка {retries+1}/{max_retries}")
retries += 1
if retries < max_retries:
time.sleep(retry_delay)
except Exception as e:
# Неожиданная ошибка
logging.error(f"Неожиданная ошибка: {str(e)}")
raise

# Если все попытки неудачны
logging.error(f"Превышено количество попыток ({max_retries})")
raise Exception(f"Превышено количество попыток ({max_retries})")
return wrapper
return decorator

@robust_ftp_operation(max_retries=5, retry_delay=10)
def download_file_robust(server, username, password, remote_path, local_path):
"""Надежная функция загрузки файла с FTP с повторными попытками"""
with ftplib.FTP(server, timeout=30) as ftp:
ftp.login(username, password)
with open(local_path, 'wb') as local_file:
ftp.retrbinary(f'RETR {remote_path}', local_file.write)

logging.info(f"Файл успешно загружен: {remote_path} -> {local_path}")
return True

import ssl
from ftplib import FTP_TLS

def secure_ftp_upload(server, username, password, local_path, remote_path):
"""Безопасная загрузка файла через FTPS с проверкой сертификата"""
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)

# Для работы с самоподписанными сертификатами (НЕ рекомендуется для продакшн)
# context.check_hostname = False
# context.verify_mode = ssl.CERT_NONE

ftps = FTP_TLS(context=context)
ftps.connect(server, 21)
ftps.login(username, password)
ftps.prot_p() # Включение защищенного режима данных

with open(local_path, 'rb') as file:
ftps.storbinary(f'STOR {remote_path}', file)

ftps.quit()
logging.info(f"Файл безопасно загружен: {local_path} -> {remote_path}")
return True

import os
from dotenv import load_dotenv

# Загрузка переменных окружения из файла .env
load_dotenv()

def secure_credentials():
"""Получение учетных данных из переменных окружения"""
ftp_host = os.getenv("FTP_HOST")
ftp_user = os.getenv("FTP_USER")
ftp_pass = os.getenv("FTP_PASS")

if not all([ftp_host, ftp_user, ftp_pass]):
raise ValueError("Не все учетные данные FTP заданы в переменных окружения")

return ftp_host, ftp_user, ftp_pass

def verify_certificate(connection, cert, errno, depth, result):
"""Проверка сертификата сервера"""
if not result:
logging.warning(f"Ошибка проверки сертификата: {errno}")

# Проверка отпечатка сертификата
expected_fingerprint = "11:22:33:44:55:66:77:88:99:00:AA:BB:CC:DD:EE:FF"
cert_fingerprint = ":".join([f"{x:02X}" for x in cert.digest("sha1")])

if cert_fingerprint != expected_fingerprint:
logging.error(f"Несоответствие отпечатка сертификата! "
f"Ожидалось: {expected_fingerprint}, "
f"Получено: {cert_fingerprint}")
return False

return result

def audit_ftp_operation(operation_type, server, username, path, result):
"""Запись операции в журнал аудита"""
log_entry = {
"timestamp": datetime.datetime.now().isoformat(),
"operation": operation_type,
"server": server,
"username": username,
"path": path,
"result": "success" if result else "failure",
"ip_address": socket.gethostbyname(socket.gethostname())
}

# Запись в JSON-лог
with open('ftp_audit.json', 'a') as f:
f.write(json.dumps(log_entry) + "\n")

# Можно также отправить в централизованную систему мониторинга
# send_to_monitoring_system(log_entry)