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

from ftplib import FTP

# Создаём экземпляр FTP-клиента
ftp = FTP()

# Подключаемся к серверу
ftp.connect('ftp.example.com', 21)

# Проверяем ответ сервера
welcome = ftp.getwelcome()
print(f"Сервер приветствует нас: {welcome}")

# Закрываем соединение
ftp.quit()

ftp = FTP()
ftp.set_pasv(True) # Включение пассивного режима
try:
ftp.connect(server, port)
ftp.login(user, password)
with open(backup_file, 'rb') as f:
ftp.storbinary(f'STOR {backup_name}', f)
except Exception as e:
send_alert(f"Backup failed: {e}")
retry_later()
finally:
ftp.quit()

from ftplib import FTP

# Создаём экземпляр FTP-клиента
ftp = FTP()

# Подключаемся к серверу
ftp.connect('ftp.example.com', 21)

# Авторизуемся
ftp.login(user='username', passwd='password')

# Проверяем текущую директорию
current_dir = ftp.pwd()
print(f"Текущая директория: {current_dir}")

# Закрываем соединение
ftp.quit()

# Анонимное подключение
ftp.login() # По умолчанию использует anonymous/anonymous@

from ftplib import FTP, error_perm

try:
ftp = FTP()
ftp.connect('ftp.example.com', 21)
try:
ftp.login('username', 'password')
print("Успешная авторизация!")
# Выполнение FTP-операций...
except error_perm as e:
if str(e).startswith('530'):
print("Ошибка авторизации: неверные учетные данные")
else:
print(f"Ошибка при авторизации: {e}")
except Exception as e:
print(f"Ошибка при подключении: {e}")
finally:
# Попытка закрыть соединение, если оно было установлено
if 'ftp' in locals() and ftp.sock is not None:
ftp.quit()

from ftplib import FTP_TLS

# Создаём экземпляр защищенного FTP-клиента
ftps = FTP_TLS()
ftps.connect('ftps.example.com', 21)
ftps.login('username', 'password')

# Явное включение защиты данных
ftps.prot_p()

# Выполнение операций...

ftps.quit()

from ftplib import FTP

# Подключаемся к серверу
ftp = FTP('ftp.example.com')
ftp.login('username', 'password')

# Получаем текущую директорию
current_dir = ftp.pwd()
print(f"Текущая директория: {current_dir}")

# Получаем список файлов в текущей директории
files = ftp.nlst()
print(f"Файлы в текущей директории: {files}")

# Переходим в другую директорию
ftp.cwd('/public/docs')
print(f"Новая текущая директория: {ftp.pwd()}")

# Возвращаемся в родительскую директорию
ftp.cwd('..')
print(f"После перехода вверх: {ftp.pwd()}")

ftp.quit()

# Получаем детальный список файлов с их атрибутами
print("Детальный список файлов:")
ftp.retrlines('LIST')

# Получаем и обрабатываем детальный список
file_info = []
ftp.retrlines('LIST', file_info.append)

for item in file_info:
# Парсинг строки вывода LIST
parts = item.split(None, 8)
if len(parts) >= 9:
permissions, _, owner, group, size, month, day, time_or_year, name = parts
is_directory = permissions.startswith('d')
print(f"{'Директория' if is_directory else 'Файл'}: {name}, Размер: {size} байт")

# Создаём новую директорию
try:
ftp.mkd('new_folder')
print("Директория успешно создана")
except:
print("Директория уже существует или нет прав на создание")

# Переходим в созданную директорию
ftp.cwd('new_folder')

# Возвращаемся и переименовываем директорию
ftp.cwd('..')
ftp.rename('new_folder', 'renamed_folder')
print("Директория переименована")

# Удаляем директорию
ftp.rmd('renamed_folder')
print("Директория удалена")

def list_ftp_dir(ftp, path):
files = []
dirs = []

try:
ftp.cwd(path)
items = []
ftp.retrlines('LIST', items.append)

