317 lines
11 KiB
Python
317 lines
11 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
import fnmatch
|
|
import logging
|
|
import asyncio
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any, List, Tuple, Set
|
|
import ftplib
|
|
import socket
|
|
|
|
logger = logging.getLogger("AGYFTPManager")
|
|
|
|
# Standard patterns for junk, temporary, confidential, and unnecessary files that must NEVER be sent to production FTP
|
|
DEFAULT_FTP_EXCLUDES = [
|
|
# Git and version control
|
|
".git",
|
|
".git/*",
|
|
".gitignore",
|
|
".gitattributes",
|
|
".gitmodules",
|
|
|
|
# Environment and secrets (local secrets must not leak to production without explicit setup)
|
|
".env",
|
|
".env.*",
|
|
"*.env",
|
|
"*.pem",
|
|
"*.key",
|
|
|
|
# AI & Bot internal session / agent configs
|
|
".agents",
|
|
".agents/*",
|
|
".gemini",
|
|
".gemini/*",
|
|
"sessions_data",
|
|
"sessions_data/*",
|
|
"uploads_temp",
|
|
"uploads_temp/*",
|
|
"*.db-shm",
|
|
"*.db-wal",
|
|
|
|
# Python cache & build
|
|
"__pycache__",
|
|
"__pycache__/*",
|
|
"*.pyc",
|
|
"*.pyo",
|
|
"*.pyd",
|
|
".venv",
|
|
".venv/*",
|
|
"venv",
|
|
"venv/*",
|
|
"env",
|
|
"env/*",
|
|
|
|
# Logs & temp files
|
|
"*.log",
|
|
"tmp",
|
|
"tmp/*",
|
|
"temp",
|
|
"temp/*",
|
|
|
|
# IDE & OS clutter
|
|
".DS_Store",
|
|
"Thumbs.db",
|
|
"desktop.ini",
|
|
".idea",
|
|
".idea/*",
|
|
".vscode",
|
|
".vscode/*",
|
|
"*.swp",
|
|
"*.swo",
|
|
"*~",
|
|
]
|
|
|
|
|
|
def should_exclude(rel_path: str, custom_excludes: Optional[List[str]] = None) -> bool:
|
|
"""Checks if a relative path matches any exclusion pattern."""
|
|
excludes = DEFAULT_FTP_EXCLUDES + (custom_excludes or [])
|
|
norm_path = rel_path.replace("\\", "/").strip("/")
|
|
parts = norm_path.split("/")
|
|
|
|
for pattern in excludes:
|
|
pat = pattern.replace("\\", "/").strip("/")
|
|
# Check against full relative path
|
|
if fnmatch.fnmatch(norm_path, pat):
|
|
return True
|
|
# Check against individual directory / file parts
|
|
for part in parts:
|
|
if fnmatch.fnmatch(part, pat):
|
|
return True
|
|
# If pattern is a directory (e.g. .git/* or .git), match if path starts with it
|
|
clean_pat = pat.rstrip("/*")
|
|
if norm_path == clean_pat or norm_path.startswith(clean_pat + "/"):
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
class FTPManager:
|
|
"""Manages FTP connections, testing, and clean intelligent deployments."""
|
|
|
|
def _create_ftp_client(
|
|
self,
|
|
host: str,
|
|
port: int = 21,
|
|
user: str = "",
|
|
password: str = "",
|
|
tls: bool = False,
|
|
timeout: int = 15,
|
|
) -> ftplib.FTP:
|
|
"""Helper to create and authenticate an FTP or FTPS client."""
|
|
if tls:
|
|
ftp = ftplib.FTP_TLS(timeout=timeout)
|
|
else:
|
|
ftp = ftplib.FTP(timeout=timeout)
|
|
|
|
ftp.connect(host=host, port=port, timeout=timeout)
|
|
ftp.login(user=user or "anonymous", passwd=password or "")
|
|
|
|
if tls and isinstance(ftp, ftplib.FTP_TLS):
|
|
ftp.prot_p() # Secure data connection
|
|
|
|
return ftp
|
|
|
|
async def test_connection(
|
|
self,
|
|
host: str,
|
|
port: int = 21,
|
|
user: str = "",
|
|
password: str = "",
|
|
path: str = "/",
|
|
tls: bool = False,
|
|
timeout: int = 10,
|
|
) -> Tuple[bool, str]:
|
|
"""Tests FTP credentials and verifies access to the target remote directory."""
|
|
if not host or not host.strip():
|
|
return False, "آدرس هاست (Host) FTP مشخص نشده است."
|
|
|
|
def _test():
|
|
try:
|
|
ftp = self._create_ftp_client(host.strip(), int(port), user.strip(), password, tls, timeout)
|
|
# Test CWD to target path if specified
|
|
target_path = path.strip() if path else "/"
|
|
if target_path and target_path != "/":
|
|
try:
|
|
ftp.cwd(target_path)
|
|
except Exception as e:
|
|
pwd = ftp.pwd()
|
|
ftp.quit()
|
|
return False, f"اتصال به FTP برقرار شد، اما مسیر ریموت «{target_path}» یافت نشد (مسیر فعلی: {pwd}): {e}"
|
|
|
|
pwd = ftp.pwd()
|
|
try:
|
|
listing = ftp.nlst()
|
|
count = len(listing)
|
|
except Exception:
|
|
count = 0
|
|
ftp.quit()
|
|
return True, f"اتصال با موفقیت برقرار شد. مسیر فعلی: <code>{pwd}</code> (تعداد آیتمها: {count})"
|
|
except (socket.gaierror, socket.timeout) as e:
|
|
return False, f"خطای شبکه / نامعتبر بودن آدرس سرور ({host}:{port}): {e}"
|
|
except ftplib.error_perm as e:
|
|
return False, f"خطای احراز هویت / دسترسی FTP: {e}"
|
|
except Exception as e:
|
|
return False, f"خطا در برقراری ارتباط با FTP: {e}"
|
|
|
|
return await asyncio.to_thread(_test)
|
|
|
|
async def deploy_project(
|
|
self,
|
|
workspace_path: str,
|
|
host: str,
|
|
port: int = 21,
|
|
user: str = "",
|
|
password: str = "",
|
|
remote_path: str = "/",
|
|
tls: bool = False,
|
|
custom_excludes: Optional[List[str]] = None,
|
|
dry_run: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Deploys files from workspace_path to the remote FTP server cleanly.
|
|
Skips all temporary, git, session, and junk files.
|
|
"""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
if not ws.exists() or not ws.is_dir():
|
|
return {
|
|
"success": False,
|
|
"error": f"دایرکتوری پروژه در مسیر {workspace_path} یافت نشد.",
|
|
"files_uploaded": 0,
|
|
"files_skipped": 0,
|
|
"bytes_transferred": 0,
|
|
"duration": 0.0,
|
|
"uploaded_list": [],
|
|
}
|
|
|
|
def _run_deploy():
|
|
start_time = time.time()
|
|
uploaded_files: List[str] = []
|
|
skipped_files: List[str] = []
|
|
total_bytes = 0
|
|
|
|
try:
|
|
ftp = self._create_ftp_client(host.strip(), int(port), user.strip(), password, tls, timeout=20)
|
|
|
|
# Navigate or create remote root path
|
|
target_root = (remote_path or "/").strip().replace("\\", "/")
|
|
if not target_root.startswith("/"):
|
|
target_root = "/" + target_root
|
|
target_root = target_root.rstrip("/")
|
|
if not target_root:
|
|
target_root = "/"
|
|
|
|
def ensure_remote_dir(r_dir: str):
|
|
if r_dir in ("/", ""):
|
|
ftp.cwd("/")
|
|
return
|
|
parts = [p for p in r_dir.split("/") if p]
|
|
curr = ""
|
|
for p in parts:
|
|
curr += "/" + p
|
|
try:
|
|
ftp.cwd(curr)
|
|
except Exception:
|
|
try:
|
|
ftp.mkd(curr)
|
|
ftp.cwd(curr)
|
|
except Exception:
|
|
pass
|
|
|
|
ensure_remote_dir(target_root)
|
|
|
|
# Scan workspace
|
|
all_files_to_upload: List[Tuple[Path, str]] = [] # (local_path, rel_path)
|
|
|
|
for root, dirs, files in os.walk(str(ws)):
|
|
rel_dir = os.path.relpath(root, str(ws))
|
|
if rel_dir == ".":
|
|
rel_dir = ""
|
|
|
|
# Filter out directories in-place to avoid descending into ignored dirs
|
|
dirs_to_keep = []
|
|
for d in dirs:
|
|
dir_rel = f"{rel_dir}/{d}".strip("/")
|
|
if should_exclude(dir_rel, custom_excludes):
|
|
skipped_files.append(dir_rel + "/")
|
|
else:
|
|
dirs_to_keep.append(d)
|
|
dirs[:] = dirs_to_keep
|
|
|
|
for f in files:
|
|
file_rel = f"{rel_dir}/{f}".strip("/")
|
|
if should_exclude(file_rel, custom_excludes):
|
|
skipped_files.append(file_rel)
|
|
else:
|
|
all_files_to_upload.append((Path(root) / f, file_rel))
|
|
|
|
if dry_run:
|
|
ftp.quit()
|
|
return {
|
|
"success": True,
|
|
"dry_run": True,
|
|
"files_to_upload": len(all_files_to_upload),
|
|
"files_skipped": len(skipped_files),
|
|
"bytes_transferred": sum(p.stat().st_size for p, _ in all_files_to_upload if p.exists()),
|
|
"duration": round(time.time() - start_time, 2),
|
|
"uploaded_list": [r for _, r in all_files_to_upload[:50]],
|
|
"remote_path": target_root,
|
|
}
|
|
|
|
# Upload files
|
|
for local_p, rel_p in all_files_to_upload:
|
|
if not local_p.exists():
|
|
continue
|
|
file_size = local_p.stat().st_size
|
|
rel_dir = str(Path(rel_p).parent).replace("\\", "/").strip(".")
|
|
target_dir = f"{target_root}/{rel_dir}".rstrip("/") if rel_dir else target_root
|
|
|
|
ensure_remote_dir(target_dir)
|
|
|
|
file_name = local_p.name
|
|
with open(local_p, "rb") as fp:
|
|
ftp.storbinary(f"STOR {file_name}", fp)
|
|
|
|
uploaded_files.append(rel_p)
|
|
total_bytes += file_size
|
|
|
|
ftp.quit()
|
|
duration = round(time.time() - start_time, 2)
|
|
return {
|
|
"success": True,
|
|
"dry_run": False,
|
|
"files_uploaded": len(uploaded_files),
|
|
"files_skipped": len(skipped_files),
|
|
"bytes_transferred": total_bytes,
|
|
"duration": duration,
|
|
"uploaded_list": uploaded_files,
|
|
"remote_path": target_root,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"FTP Deploy error: {e}", exc_info=True)
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"files_uploaded": len(uploaded_files),
|
|
"files_skipped": len(skipped_files),
|
|
"bytes_transferred": total_bytes,
|
|
"duration": round(time.time() - start_time, 2),
|
|
"uploaded_list": uploaded_files,
|
|
}
|
|
|
|
return await asyncio.to_thread(_run_deploy)
|
|
|
|
|
|
ftp_manager = FTPManager()
|