415 lines
15 KiB
Python
415 lines
15 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
|
|
import subprocess
|
|
import tempfile
|
|
import shutil
|
|
|
|
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/*",
|
|
|
|
# Dependencies & local packages
|
|
"node_modules",
|
|
"node_modules/*",
|
|
|
|
# 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 = [p for p in norm_path.split("/") if p]
|
|
|
|
for pattern in excludes:
|
|
pat = pattern.replace("\\", "/").strip("/")
|
|
clean_pat = pat.rstrip("/*")
|
|
|
|
# Check against full relative path
|
|
if fnmatch.fnmatch(norm_path, pat) or fnmatch.fnmatch(norm_path, clean_pat):
|
|
return True
|
|
|
|
# If path starts with directory pattern
|
|
if norm_path == clean_pat or norm_path.startswith(clean_pat + "/"):
|
|
return True
|
|
|
|
# Check against individual directory / file parts
|
|
for part in parts:
|
|
if fnmatch.fnmatch(part, pat) or fnmatch.fnmatch(part, 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."""
|
|
clean_host = host.strip()
|
|
clean_user = user.strip() if user else ""
|
|
clean_pass = password.strip() if password else ""
|
|
|
|
if tls:
|
|
ftp = ftplib.FTP_TLS(timeout=timeout)
|
|
else:
|
|
ftp = ftplib.FTP(timeout=timeout)
|
|
|
|
ftp.encoding = "utf-8"
|
|
ftp.connect(host=clean_host, port=int(port), timeout=timeout)
|
|
|
|
if tls and isinstance(ftp, ftplib.FTP_TLS):
|
|
ftp.auth()
|
|
ftp.login(user=clean_user or "anonymous", passwd=clean_pass)
|
|
ftp.prot_p() # Secure data connection
|
|
else:
|
|
ftp.login(user=clean_user or "anonymous", passwd=clean_pass)
|
|
|
|
# Force passive mode (standard for firewalls / NAT)
|
|
ftp.set_pasv(True)
|
|
return ftp
|
|
|
|
async def test_connection(
|
|
self,
|
|
host: str,
|
|
port: int = 21,
|
|
user: str = "",
|
|
password: str = "",
|
|
path: str = "/",
|
|
tls: bool = False,
|
|
timeout: int = 12,
|
|
) -> 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():
|
|
ftp = None
|
|
try:
|
|
ftp = self._create_ftp_client(host.strip(), int(port), user.strip(), password, tls, timeout)
|
|
|
|
# Check root PWD
|
|
init_pwd = ftp.pwd()
|
|
|
|
# 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:
|
|
curr_pwd = ftp.pwd() if ftp else "/"
|
|
try:
|
|
ftp.quit()
|
|
except Exception:
|
|
pass
|
|
return False, f"اتصال به سرور برقرار شد، اما مسیر ریموت «{target_path}» یافت نشد (مسیر فعلی: {curr_pwd}): {e}"
|
|
|
|
final_pwd = ftp.pwd()
|
|
|
|
# Test directory listing
|
|
item_count = 0
|
|
try:
|
|
listing = ftp.nlst()
|
|
item_count = len(listing)
|
|
except Exception:
|
|
try:
|
|
lines = []
|
|
ftp.dir(lines.append)
|
|
item_count = len(lines)
|
|
except Exception:
|
|
item_count = 0
|
|
|
|
try:
|
|
ftp.quit()
|
|
except Exception:
|
|
pass
|
|
|
|
tls_label = " (FTPS/TLS امن)" if tls else " (FTP معمولی)"
|
|
return True, f"اتصال با موفقیت برقرار شد{tls_label} | مسیر: <code>{final_pwd}</code> | تعداد فایلها و پوشهها: {item_count}"
|
|
|
|
except socket.gaierror as e:
|
|
return False, f"نام دامنه/هاست سرور یافت نشد ({host}): {e}"
|
|
except (socket.timeout, TimeoutError) as e:
|
|
return False, f"مهلت زمانی اتصال به سرور به پایان رسید (Timeout روی پورت {port}): {e}"
|
|
except ConnectionRefusedError as e:
|
|
return False, f"اتصال توسط سرور رد شد (پورت {port} بسته است یا FTP روی آن فعال نیست): {e}"
|
|
except ftplib.error_perm as e:
|
|
return False, f"خطای نام کاربری یا رمز عبور (دسترسی نامعتبر): {e}"
|
|
except Exception as e:
|
|
return False, f"خطا در برقراری ارتباط با FTP: {e}"
|
|
finally:
|
|
if ftp:
|
|
try:
|
|
ftp.close()
|
|
except Exception:
|
|
pass
|
|
|
|
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,
|
|
branch: str = "production",
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Deploys files from the specified git branch (default: production) to the remote FTP server cleanly.
|
|
Extracts files directly from the git branch to ensure exact production code is deployed.
|
|
Skips all temporary, git, session, and junk files.
|
|
"""
|
|
import tempfile
|
|
import shutil
|
|
|
|
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
|
|
temp_export_dir = None
|
|
|
|
try:
|
|
# 1. Determine source directory: if git exists, export the exact production branch
|
|
source_dir = ws
|
|
if (ws / ".git").exists():
|
|
try:
|
|
# Check if branch exists
|
|
branches_proc = subprocess.run(
|
|
["git", "branch", "--list", branch],
|
|
cwd=str(ws),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
target_branch = branch if branch in branches_proc.stdout else "HEAD"
|
|
|
|
temp_export_dir = Path(tempfile.mkdtemp(prefix="ftp_deploy_prod_"))
|
|
# Use git archive to cleanly extract files from the target branch
|
|
archive_proc = subprocess.Popen(
|
|
["git", "archive", target_branch],
|
|
cwd=str(ws),
|
|
stdout=subprocess.PIPE,
|
|
)
|
|
tar_proc = subprocess.Popen(
|
|
["tar", "-x", "-C", str(temp_export_dir)],
|
|
stdin=archive_proc.stdout,
|
|
)
|
|
archive_proc.stdout.close()
|
|
tar_proc.communicate()
|
|
|
|
if tar_proc.returncode == 0:
|
|
source_dir = temp_export_dir
|
|
except Exception as ge:
|
|
logger.warning(f"Git archive export failed for {branch} (falling back to workspace): {ge}")
|
|
source_dir = ws
|
|
|
|
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 source directory
|
|
all_files_to_upload: List[Tuple[Path, str]] = [] # (local_path, rel_path)
|
|
|
|
for root, dirs, files in os.walk(str(source_dir)):
|
|
rel_dir = os.path.relpath(root, str(source_dir))
|
|
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,
|
|
}
|
|
finally:
|
|
if temp_export_dir and Path(temp_export_dir).exists():
|
|
try:
|
|
shutil.rmtree(temp_export_dir, ignore_errors=True)
|
|
except Exception:
|
|
pass
|
|
|
|
return await asyncio.to_thread(_run_deploy)
|
|
|
|
|
|
ftp_manager = FTPManager()
|