for item in items:
parts = item.split(None, 8)
if len(parts) < 9:
continue

name = parts[8]
if name in ('.', '..'):
continue

if parts[0].startswith('d'):
dirs.append(name)
else:
size = int(parts[4])
files.append((name, size))

for subdir in dirs:
subpath = path + '/' + subdir if path != '/' else '/' + subdir
subfiles = list_ftp_dir(ftp, subpath)
for file_info in subfiles:
name, size = file_info
files.append((subdir + '/' + name, size))

if path != '/':
ftp.cwd('..')
except Exception as e:
print(f"Ошибка при обходе {path}: {e}")

return files

from ftplib import FTP

# Подключаемся к серверу
ftp = FTP('ftp.example.com')
ftp.login('username', 'password')

# Загрузка бинарного файла (например, изображения)
with open('image.png', 'rb') as file:
ftp.storbinary(f'STOR image.png', file)
print("Изображение успешно загружено")

# Загрузка текстового файла
with open('data.txt', 'rb') as file: # Открываем в бинарном режиме даже для текста
ftp.storbinary(f'STOR data.txt', file)
print("Текстовый файл успешно загружен")

ftp.quit()

# Скачивание бинарного файла
with open('downloaded_image.png', 'wb') as file:
ftp.retrbinary(f'RETR image.png', file.write)
print("Изображение успешно скачано")

# Скачивание текстового файла
with open('downloaded_data.txt', 'wb') as file: # Также используем бинарный режим
ftp.retrbinary(f'RETR data.txt', file.write)
print("Текстовый файл успешно скачан")

def download_with_progress(ftp, remote_file, local_file):
file_size = ftp.size(remote_file)
downloaded = 0

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

print(f"Начинаю загрузку файла {remote_file} размером {file_size} байт")
ftp.retrbinary(f"RETR {remote_file}", callback)
print("\nЗагрузка завершена!")

# Использование функции для скачивания с отслеживанием прогресса
with open('large_file.zip', 'wb') as file:
download_with_progress(ftp, 'large_file.zip', file)

import time
from ftplib import FTP, error_temp

def upload_with_retry(ftp, local_file, remote_name, max_attempts=3):
attempts = 0
while attempts < max_attempts:
try:
with open(local_file, 'rb') as file:
ftp.storbinary(f'STOR {remote_name}', file)
print(f"Файл {local_file} успешно загружен как {remote_name}")
return True
except error_temp as e:
attempts += 1
wait_time = 2 ** attempts # Экспоненциальная выдержка
print(f"Временная ошибка: {e}. Повторная попытка через {wait_time} сек...")
time.sleep(wait_time)

print(f"Не удалось загрузить файл после {max_attempts} попыток")
return False

# Использование функции
upload_with_retry(ftp, 'important_data.csv', 'data.csv')

from ftplib import FTP_TLS

# Создаем защищенное соединение
ftps = FTP_TLS()
ftps.connect('ftps.example.com', 21)
ftps.login('username', 'password')
ftps.prot_p() # Включаем защиту данных

# Теперь загрузка и скачивание будут происходить по защищенному каналу
with open('confidential.pdf', 'rb') as file:
ftps.storbinary(f'STOR confidential.pdf', file)

ftps.quit()

import os
import time
import schedule
from datetime import datetime
from ftplib import FTP

def backup_to_ftp():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"backup_{timestamp}.zip"

# Здесь код для создания архива (используйте модуль zipfile или subprocess)
# ...

# Загрузка архива на FTP
try:
ftp = FTP('ftp.example.com')
ftp.login('backup_user', 'password')

# Проверка и создание директории для текущего месяца
current_month = datetime.now().strftime("%Y-%m")
try:
ftp.mkd(current_month)
except:
pass # Директория уже существует

ftp.cwd(current_month)

with open(backup_filename, 'rb') as file:
ftp.storbinary(f'STOR {backup_filename}', file)

