import os import io import logging import datetime import ftplib import stat from typing import Optional from cryptography.fernet import Fernet logger = logging.getLogger(__name__) _fernet_key = None def _get_fernet(): global _fernet_key if _fernet_key is None: key = os.environ.get("FTP_ENCRYPTION_KEY") if not key: key = Fernet.generate_key().decode("utf-8") logger.warning("FTP_ENCRYPTION_KEY not set — generated ephemeral key. FTP passwords will not survive restarts.") _fernet_key = key return Fernet(_fernet_key.encode("utf-8") if isinstance(_fernet_key, str) else _fernet_key) def encrypt_password(password: str) -> str: f = _get_fernet() return f.encrypt(password.encode("utf-8")).decode("utf-8") def decrypt_password(encrypted: str) -> str: f = _get_fernet() return f.decrypt(encrypted.encode("utf-8")).decode("utf-8") def test_ftp_connection(host: str, port: int, username: str, password: str, protocol: str = "sftp", base_path: str = "/") -> dict: try: if protocol == "sftp": return _test_sftp(host, port, username, password, base_path) elif protocol in ("ftp", "ftps"): return _test_ftp(host, port, username, password, protocol, base_path) else: return {"success": False, "error": f"Unknown protocol: {protocol}"} except Exception as e: return {"success": False, "error": str(e)} def _test_sftp(host, port, username, password, base_path): import paramiko transport = paramiko.Transport((host, port)) try: transport.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(transport) try: listing = sftp.listdir(base_path) return {"success": True, "message": f"Connected. Directory '{base_path}' contains {len(listing)} items."} finally: sftp.close() finally: transport.close() def _test_ftp(host, port, username, password, protocol, base_path): if protocol == "ftps": ftp = ftplib.FTP_TLS() else: ftp = ftplib.FTP() try: ftp.connect(host, port, timeout=15) ftp.login(username, password) if protocol == "ftps": ftp.prot_p() ftp.cwd(base_path) listing = ftp.nlst() return {"success": True, "message": f"Connected. Directory '{base_path}' contains {len(listing)} items."} finally: ftp.quit() def deploy_via_sftp(host, port, username, password, remote_path, files: dict, progress_callback=None): import paramiko transport = paramiko.Transport((host, port)) uploaded = 0 total = len(files) try: transport.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(transport) try: _sftp_makedirs(sftp, remote_path) for relative_path, content in files.items(): full_remote = os.path.join(remote_path, relative_path).replace("\\", "/") remote_dir = os.path.dirname(full_remote) _sftp_makedirs(sftp, remote_dir) if isinstance(content, bytes): sftp.putfo(io.BytesIO(content), full_remote) else: sftp.putfo(io.BytesIO(content.encode("utf-8")), full_remote) uploaded += 1 if progress_callback: progress_callback(uploaded, total, relative_path) finally: sftp.close() finally: transport.close() return uploaded def deploy_via_ftp(host, port, username, password, protocol, remote_path, files: dict, progress_callback=None): if protocol == "ftps": ftp = ftplib.FTP_TLS() else: ftp = ftplib.FTP() uploaded = 0 total = len(files) try: ftp.connect(host, port, timeout=30) ftp.login(username, password) if protocol == "ftps": ftp.prot_p() _ftp_makedirs(ftp, remote_path) for relative_path, content in files.items(): full_remote = os.path.join(remote_path, relative_path).replace("\\", "/") remote_dir = os.path.dirname(full_remote) _ftp_makedirs(ftp, remote_dir) if isinstance(content, bytes): ftp.storbinary(f"STOR {full_remote}", io.BytesIO(content)) else: ftp.storbinary(f"STOR {full_remote}", io.BytesIO(content.encode("utf-8"))) uploaded += 1 if progress_callback: progress_callback(uploaded, total, relative_path) finally: ftp.quit() return uploaded def _sftp_makedirs(sftp, remote_dir): dirs_to_create = [] current = remote_dir while current and current != "/": try: sftp.stat(current) break except FileNotFoundError: dirs_to_create.append(current) current = os.path.dirname(current) for d in reversed(dirs_to_create): try: sftp.mkdir(d) except Exception: pass def _ftp_makedirs(ftp, remote_dir): parts = remote_dir.strip("/").split("/") current = "" for part in parts: current += "/" + part try: ftp.cwd(current) except Exception: try: ftp.mkd(current) except Exception: pass def collect_deploy_files(domain: str, db_session) -> dict: from app.models import Package, Domain, Augment, BrandKit, BrandKitAsset from app.main import ( generate_standalone_site_html, generate_standalone_sales_html, render_markdown, generate_standalone_augment_html, _augment_slug, generate_standalone_admin_html, _generate_standalone_doc_html, ) from app.services.graphics import flatten_pack_assets from app.services.standalone_renderer import ( render_standalone_site, render_standalone_admin, render_standalone_doc, render_standalone_augment, _hex_to_rgb, ) dom = db_session.query(Domain).filter(Domain.domain == domain).first() if not dom: return {} pkg = db_session.query(Package).filter(Package.domain_id == dom.id).order_by(Package.id.desc()).first() if not pkg: return {} style_tier = getattr(pkg, "style_tier", None) or "premium" files = {} brand = pkg.brand or {} brand_data = brand site_copy = pkg.site_copy or {} hero_image_url = pkg.hero_image_url augments = db_session.query(Augment).filter(Augment.domain_name == domain).order_by(Augment.created_at).all() recommended_idx = brand.get("recommended", 0) options = brand.get("options", []) chosen_brand = options[recommended_idx] if options and recommended_idx < len(options) else {"name": domain, "tagline": ""} primary = brand_data.get("color_primary", "#4F46E5") secondary = brand_data.get("color_secondary", "#7C3AED") accent = brand_data.get("color_accent", "#06B6D4") brand_name = chosen_brand.get("name", domain) try: kit = db_session.query(BrandKit).filter(BrandKit.domain == domain).first() if kit: hero_assets = db_session.query(BrandKitAsset).filter( BrandKitAsset.brand_kit_id == kit.id, BrandKitAsset.classification.in_(["hero_banner", "lifestyle", "background_texture"]) ).all() if hero_assets: bk_hero_path = hero_assets[0].file_path if bk_hero_path and os.path.exists(bk_hero_path.lstrip("/")): hero_image_url = bk_hero_path except Exception as e: logger.warning(f"Brand kit hero resolution failed for {domain}: {e}") try: site_html = render_standalone_site(domain, chosen_brand, brand_data, site_copy, hero_image_url, augments=augments, tier=style_tier, template_type=pkg.template_type if pkg else None) files["index.html"] = site_html except Exception as e: logger.error(f"Failed to generate site HTML for {domain}: {e}") try: site_html = generate_standalone_site_html(domain, chosen_brand, brand_data, site_copy, hero_image_url, augments=augments) files["index.html"] = site_html except Exception as e2: logger.error(f"Nano fallback also failed for {domain}: {e2}") if pkg.sales_letter: try: sales_html = generate_standalone_sales_html(domain, pkg.chosen_niche, pkg.sales_letter, brand) files["sales.html"] = sales_html except Exception as e: logger.error(f"Failed to generate sales HTML for {domain}: {e}") for aug in augments: try: slug = _augment_slug(aug.id, aug.title) aug_html = render_standalone_augment(domain, chosen_brand, brand_data, aug, all_augments=augments, tier=style_tier) files[f"tools/{slug}.html"] = aug_html except Exception as e: logger.error(f"Failed to generate augment page for '{aug.title}': {e}") if hero_image_url and os.path.exists(hero_image_url.lstrip("/")): try: with open(hero_image_url.lstrip("/"), "rb") as f: img_filename = os.path.basename(hero_image_url) files[f"images/{img_filename}"] = f.read() except Exception as e: logger.error(f"Failed to read hero image: {e}") feature_images = pkg.feature_images or [] for img_url in feature_images: if img_url and os.path.exists(img_url.lstrip("/")): try: with open(img_url.lstrip("/"), "rb") as f: img_filename = os.path.basename(img_url) files[f"images/{img_filename}"] = f.read() except Exception as e: logger.error(f"Failed to read feature image: {e}") graphics_pack = pkg.graphics_pack or {} graphics_assets = flatten_pack_assets(graphics_pack) for asset in graphics_assets: asset_url = asset.get("url", "") if asset_url and os.path.exists(asset_url.lstrip("/")): try: with open(asset_url.lstrip("/"), "rb") as f: gp_filename = asset.get("filename", os.path.basename(asset_url)) files[f"images/graphics/{gp_filename}"] = f.read() except Exception as e: logger.error(f"Failed to read graphics pack asset: {e}") brand_kit_assets = [] try: kit = db_session.query(BrandKit).filter(BrandKit.domain == domain).first() if kit: brand_kit_assets = db_session.query(BrandKitAsset).filter(BrandKitAsset.brand_kit_id == kit.id).all() for asset in brand_kit_assets: disk_path = asset.file_path.lstrip("/") if os.path.exists(disk_path): try: with open(disk_path, "rb") as f: classification = asset.classification or "uncategorized" files[f"assets/brand-kit/{classification}/{asset.filename}"] = f.read() except Exception as e: logger.error(f"Failed to read brand kit asset '{asset.filename}': {e}") except Exception as e: logger.error(f"Failed to collect brand kit assets for {domain}: {e}") business_docs = {} all_docs_by_tier = {} business_box = pkg.business_box or {} raw_docs = business_box.get("documents", {}) if raw_docs and isinstance(raw_docs, dict): business_docs = raw_docs for dk, dv in business_docs.items(): tier = dv.get("tier", "general") all_docs_by_tier.setdefault(tier, {})[dk] = dv for dk, dv in business_docs.items(): try: doc_title = dv.get("title", dk) doc_tier = dv.get("tier", "general") doc_content = dv.get("content", "") tier_slug = doc_tier.lower().replace(" ", "-") if doc_content: rendered_md = render_markdown(doc_content) same_tier = all_docs_by_tier.get(doc_tier, {}) tier_nav_links = [(f"{k}.html", v.get("title", k)) for k, v in same_tier.items()] doc_html = render_standalone_doc( domain, brand_name, primary, secondary, accent, doc_title, doc_tier, rendered_md, tier_nav_links, tier=style_tier ) if doc_html is None: doc_html = _generate_standalone_doc_html( domain, brand_name, primary, secondary, doc_title, doc_tier, doc_content, all_docs_by_tier ) files[f"docs/{tier_slug}/{dk}.html"] = doc_html except Exception as e: logger.error(f"Failed to generate doc page for '{dk}': {e}") deployed_files_manifest = {} for fpath, content in files.items(): if isinstance(content, bytes): deployed_files_manifest[fpath] = len(content) else: deployed_files_manifest[fpath] = len(content.encode("utf-8")) if content else 0 deployed_files_manifest["admin.html"] = "pending" try: admin_html = render_standalone_admin( domain, chosen_brand, brand_data, site_copy, pkg, augments, brand_kit_assets, business_docs, graphics_assets, deployed_files_manifest, tier=style_tier ) files["admin.html"] = admin_html deployed_files_manifest["admin.html"] = len(admin_html.encode("utf-8")) except Exception as e: logger.error(f"Failed to generate admin HTML for {domain}: {e}") try: admin_html = generate_standalone_admin_html( domain, chosen_brand, brand_data, site_copy, pkg, augments, brand_kit_assets, business_docs, graphics_assets, deployed_files_manifest ) files["admin.html"] = admin_html except Exception as e2: logger.error(f"Admin fallback also failed: {e2}") return files FTP_DEPLOY_STEPS = [ {"key": "init", "label": "Preparing deployment", "description": "Loading project files and FTP profile", "est_seconds": 2}, {"key": "collect", "label": "Collecting files", "description": "Rendering site HTML and gathering assets", "est_seconds": 5}, {"key": "connect", "label": "Connecting to server", "description": "Establishing connection to remote server", "est_seconds": 3}, {"key": "upload", "label": "Uploading files", "description": "Transferring files to remote server", "est_seconds": 15}, {"key": "complete", "label": "Deployment complete", "description": "All files uploaded successfully", "est_seconds": 0}, ]