# Удаление старых резервных копий (оставляем только последние 5)
ftp.cwd('/')
for month_dir in ftp.nlst():
if not month_dir.startswith('20'): # Пропускаем нерелевантные директории
continue

ftp.cwd(month_dir)
backup_files = sorted([f for f in ftp.nlst() if f.startswith('backup_')])

# Удаляем старые файлы, оставляя последние 5
for old_file in backup_files[:-5]:
try:
ftp.delete(old_file)
print(f"Удален устаревший файл: {month_dir}/{old_file}")
except:
print(f"Не удалось удалить файл: {month_dir}/{old_file}")

ftp.cwd('/')

ftp.quit()

# Удаляем локальный файл бэкапа после успешной загрузки
os.remove(backup_filename)
print(f"Резервное копирование выполнено успешно: {backup_filename}")
except Exception as e:
print(f"Ошибка при выполнении резервного копирования: {e}")

# Планируем выполнение каждый день в 3 часа ночи
schedule.every().day.at("03:00").do(backup_to_ftp)

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

import os
from ftplib import FTP

def sync_directories(local_dir, remote_dir, ftp):
"""Синхронизирует локальную директорию с удаленной на FTP-сервере"""

# Создаем удаленную директорию, если она не существует
try:
ftp.cwd(remote_dir)
except:
parent_dir = os.path.dirname(remote_dir)
dir_name = os.path.basename(remote_dir)

if parent_dir:
ftp.cwd(parent_dir)

ftp.mkd(dir_name)
ftp.cwd(dir_name)

# Получаем списки файлов
local_files = os.listdir(local_dir)
remote_files = ftp.nlst()

# Загружаем новые или измененные файлы
for local_file in local_files:
local_path = os.path.join(local_dir, local_file)

# Если это директория, рекурсивно обрабатываем ее
if os.path.isdir(local_path):
new_remote_dir = os.path.join(remote_dir, local_file).replace('\\', '/')
sync_directories(local_path, new_remote_dir, ftp)
else:
# Проверяем, существует ли файл на сервере и отличается ли по размеру
local_size = os.path.getsize(local_path)

if local_file in remote_files:
try:
remote_size = ftp.size(local_file)
if remote_size == local_size:
print(f"Пропуск существующего файла: {local_file}")
continue
except:
pass # Файл не существует или не получилось определить размер

# Загружаем файл
print(f"Загрузка файла: {local_file}")
with open(local_path, 'rb') as f:
ftp.storbinary(f'STOR {local_file}', f)

# Возвращаемся в родительскую директорию
if remote_dir != '/':
ftp.cwd('..')

# Использование функции
try:
ftp = FTP('ftp.example.com')
ftp.login('username', 'password')

sync_directories('./local_website', '/public_html', ftp)

ftp.quit()
print("Синхронизация успешно завершена")
except Exception as e:
print(f"Ошибка при синхронизации: {e}")

import time
from ftplib import FTP

def process_file(filename):
"""Обрабатывает скачанный файл (пример)"""
print(f"Обработка файла: {filename}")
# Здесь код для обработки файла
# ...

def monitor_ftp_directory():
ftp = FTP('ftp.example.com')
ftp.login('username', 'password')
ftp.cwd('/incoming')

processed_files = set() # Множество уже обработанных файлов

while True:
try:
current_files = set(ftp.nlst())
new_files = current_files – processed_files

for filename in new_files:
print(f"Обнаружен новый файл: {filename}")

# Скачиваем файл
with open(filename, 'wb') as f:
ftp.retrbinary(f'RETR {filename}', f.write)

# Обрабатываем файл
process_file(filename)

# Перемещаем файл в архивную директорию на сервере
try:
ftp.rename(filename, f'/processed/{filename}')
except:
# Возможно, директория не существует, создаем ее
try:
ftp.mkd('/processed')
ftp.rename(filename, f'/processed/{filename}')
except Exception as e:
print(f"Ошибка при перемещении файла в архив: {e}")

# Добавляем в список обработанных
processed_files.add(filename)

# Пауза перед следующей проверкой
time.sleep(60)

# Обновляем подключение, чтобы избежать тайм-аута
ftp.voidcmd('NOOP')

except Exception as e:
print(f"Ошибка при мониторинге: {e}")

# Повторное подключение в случае ошибки
try:
ftp.quit()
except:
pass

time.sleep(30) # Пауза перед повторным подключением

ftp = FTP('ftp.example.com')
ftp.login('username', 'password')
ftp.cwd('/incoming')

# Запуск мониторинга
monitor_ftp_directory()

import logging
import os
from ftplib import FTP, all_errors
from datetime import datetime

# Настраиваем логирование
logging.basicConfig(
filename='ftp_automation.log',
level=logging.INFO,
format='%(asctime)s – %(levelname)s – %(message)s'
)

class FTPAutomation:
def __init__(self, server, username, password):
self.server = server
self.username = username
self.password = password
self.ftp = None

def connect(self):
"""Устанавливает соединение с FTP-сервером"""
try:
self.ftp = FTP(self.server)
self.ftp.login(self.username, self.password)
logging.info(f"Успешное подключение к {self.server}")
return True
except all_errors as e:
logging.error(f"Ошибка подключения к FTP: {e}")
return False

def disconnect(self):
"""Закрывает соединение с сервером"""
if self.ftp:
try:
self.ftp.quit()
logging.info("Соединение с FTP закрыто")
except:
logging.warning("Ошибка при закрытии соединения")

def upload_directory(self, local_dir, remote_dir):
"""Загружает директорию на FTP-сервер"""
if not self.ftp:
if not self.connect():
return False

try:
# Проверяем и создаем удаленную директорию
self.ensure_remote_directory(remote_dir)

# Загружаем все файлы и поддиректории
for root, dirs, files in os.walk(local_dir):
# Вычисляем относительный путь
rel_path = os.path.relpath(root, local_dir)
if rel_path == '.':
rel_path = ''

# Создаем текущую директорию на сервере
current_remote_dir = os.path.join(remote_dir, rel_path).replace('\\', '/')
self.ensure_remote_directory(current_remote_dir)

# Загружаем все файлы в текущей директории
for file in files:
local_file = os.path.join(root, file)
remote_file = os.path.join(current_remote_dir, file).replace('\\', '/')

try:
with open(local_file, 'rb') as f:
self.ftp.storbinary(f'STOR {remote_file}', f)
logging.info(f"Загружен файл: {remote_file}")
except all_errors as e:
logging.error(f"Ошибка загрузки файла {local_file}: {e}")

return True

except all_errors as e:
logging.error(f"Ошибка при загрузке директории {local_dir}: {e}")
return False

def ensure_remote_directory(self, path):
"""Проверяет существование директории на сервере и создает ее при необходимости"""
if path == '/' or not path:
return

# Разбиваем путь на компоненты
components = path.split('/')
current_path = ''

for component in components:
if not component:
continue

if current_path:
current_path += '/' + component
else:
current_path = '/' + component

try:
self.ftp.cwd(current_path)
# Если успешно перешли, директория существует
except all_errors:
# Создаем директорию
try:
self.ftp.mkd(component)
self.ftp.cwd(component)
logging.info(f"Создана директория: {current_path}")
except all_errors as e:
logging.error(f"Ошибка создания директории {current_path}: {e}")
raise

# Возвращаемся в корневую директорию
self.ftp.cwd('/')

# Пример использования класса
if __name__ == "__main__":
automation = FTPAutomation('ftp.example.com', 'username', 'password')

try:
# Загружаем директорию с проектом на сервер
result = automation.upload_directory('./project', '/www/project')

if result:
print("Загрузка проекта выполнена успешно")
else:
print("Произошла ошибка при загрузке проекта")
finally:
automation.disconnect()